feat(sidecar): restart the vLLM sidecar when vLLM stops serving

lib/backend-common/AGENTS.md

@@ -83,6 +83,10 @@ opt-out and lets `run.rs` stay non-generic.
   reliable idle signal); SGLang overrides it by counting in-flight prefill
   streams. The mode-gate, poll loop, and SIGTERM/SIGINT ownership live in the
   `Worker`.
+- `engine_lost(&self) -> DynamoError` — optional, default never resolves.
+  Resolves once the engine can no longer serve. The `Worker` then shuts down
+  as on SIGTERM (withdraw, grace period, drain, `cleanup`) and returns this
+  error, so the process exits and its supervisor restarts it.
 - `cleanup(&self) -> Result<(), DynamoError>` — called exactly once.
   Runs after `start()` returns Ok on shutdown (even if registration /
   serve fails), **and** after `start()` raises — so implementations

lib/backend-common/src/engine.rs

@@ -280,6 +280,19 @@ pub trait LLMEngine: Send + Sync + 'static {
         Ok(None)
     }
 
+    /// Resolves once the engine can no longer serve, with the error the
+    /// `Worker` returns. The `Worker` then shuts down as on SIGTERM, withdrawing
+    /// from discovery, draining and running `cleanup`, so the process exits
+    /// and its supervisor can restart it. Default: never resolves.
+    ///
+    /// The `Worker` may poll this before `start` completes, and drops the
+    /// future once it finishes or starts shutting down another way.
+    /// Implementations must allow polling during startup and must not use this
+    /// future to clean up.
+    async fn engine_lost(&self) -> DynamoError {
+        std::future::pending().await
+    }
+
     /// Release all engine resources. Called exactly once.
     ///
     /// `Worker` guarantees:

lib/backend-common/src/worker.rs

@@ -337,6 +337,14 @@ impl EngineKind {
         }
     }
 
+    /// See [`LLMEngine::engine_lost`].
+    async fn engine_lost(&self) -> DynamoError {
+        match self {
+            EngineKind::Llm(e) => e.engine_lost().await,
+            EngineKind::Raw(_) => std::future::pending().await,
+        }
+    }
+
     async fn kv_event_sources(&self) -> Result<Vec<KvEventSource>, DynamoError> {
         match self {
             EngineKind::Llm(e) => e.kv_event_sources().await,
@@ -599,14 +607,25 @@ impl Worker {
         // grace-period sleep, which is a fixed wait rather than a hang
         // risk), otherwise we exit(911). Healthy long-running workers
         // never hit this — the timer only starts after `shutdown_token`
-        // is cancelled.
+        // is cancelled. A lost engine starts the same shutdown.
+        let engine = self.engine.clone();
         let outcome = {
             let inner_fut = self.run_inner(runtime.clone(), drt, &shutdown_token);
             tokio::pin!(inner_fut);
 
-            tokio::select! {
-                result = &mut inner_fut => result,
-                _ = shutdown_token.cancelled() => {
+            let (finished, lost) = tokio::select! {
+                biased;
+                result = &mut inner_fut => (Some(result), None),
+                _ = shutdown_token.cancelled() => (None, None),
+                error = engine.engine_lost() => (None, Some(error)),
+            };
+            match finished {
+                Some(result) => result,
+                None => {
+                    if let Some(error) = &lost {
+                        tracing::error!(%error, "Engine lost; shutting down the worker");
+                        shutdown_token.cancel();
+                    }
                     runtime.mark_shutting_down();
                     let timeout = graceful_shutdown_timeout();
                     let grace = grace_period_secs();
@@ -617,7 +636,7 @@ impl Worker {
                         timeout.as_secs(),
                         grace,
                     );
-                    match tokio::time::timeout(deadline, &mut inner_fut).await {
+                    let result = match tokio::time::timeout(deadline, &mut inner_fut).await {
                         Ok(result) => result,
                         Err(_) => {
                             tracing::error!(
@@ -627,6 +646,15 @@ impl Worker {
                             );
                             std::process::exit(911);
                         }
+                    };
+                    match lost {
+                        Some(error) => {
+                            if let Err(stopped) = result {
+                                tracing::warn!(error = %stopped, "Worker shutdown failed");
+                            }
+                            Err(error)
+                        }
+                        None => result,
                     }
                 }
             }
@@ -4353,6 +4381,109 @@ mod handoff_and_lifecycle_tests {
         worker.begin_engine_route_shutdown().await;
     }
 
+    /// A lost engine withdraws the serving worker, runs cleanup, and makes the
+    /// worker return the engine's error.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn lost_engine_withdraws_the_worker_and_returns_its_error() {
+        struct LostEngine {
+            lost: CancellationToken,
+            cleanup_calls: std::sync::atomic::AtomicUsize,
+        }
+
+        #[async_trait]
+        impl LLMEngine for LostEngine {
+            async fn start(&self, _worker_id: u64) -> Result<EngineConfig, DynamoError> {
+                Ok(EngineConfig {
+                    model: "lost-mock".to_string(),
+                    ..EngineConfig::default()
+                })
+            }
+
+            async fn generate(
+                &self,
+                _request: PreprocessedRequest,
+                _ctx: crate::engine::GenerateContext,
+            ) -> Result<
+                BoxStream<'static, Result<crate::engine::LLMEngineOutput, DynamoError>>,
+                DynamoError,
+            > {
+                unreachable!("not used in engine-loss tests")
+            }
+
+            async fn cleanup(&self) -> Result<(), DynamoError> {
+                self.cleanup_calls
+                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                Ok(())
+            }
+
+            async fn engine_lost(&self) -> DynamoError {
+                self.lost.cancelled().await;
+                DynamoError::builder()
+                    .error_type(ErrorType::Backend(BackendError::EngineShutdown))
+                    .message("synthetic engine loss")
+                    .build()
+            }
+        }
+
+        let runtime = dynamo_runtime::Runtime::from_current().unwrap();
+        let config = dynamo_runtime::distributed::DistributedConfig::process_local();
+        let drt = DistributedRuntime::new(runtime, config).await.unwrap();
+        let engine = Arc::new(LostEngine {
+            lost: CancellationToken::new(),
+            cleanup_calls: Default::default(),
+        });
+        let config = WorkerConfig::default();
+        let query = DiscoveryQuery::Endpoint {
+            namespace: config.namespace.clone(),
+            component: config.component.clone(),
+            endpoint: config.endpoint.clone(),
+        };
+        let shutdown = CancellationToken::new();
+        let run = tokio::spawn(
+            Worker::new(engine.clone(), config).run_with_drt(drt.clone(), shutdown.clone()),
+        );
+        tokio::time::timeout(Duration::from_secs(30), async {
+            while drt
+                .discovery()
+                .list(query.clone())
+                .await
+                .unwrap()
+                .is_empty()
+            {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            }
+        })
+        .await
+        .expect("worker registers its endpoint");
+
+        engine.lost.cancel();
+        let error = tokio::time::timeout(Duration::from_secs(30), run)
+            .await
+            .expect("a lost engine stops the worker")
+            .unwrap()
+            .expect_err("a lost engine fails the worker");
+        assert_eq!(
+            error.error_type(),
+            ErrorType::Backend(BackendError::EngineShutdown)
+        );
+        assert!(
+            error.to_string().contains("synthetic engine loss"),
+            "{error}"
+        );
+        assert!(
+            drt.discovery().list(query).await.unwrap().is_empty(),
+            "a lost engine must leave no endpoint"
+        );
+        assert_eq!(
+            engine
+                .cleanup_calls
+                .load(std::sync::atomic::Ordering::SeqCst),
+            1
+        );
+        assert!(shutdown.is_cancelled(), "engine loss uses the SIGTERM path");
+    }
+
     /// Ensures a payload-free Rust backend publishes readiness while it is
     /// serviceable and withdraws it again on shutdown.
     #[tokio::test]

lib/sidecar/testkit/README.md

@@ -183,8 +183,8 @@ clients. That makes peer-loss tests deterministic.
 
 | Suite | Scope | Execution |
 | --- | --- | --- |
-| `sidecar_mocker_integration.rs` | Shared streaming, errors, cancellation, cleanup, active work release, consumer drop, request/logprob fields and peer teardown for vLLM, SGLang and TensorRT-LLM; native rejection, malformed responses and shutdown during pending SGLang health checks | CPU, ordinary pre-merge Cargo tests |
-| `router_sidecar_mocker_integration.rs` | All three backends: registration/error recovery, delayed/interrupted startup, cancellation, SIGTERM, frontend HTTP errors/recovery and real PrefillRouter handoff; vLLM/SGLang model alias publication and native-health readiness; SGLang tokenizer/parser discovery, native tracing and changed-role startup | CPU, ordinary pre-merge Cargo tests |
+| `sidecar_mocker_integration.rs` | Shared streaming, errors, cancellation, cleanup, active work release, consumer drop, request/logprob fields and peer teardown for vLLM, SGLang and TensorRT-LLM; native rejection, malformed responses and shutdown during pending SGLang health checks; vLLM engine loss from its health stream | CPU, ordinary pre-merge Cargo tests |
+| `router_sidecar_mocker_integration.rs` | All three backends: registration/error recovery, delayed/interrupted startup, cancellation, SIGTERM, frontend HTTP errors/recovery and real PrefillRouter handoff; vLLM/SGLang model alias publication and native-health readiness; SGLang tokenizer/parser discovery, native tracing and changed-role startup; vLLM withdrawal and process exit on engine loss | CPU, ordinary pre-merge Cargo tests |
 | `tests/serve/test_sidecar.py` | vLLM/SGLang: real engine logprobs, structured output, aggregated and disaggregated cancellation/recovery (decode stops early), KV transfer and routing, where the `*_kv_router` configs warm two caches and require cache-based selection, plus a completed transfer to decode when disaggregated; TensorRT-LLM: aggregated/disaggregated chat and cancellation/recovery | GPU, existing sidecar E2E jobs in post-merge and nightly |
 
 A generic scenario is reusable code, not evidence that every backend runs it.

lib/sidecar/testkit/tests/router_sidecar_mocker_integration.rs

@@ -724,6 +724,51 @@ async fn load_loss_withdraws_worker_and_fails_the_process() {
     peer.shutdown().await;
 }
 
+#[tokio::test]
+async fn vllm_engine_loss_withdraws_worker_and_fails_the_process() {
+    tokio::time::timeout(
+        Duration::from_secs(60),
+        engine_loss_withdraws_worker_and_fails_the_process(),
+    )
+    .await
+    .expect("process scenario exceeded its overall deadline");
+}
+
+/// A vLLM engine lost after startup withdraws the worker, and the sidecar then
+/// exits with an error.
+async fn engine_loss_withdraws_worker_and_fails_the_process() {
+    let env = Environment::new().await;
+    let mut peer = vllm::Fixture::start(
+        Controller::default(),
+        FixtureConfig {
+            model: env.model.clone(),
+            ..Default::default()
+        },
+    )
+    .await;
+    // The grace keeps the withdrawn process up while the test observes it.
+    let mut child =
+        env.spawn_with_grace::<vllm::Fixture>(&peer.endpoint(), DisaggregationMode::Aggregated, 5);
+    let router = env.ready("backend").await;
+    peer.shutdown().await;
+    tokio::time::timeout(Duration::from_secs(20), async {
+        while !env.registrations("backend").await.is_empty() {
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .unwrap_or_else(|_| panic!("engine loss did not withdraw the worker\n{}", child.logs()));
+    env.withdrawn("backend", &router).await;
+    assert!(child.is_running(), "withdrawal must precede process exit");
+    // 1 is the engine-loss error; a forced shutdown (exit 911) would read 143.
+    assert_eq!(child.exit().await.code(), Some(1), "{}", child.logs());
+    assert!(
+        child.logs().contains("vLLM engine lost"),
+        "{}",
+        child.logs()
+    );
+}
+
 async fn ready(router: &PrefillRouter) {
     bounded("real PrefillRouter availability", async {
         loop {

lib/sidecar/testkit/tests/sidecar_mocker_integration.rs

@@ -3,8 +3,8 @@
 
 use dynamo_backend_common::testing::mock_context;
 use dynamo_backend_common::{
-    BackendError, DisaggregationMode, DynamoError, FinishReason, GenerateContext, LLMEngine,
-    LLMEngineOutput, PreprocessedRequest,
+    BackendError, DisaggregationMode, DynamoError, ErrorType, FinishReason, GenerateContext,
+    LLMEngine, LLMEngineOutput, PreprocessedRequest,
 };
 use dynamo_llm::protocols::common::preprocessor::PrefillResult;
 use dynamo_sidecar_testkit::assert::{failure, terminal};
@@ -760,6 +760,29 @@ async fn vllm_malformed_terminal_fails_then_recovers() {
     .await;
 }
 
+/// vLLM reporting NOT_SERVING after startup resolves `engine_lost`; the
+/// process test covers a dropped connection.
+#[tokio::test]
+async fn vllm_engine_lost_on_not_serving() {
+    let mut fixture =
+        vllm_fixture::Fixture::start(Controller::default(), FixtureConfig::default()).await;
+    let engine = fixture.engine().await;
+    engine.start(0).await.unwrap();
+    let lost = engine.engine_lost();
+    tokio::pin!(lost);
+    tokio::time::timeout(std::time::Duration::from_millis(300), &mut lost)
+        .await
+        .expect_err("a SERVING engine is not lost");
+    fixture.set_overall_serving(false).await;
+    let error = bounded("engine lost on NOT_SERVING", lost).await;
+    assert_eq!(
+        error.error_type(),
+        ErrorType::Backend(BackendError::EngineShutdown)
+    );
+    assert!(error.to_string().contains("vLLM engine lost"), "{error}");
+    finish(&mut fixture, &engine).await;
+}
+
 #[tokio::test]
 async fn sglang_malformed_terminal_fails_then_recovers() {
     bounded("malformed terminal and recovery", async {

lib/sidecar/testkit/tests/support/vllm.rs

@@ -46,9 +46,21 @@ pub struct Fixture {
     reasoning_parser: Arc<Mutex<String>>,
     readiness: watch::Sender<Option<bool>>,
     health_received: watch::Receiver<bool>,
+    health: tonic_health::server::HealthReporter,
 }
 
 impl Fixture {
+    /// Sets what `Watch` reports for overall health (`""`); `Check` follows
+    /// `set_health` instead.
+    pub async fn set_overall_serving(&self, is_serving: bool) {
+        let status = if is_serving {
+            tonic_health::ServingStatus::Serving
+        } else {
+            tonic_health::ServingStatus::NotServing
+        };
+        self.health.set_service_status("", status).await;
+    }
+
     pub fn set_reasoning_parser(&self, parser: &str) {
         *self.reasoning_parser.lock().unwrap() = parser.to_owned();
     }
@@ -107,7 +119,7 @@ impl SidecarFixture for Fixture {
         let (readiness, is_healthy) = watch::channel(Some(true));
         let (health_requested, health_received) = watch::channel(false);
         let health_service = HealthServer::new(ControlledHealth {
-            inner: tonic_health::server::HealthService::from_health_reporter(health),
+            inner: tonic_health::server::HealthService::from_health_reporter(health.clone()),
             is_healthy,
             health_requested,
         });
@@ -137,6 +149,7 @@ impl SidecarFixture for Fixture {
             reasoning_parser,
             readiness,
             health_received,
+            health,
         }
     }
 

lib/sidecar/vllm/src/client.rs

@@ -55,6 +55,10 @@ impl VllmClient {
         self.pool.len()
     }
 
+    pub(crate) fn health_channel(&self) -> Channel {
+        self.pool.next_channel()
+    }
+
     pub(crate) fn control_client(&self) -> pb::control_client::ControlClient<Channel> {
         pb::control_client::ControlClient::new(self.pool.next_channel())
             .max_encoding_message_size(DEFAULT_MAX_GRPC_MESSAGE_SIZE)
@@ -251,6 +255,34 @@ pub(crate) fn protocol_error(message: impl Into<String>) -> DynamoError {
     dynamo_sidecar_common::protocol_error("vLLM", message)
 }
 
+/// Follows vLLM's overall gRPC health (`""`) until vLLM stops serving, and
+/// returns why. vLLM changes it only on engine failure or shutdown, never while
+/// paused or asleep.
+pub(crate) async fn watch_health(channel: Channel, open_timeout: Duration) -> String {
+    let request = HealthCheckRequest {
+        service: String::new(),
+    };
+    let mut stream =
+        match tokio::time::timeout(open_timeout, HealthClient::new(channel).watch(request)).await {
+            Ok(Ok(response)) => response.into_inner(),
+            Ok(Err(status)) => return format!("health watch failed: {status}"),
+            Err(_) => return format!("health watch did not open in {open_timeout:?}"),
+        };
+    loop {
+        match stream.message().await {
+            Ok(Some(response)) => {
+                let status =
+                    ServingStatus::try_from(response.status).unwrap_or(ServingStatus::Unknown);
+                if status != ServingStatus::Serving {
+                    return format!("health reported {}", status.as_str_name());
+                }
+            }
+            Ok(None) => return "health stream ended".to_string(),
+            Err(status) => return format!("health stream failed: {status}"),
+        }
+    }
+}
+
 // Bound how long a stalled RPC can hold the lifecycle lock.
 pub(crate) const LORA_RPC_DEADLINE: Duration = Duration::from_secs(60);