From ead2dcc9f76861c8be313c51704b282d0b6ae934 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Tue, 4 Aug 2026 15:06:53 -0700 Subject: [PATCH 1/3] fix(server): handle graceful process shutdown Signed-off-by: nachiketb --- crates/switchyard-py/src/server_bindings.rs | 1 + crates/switchyard-server/README.md | 3 + crates/switchyard-server/src/cli.rs | 9 +- crates/switchyard-server/src/lib.rs | 184 ++++++++++++++++++-- 4 files changed, 182 insertions(+), 15 deletions(-) diff --git a/crates/switchyard-py/src/server_bindings.rs b/crates/switchyard-py/src/server_bindings.rs index 77f7d5973..6ea78c005 100644 --- a/crates/switchyard-py/src/server_bindings.rs +++ b/crates/switchyard-py/src/server_bindings.rs @@ -46,6 +46,7 @@ impl PyServer { addr: SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), port), backlog: DEFAULT_LISTEN_BACKLOG, dry_run: false, + shutdown_timeout: Duration::from_secs(2), tls: None, }, ) diff --git a/crates/switchyard-server/README.md b/crates/switchyard-server/README.md index fe2bf2ac3..e21da15e8 100644 --- a/crates/switchyard-server/README.md +++ b/crates/switchyard-server/README.md @@ -58,6 +58,9 @@ export API_KEY="..." cargo run -p switchyard-server -- --config routes.toml ``` +Ctrl+C and Unix `SIGTERM` stop new connections and allow active requests to drain for up to +`--shutdown-timeout` (30 seconds by default) before they are terminated. + The server logs exactly one structured terminal event per LLM request: successful responses at `INFO`, 4xx responses at `WARN`, and 5xx responses at `ERROR`. Set `RUST_LOG=switchyard_server=debug,libsy=debug` to include routing decisions and nested failure diff --git a/crates/switchyard-server/src/cli.rs b/crates/switchyard-server/src/cli.rs index 314be9066..231d4e2bc 100644 --- a/crates/switchyard-server/src/cli.rs +++ b/crates/switchyard-server/src/cli.rs @@ -9,8 +9,8 @@ use std::path::PathBuf; use clap::Parser; use switchyard_server::config::load_server_state; use switchyard_server::{ - DEFAULT_LISTEN_BACKLOG, ServerError, ServerResult, ServerRunOptions, ServerState, TlsOptions, - run_server, + DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT, DEFAULT_LISTEN_BACKLOG, ServerError, ServerResult, + ServerRunOptions, ServerState, TlsOptions, run_server, }; const DEFAULT_HOST: IpAddr = IpAddr::V4(Ipv4Addr::UNSPECIFIED); @@ -40,6 +40,10 @@ pub(crate) struct ServerArgs { #[arg(long, default_value_t = DEFAULT_LISTEN_BACKLOG)] backlog: u32, + /// Maximum time active requests may drain during shutdown. + #[arg(long, default_value_t = humantime::Duration::from(DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT))] + shutdown_timeout: humantime::Duration, + /// Validate the algorithm and client configuration without binding a socket. #[arg(long)] dry_run: bool, @@ -85,6 +89,7 @@ impl ServerArgs { addr: SocketAddr::new(self.host, self.port), backlog: self.backlog, dry_run: self.dry_run, + shutdown_timeout: self.shutdown_timeout.into(), tls, }; Ok((state, options)) diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 9196f0844..24b58cea0 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -50,6 +50,9 @@ pub use observability::{flush_observability, initialize_observability}; /// Default TCP listen backlog used by the Rust server. pub const DEFAULT_LISTEN_BACKLOG: u32 = 65_535; +/// Default time allowed for active requests to finish during shutdown. +pub const DEFAULT_GRACEFUL_SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(30); + /// Maximum buffered JSON request size accepted by the LLM endpoints. pub const DEFAULT_MAX_REQUEST_BODY_BYTES: usize = 32 * 1024 * 1024; @@ -214,6 +217,8 @@ pub struct ServerRunOptions { pub backlog: u32, /// Validate runtime construction without binding a socket. pub dry_run: bool, + /// Maximum time active requests may drain after shutdown begins. + pub shutdown_timeout: Duration, /// TLS certificate configuration, when HTTPS is enabled. pub tls: Option, } @@ -276,10 +281,11 @@ impl BoundServer { self, shutdown: impl Future + Send + 'static, ) -> ServerResult<()> { + let shutdown_timeout = self.options.shutdown_timeout; if let Some(tls) = self.options.tls { - serve_tls(self.listener, self.router, tls, shutdown).await + serve_tls(self.listener, self.router, tls, shutdown_timeout, shutdown).await } else { - serve(self.listener, self.router, shutdown).await + serve(self.listener, self.router, shutdown_timeout, shutdown).await } } @@ -292,6 +298,7 @@ async fn serve_tls( listener: TcpListener, router: Router, tls: TlsOptions, + shutdown_timeout: Duration, shutdown: impl Future + Send + 'static, ) -> ServerResult<()> { if let Err(error) = rustls::crypto::aws_lc_rs::default_provider().install_default() { @@ -302,31 +309,51 @@ async fn serve_tls( .await .map_err(server_io_error)?; let handle = axum_server::Handle::new(); - - let shutdown_handle = handle.clone(); - tokio::spawn(async move { - shutdown.await; - shutdown_handle.graceful_shutdown(Some(Duration::from_secs(2))); - }); + let shutdown_task = schedule_shutdown(handle.clone(), shutdown_timeout, shutdown); let std_listener = listener.into_std().map_err(server_io_error)?; - axum_server::from_tcp_rustls(std_listener, config) + let result = axum_server::from_tcp_rustls(std_listener, config) .map_err(server_io_error)? .handle(handle) .serve(router.into_make_service()) .await - .map_err(server_io_error) + .map_err(server_io_error); + shutdown_task.abort(); + result } async fn serve( listener: TcpListener, router: Router, + shutdown_timeout: Duration, shutdown: impl Future + Send + 'static, ) -> ServerResult<()> { - axum::serve(listener, router) - .with_graceful_shutdown(shutdown) + let handle = axum_server::Handle::new(); + let shutdown_task = schedule_shutdown(handle.clone(), shutdown_timeout, shutdown); + let std_listener = listener.into_std().map_err(server_io_error)?; + let result = axum_server::from_tcp(std_listener) + .map_err(server_io_error)? + .handle(handle) + .serve(router.into_make_service()) .await - .map_err(server_io_error) + .map_err(server_io_error); + shutdown_task.abort(); + result +} + +fn schedule_shutdown( + handle: axum_server::Handle, + timeout: Duration, + shutdown: impl Future + Send + 'static, +) -> task::JoinHandle<()> { + tokio::spawn(async move { + shutdown.await; + tracing::info!( + ?timeout, + "shutdown signal received; draining active requests" + ); + handle.graceful_shutdown(Some(timeout)); + }) } /// Ingress timestamp for one request, taken before any body is read. @@ -412,6 +439,17 @@ fn server_io_error(error: std::io::Error) -> ServerError { } async fn shutdown_signal() { + #[cfg(unix)] + tokio::select! { + _ = ctrl_c_signal() => {}, + _ = terminate_signal() => {}, + } + + #[cfg(not(unix))] + ctrl_c_signal().await; +} + +async fn ctrl_c_signal() { if let Err(error) = tokio::signal::ctrl_c().await { tracing::warn!( error = %error, @@ -421,6 +459,23 @@ async fn shutdown_signal() { } } +#[cfg(unix)] +async fn terminate_signal() { + let mut signal = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + { + Ok(signal) => signal, + Err(error) => { + tracing::warn!( + error = %error, + "SIGTERM shutdown signal unavailable; continuing without SIGTERM trigger" + ); + std::future::pending::<()>().await; + return; + } + }; + signal.recv().await; +} + async fn openai_chat_completions( State(state): State, Extension(started): Extension, @@ -1079,8 +1134,111 @@ fn endpoint_listing(has_routing_log: bool) -> String { #[cfg(test)] mod tests { + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::sync::{Notify, oneshot}; + use super::*; + #[derive(Clone)] + struct ShutdownTestState { + started: Arc, + release: Arc, + } + + struct ShutdownTestServer { + state: ShutdownTestState, + shutdown: oneshot::Sender<()>, + server: task::JoinHandle>, + request: task::JoinHandle>>, + } + + async fn blocked_request(State(state): State) -> &'static str { + state.started.notify_one(); + state.release.notified().await; + "done" + } + + async fn raw_request(addr: SocketAddr) -> std::io::Result> { + let mut stream = tokio::net::TcpStream::connect(addr).await?; + stream + .write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n") + .await?; + let mut response = Vec::new(); + stream.read_to_end(&mut response).await?; + Ok(response) + } + + fn shutdown_test_server(shutdown_timeout: Duration) -> ShutdownTestServer { + let state = ShutdownTestState { + started: Arc::new(Notify::new()), + release: Arc::new(Notify::new()), + }; + let router = Router::new() + .route("/", get(blocked_request)) + .with_state(state.clone()); + let listener = bind_tcp_listener("127.0.0.1:0".parse().expect("valid address"), 16) + .expect("listener binds"); + let addr = listener.local_addr().expect("listener has an address"); + let (shutdown, shutdown_receiver) = oneshot::channel(); + let server = tokio::spawn(serve(listener, router, shutdown_timeout, async move { + let _ = shutdown_receiver.await; + })); + let request = tokio::spawn(raw_request(addr)); + ShutdownTestServer { + state, + shutdown, + server, + request, + } + } + + // Active requests may finish within the grace period, while stuck requests are bounded. + #[tokio::test] + async fn shutdown_drains_until_configured_deadline() { + let ShutdownTestServer { + state, + shutdown, + mut server, + request, + } = shutdown_test_server(Duration::from_secs(1)); + state.started.notified().await; + shutdown.send(()).expect("server receives shutdown"); + assert!( + tokio::time::timeout(Duration::from_millis(25), &mut server) + .await + .is_err(), + "server must wait for the active request" + ); + state.release.notify_one(); + tokio::time::timeout(Duration::from_secs(1), server) + .await + .expect("server stops after request drains") + .expect("server task completes") + .expect("server exits cleanly"); + let response = request + .await + .expect("request task completes") + .expect("request succeeds"); + assert!(response.windows(8).any(|part| part == b"200 OK\r\n")); + assert!(response.ends_with(b"done")); + + let ShutdownTestServer { + state, + shutdown, + server, + request, + } = shutdown_test_server(Duration::from_millis(25)); + state.started.notified().await; + shutdown.send(()).expect("server receives shutdown"); + tokio::time::timeout(Duration::from_secs(1), server) + .await + .expect("shutdown deadline is enforced") + .expect("server task completes") + .expect("server exits cleanly"); + state.release.notify_one(); + request.abort(); + } + // Terminal request severity follows HTTP status instead of error-path bookkeeping. #[test] fn request_log_level_follows_http_status() { From 7151c7d326b44e8c07a80ae42595b1bd3f64e423 Mon Sep 17 00:00:00 2001 From: nachiketb Date: Tue, 4 Aug 2026 15:13:24 -0700 Subject: [PATCH 2/3] refactor(server): isolate platform shutdown signals Signed-off-by: nachiketb --- crates/switchyard-server/src/lib.rs | 41 +------------------ crates/switchyard-server/src/shutdown.rs | 52 ++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 39 deletions(-) create mode 100644 crates/switchyard-server/src/shutdown.rs diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 24b58cea0..8de709a2b 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -8,6 +8,7 @@ mod metrics; mod observability; mod response; mod routing_log; +mod shutdown; mod sse; mod stats; mod usage_metrics; @@ -247,7 +248,7 @@ pub async fn run_server(state: ServerState, options: ServerRunOptions) -> Server let server = BoundServer::bind(state, options)?; println!("{}", server.startup_banner(std::io::stdout().is_terminal())); - server.serve(shutdown_signal()).await + server.serve(shutdown::signal()).await } /// A configured server with its listening socket already bound. @@ -438,44 +439,6 @@ fn server_io_error(error: std::io::Error) -> ServerError { ServerError::new(error.to_string()) } -async fn shutdown_signal() { - #[cfg(unix)] - tokio::select! { - _ = ctrl_c_signal() => {}, - _ = terminate_signal() => {}, - } - - #[cfg(not(unix))] - ctrl_c_signal().await; -} - -async fn ctrl_c_signal() { - if let Err(error) = tokio::signal::ctrl_c().await { - tracing::warn!( - error = %error, - "ctrl-c shutdown signal unavailable; continuing without shutdown trigger" - ); - std::future::pending::<()>().await; - } -} - -#[cfg(unix)] -async fn terminate_signal() { - let mut signal = match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) - { - Ok(signal) => signal, - Err(error) => { - tracing::warn!( - error = %error, - "SIGTERM shutdown signal unavailable; continuing without SIGTERM trigger" - ); - std::future::pending::<()>().await; - return; - } - }; - signal.recv().await; -} - async fn openai_chat_completions( State(state): State, Extension(started): Extension, diff --git a/crates/switchyard-server/src/shutdown.rs b/crates/switchyard-server/src/shutdown.rs new file mode 100644 index 000000000..47ae0a6ad --- /dev/null +++ b/crates/switchyard-server/src/shutdown.rs @@ -0,0 +1,52 @@ +// SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved. +// SPDX-License-Identifier: Apache-2.0 + +//! Platform-specific process shutdown signals. + +/// Waits for the platform's normal process termination signal. +pub(crate) async fn signal() { + platform::signal().await; +} + +async fn ctrl_c() { + if let Err(error) = tokio::signal::ctrl_c().await { + tracing::warn!( + error = %error, + "ctrl-c shutdown signal unavailable; continuing without shutdown trigger" + ); + std::future::pending::<()>().await; + } +} + +#[cfg(unix)] +mod platform { + pub(super) async fn signal() { + tokio::select! { + _ = super::ctrl_c() => {}, + _ = terminate() => {}, + } + } + + async fn terminate() { + let mut signal = + match tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) { + Ok(signal) => signal, + Err(error) => { + tracing::warn!( + error = %error, + "SIGTERM shutdown signal unavailable; continuing without SIGTERM trigger" + ); + std::future::pending::<()>().await; + return; + } + }; + signal.recv().await; + } +} + +#[cfg(not(unix))] +mod platform { + pub(super) async fn signal() { + super::ctrl_c().await; + } +} From 7010281dda6b071bac510d6d0be13c1e330f301c Mon Sep 17 00:00:00 2001 From: nachiketb Date: Tue, 4 Aug 2026 15:34:59 -0700 Subject: [PATCH 3/3] refactor(server): await shutdown without spawning Signed-off-by: nachiketb --- crates/switchyard-server/src/lib.rs | 59 ++++++++++++++--------------- 1 file changed, 28 insertions(+), 31 deletions(-) diff --git a/crates/switchyard-server/src/lib.rs b/crates/switchyard-server/src/lib.rs index 8de709a2b..13374da5f 100644 --- a/crates/switchyard-server/src/lib.rs +++ b/crates/switchyard-server/src/lib.rs @@ -309,18 +309,13 @@ async fn serve_tls( let config = RustlsConfig::from_pem_file(tls.cert, tls.key) .await .map_err(server_io_error)?; - let handle = axum_server::Handle::new(); - let shutdown_task = schedule_shutdown(handle.clone(), shutdown_timeout, shutdown); - let std_listener = listener.into_std().map_err(server_io_error)?; - let result = axum_server::from_tcp_rustls(std_listener, config) - .map_err(server_io_error)? - .handle(handle) - .serve(router.into_make_service()) - .await - .map_err(server_io_error); - shutdown_task.abort(); - result + let server = axum_server::from_tcp_rustls(std_listener, config).map_err(server_io_error)?; + let handle = axum_server::Handle::new(); + let server = server + .handle(handle.clone()) + .serve(router.into_make_service()); + serve_until_shutdown(server, handle, shutdown_timeout, shutdown).await } async fn serve( @@ -329,32 +324,34 @@ async fn serve( shutdown_timeout: Duration, shutdown: impl Future + Send + 'static, ) -> ServerResult<()> { - let handle = axum_server::Handle::new(); - let shutdown_task = schedule_shutdown(handle.clone(), shutdown_timeout, shutdown); let std_listener = listener.into_std().map_err(server_io_error)?; - let result = axum_server::from_tcp(std_listener) - .map_err(server_io_error)? - .handle(handle) - .serve(router.into_make_service()) - .await - .map_err(server_io_error); - shutdown_task.abort(); - result + let server = axum_server::from_tcp(std_listener).map_err(server_io_error)?; + let handle = axum_server::Handle::new(); + let server = server + .handle(handle.clone()) + .serve(router.into_make_service()); + serve_until_shutdown(server, handle, shutdown_timeout, shutdown).await } -fn schedule_shutdown( +/// Runs the server until it exits or shutdown begins, then drains active requests. +async fn serve_until_shutdown( + server: impl Future>, handle: axum_server::Handle, timeout: Duration, shutdown: impl Future + Send + 'static, -) -> task::JoinHandle<()> { - tokio::spawn(async move { - shutdown.await; - tracing::info!( - ?timeout, - "shutdown signal received; draining active requests" - ); - handle.graceful_shutdown(Some(timeout)); - }) +) -> ServerResult<()> { + tokio::pin!(server); + tokio::select! { + result = &mut server => result.map_err(server_io_error), + _ = shutdown => { + tracing::info!( + ?timeout, + "shutdown signal received; draining active requests" + ); + handle.graceful_shutdown(Some(timeout)); + server.await.map_err(server_io_error) + } + } } /// Ingress timestamp for one request, taken before any body is read.