Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .github/codeowners/areas.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ areas:
- lib/llm/src/grpc/protos/
- lib/llm/src/grpc/service.rs
- lib/llm/src/grpc/service/
- lib/http-server/
- lib/llm/src/http.rs
- lib/llm/src/http/
- lib/llm/src/http/service.rs
Expand Down Expand Up @@ -1074,6 +1075,10 @@ shared:
owners:
- frontend
- observability
- glob: lib/http-server/src/metrics.rs
owners:
- frontend
- observability
- glob: lib/llm/src/kv_router/shared_cache.rs
owners:
- router
Expand Down
2 changes: 2 additions & 0 deletions CODEOWNERS
Original file line number Diff line number Diff line change
Expand Up @@ -65,6 +65,7 @@
* @ai-dynamo/dynamo-runtime-codeowners

# === frontend (@ai-dynamo/dynamo-frontend-codeowners) ===
/lib/http-server/ @ai-dynamo/dynamo-frontend-codeowners
/examples/chat_templates/ @ai-dynamo/dynamo-frontend-codeowners
/components/src/dynamo/frontend/ @ai-dynamo/dynamo-frontend-codeowners
/components/src/dynamo/frontend/tests/ @ai-dynamo/dynamo-frontend-codeowners
Expand Down Expand Up @@ -643,6 +644,7 @@ requirements.txt
/examples/diffusers/Dockerfile @ai-dynamo/dynamo-diffusion-codeowners @ai-dynamo/dynamo-ops-codeowners
/lib/kvbm-logical/src/metrics/ @ai-dynamo/dynamo-kv-memory-codeowners @ai-dynamo/dynamo-observability-codeowners
/tests/serve/test_vllm_omni.py @ai-dynamo/dynamo-runtime-codeowners @ai-dynamo/dynamo-backend-vllm-codeowners
/lib/http-server/src/metrics.rs @ai-dynamo/dynamo-frontend-codeowners @ai-dynamo/dynamo-observability-codeowners
/tests/test_kustomize_matrix.py @ai-dynamo/dynamo-runtime-codeowners @ai-dynamo/dynamo-ops-codeowners
/.github/workflows/fern-docs.yml @ai-dynamo/dynamo-ops-codeowners @ai-dynamo/dynamo-docs-codeowners
/examples/deployments/**/vllm/** @ai-dynamo/dynamo-operator-codeowners @ai-dynamo/dynamo-backend-vllm-codeowners
Expand Down
25 changes: 24 additions & 1 deletion Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

2 changes: 2 additions & 0 deletions Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

[workspace]
members = [
"lib/http-server",
"lib/llm",
"lib/rl",
"lib/runtime",
Expand Down Expand Up @@ -55,6 +56,7 @@ dynamo-custom-policy-builtin = { path = "lib/router-plugins/builtin", version =
aisimulate-core = { version = "=0.13.0-dev.202610160000000073" }

# Local crates
dynamo-http-server = { path = "lib/http-server", version = "1.6.0" }
dynamo-runtime = { path = "lib/runtime", version = "1.6.0" }
dynamo-llm = { path = "lib/llm", version = "1.6.0", default-features = false }
dynamo-rl = { path = "lib/rl", version = "1.6.0" }
Expand Down
1 change: 1 addition & 0 deletions deploy/inference-gateway/ext-proc/Cargo.toml
Original file line number Diff line number Diff line change
Expand Up @@ -16,6 +16,7 @@ name = "dynamo-ext-proc"
path = "src/bin/dynamo-ext-proc.rs"

[dependencies]
dynamo-http-server = { workspace = true }
dynamo-custom-policy-builtin = { workspace = true }
async-trait = { workspace = true }
dynamo-kv-router = { workspace = true, features = [
Expand Down
2 changes: 1 addition & 1 deletion deploy/inference-gateway/ext-proc/src/metrics.rs
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@ use axum::{
Router as AxumRouter, http::StatusCode, http::header::CONTENT_TYPE, response::IntoResponse,
routing::get,
};
use dynamo_llm::http::service::metrics::generate_log_buckets;
use dynamo_http_server::metrics::generate_log_buckets;
use prometheus::{Encoder, HistogramOpts, HistogramVec, Registry, TEXT_FORMAT, TextEncoder};

/// Port the `/metrics` endpoint binds to unless `DYN_EPP_METRICS_PORT` says
Expand Down
23 changes: 21 additions & 2 deletions lib/bindings/kvbm/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

23 changes: 21 additions & 2 deletions lib/bindings/python/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

33 changes: 33 additions & 0 deletions lib/http-server/Cargo.toml
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
# SPDX-FileCopyrightText: Copyright (c) 2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
# SPDX-License-Identifier: Apache-2.0

[package]
name = "dynamo-http-server"
version.workspace = true
edition.workspace = true
authors.workspace = true
license.workspace = true
homepage.workspace = true
repository.workspace = true
description = "Plane-independent production HTTP server for Dynamo services"

[dependencies]
anyhow = { workspace = true }
axum = { workspace = true }
hyper-util = { version = "0.1.17", features = ["server-auto", "service", "tokio"] }
tokio-rustls = { workspace = true }
tower = { workspace = true }
parking_lot = { workspace = true }
rand = { workspace = true }
serde = { workspace = true }
dynamo-runtime = { workspace = true }
futures = { workspace = true }
prometheus = { workspace = true }
tokio = { workspace = true }
tokio-util = { workspace = true }
tracing = { workspace = true }

[dev-dependencies]
rcgen = { workspace = true }
rustls = { workspace = true }
tokio = { workspace = true, features = ["test-util"] }
138 changes: 138 additions & 0 deletions lib/http-server/src/disconnect.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,138 @@
// SPDX-FileCopyrightText: Copyright (c) 2024-2026 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
// SPDX-License-Identifier: Apache-2.0

//! Connection monitoring for HTTP and gRPC services.

use crate::metrics::{CancellationLabels, Metrics};
use dynamo_runtime::engine::AsyncEngineContext;
use std::sync::Arc;

#[derive(Clone, Copy)]
pub enum ConnectionStatus {
Disabled,
ClosedUnexpectedly,
ClosedGracefully,
}

pub struct ConnectionHandle {
sender: Option<tokio::sync::oneshot::Sender<ConnectionStatus>>,
on_drop: ConnectionStatus,
}

impl ConnectionHandle {
/// Handle which by default will issue a [`ConnectionStatus::ClosedGracefully`] signal when dropped.
pub fn create_disarmed(sender: tokio::sync::oneshot::Sender<ConnectionStatus>) -> Self {
Self {
sender: Some(sender),
on_drop: ConnectionStatus::ClosedGracefully,
}
}

/// Handle which will issue a [`ConnectionStatus::ClosedUnexpectedly`] signal when dropped.
pub fn create_armed(sender: tokio::sync::oneshot::Sender<ConnectionStatus>) -> Self {
Self {
sender: Some(sender),
on_drop: ConnectionStatus::ClosedUnexpectedly,
}
}

/// Handle which will not issue a signal when dropped.
pub fn create_disabled(sender: tokio::sync::oneshot::Sender<ConnectionStatus>) -> Self {
Self {
sender: Some(sender),
on_drop: ConnectionStatus::Disabled,
}
}

/// Handle which will issue a [`ConnectionStatus::ClosedGracefully`] signal when dropped.
pub fn disarm(&mut self) {
self.on_drop = ConnectionStatus::ClosedGracefully;
}

/// Handle which will issue a [`ConnectionStatus::ClosedUnexpectedly`] signal when dropped.
pub fn arm(&mut self) {
self.on_drop = ConnectionStatus::ClosedUnexpectedly;
}
}

impl Drop for ConnectionHandle {
fn drop(&mut self) {
if let Some(sender) = self.sender.take() {
let _ = sender.send(self.on_drop);
}
}
}

/// Creates a pair of handles which will monitor for disconnects from the client.
///
/// The first handle is armed and will issue a [`ConnectionStatus::ClosedUnexpectedly`] signal when dropped.
/// The second handle is disarmed and will issue a [`ConnectionStatus::ClosedGracefully`] signal when dropped.
///
/// The handles are returned in the order of the first being armed and the second being disarmed.
pub async fn create_connection_monitor(
engine_context: Arc<dyn AsyncEngineContext>,
metrics: Option<Arc<Metrics>>,
cancellation_labels: CancellationLabels,
) -> (ConnectionHandle, ConnectionHandle) {
// these oneshot channels monitor possible disconnects from the client in two different scopes:
// - the local task (connection_handle)
// - an optionally streaming response (stream_handle)
let (connection_tx, connection_rx) = tokio::sync::oneshot::channel();
let (stream_tx, stream_rx) = tokio::sync::oneshot::channel();

// detached task that will naturally close when both handles are dropped
tokio::spawn(connection_monitor(
engine_context.clone(),
connection_rx,
stream_rx,
metrics,
cancellation_labels,
));

// Two handles, the first is armed, the second is disarmed
(
ConnectionHandle::create_armed(connection_tx),
ConnectionHandle::create_disabled(stream_tx),
)
}

/// Monitor handler and response-body completion and cancel on unexpected drops.
#[tracing::instrument(level = "trace", skip_all, fields(request_id = %engine_context.id()))]
pub async fn connection_monitor(
engine_context: Arc<dyn AsyncEngineContext>,
connection_rx: tokio::sync::oneshot::Receiver<ConnectionStatus>,
stream_rx: tokio::sync::oneshot::Receiver<ConnectionStatus>,
metrics: Option<Arc<Metrics>>,
cancellation_labels: CancellationLabels,
) {
match connection_rx.await {
Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => {
// the client has disconnected, no need to gracefully cancel, just kill the context
tracing::warn!("Connection closed unexpectedly; issuing cancellation");
if let Some(metrics) = &metrics {
metrics.inc_client_disconnect();
metrics.inc_cancellation(&cancellation_labels);
}
engine_context.kill();
}
Ok(ConnectionStatus::ClosedGracefully) => {
tracing::trace!("Connection closed gracefully");
}
Ok(ConnectionStatus::Disabled) => {}
}

match stream_rx.await {
Err(_) | Ok(ConnectionStatus::ClosedUnexpectedly) => {
tracing::warn!("Stream closed unexpectedly; issuing cancellation");
if let Some(metrics) = &metrics {
metrics.inc_client_disconnect();
metrics.inc_cancellation(&cancellation_labels);
}
engine_context.kill();
}
Ok(ConnectionStatus::ClosedGracefully) => {
tracing::trace!("Stream closed gracefully");
}
Ok(ConnectionStatus::Disabled) => {}
}
}
Loading
Loading