methanium/mtp
Watch
3
0
Fork
You've already forked mtp
 0
Code Issues Pull requests 3 Projects Releases 22 Packages Wiki Activity Actions

[Add] TCP server to core MTP (HTTP/1.1 & HTTP/2) compatibility
All checks were successful
CI / checks (push) Successful in 5m29s
Details

Browse source
This commit is contained in:
Alex Emmet 2026-07-21 00:43:00 +02:00
commit 00f0aaeeff
21 changed files with 1627 additions and 677 deletions
Download patch file Download diff file Expand all files Collapse all files

549
mtp-webserver/src/server.rs
View file

@ -1,14 +1,17 @@
use crate::{
Http3Request, Http3Response, Router, RouterError, WebMTPConnection, WebServerError,
transport::accept_web_connection,
HttpRequest, HttpResponse, Router, RouterError, WebMTPConnection, WebServerError,
h3::{DriverConfig, run_driver},
tcp::{TcpDriverConfig, run_driver as run_tcp_driver},
};
use bytes::{Buf, Bytes};
use http::{Method, Request, Response, StatusCode};
use http::Method;
use mtp_common::CommunicationError;
use mtp_host::HostConfig;
use rustls::pki_types::{PrivateKeyDer, pem::PemObject};
use std::{net::SocketAddr, sync::Arc, time::Duration};
use tokio::sync::{Semaphore, watch};
use tokio::{
net::TcpListener,
sync::{Semaphore, watch},
};
/// Observability hooks for the web server.
///
@ -23,7 +26,7 @@ pub trait WebServerMetrics: Send + Sync {
fn error_occurred(&self, _error: &WebServerError) {}
}
/// Configuration for the HTTP/3 server and MTP routing.
/// Configuration for the HTTPS/HTTP/3 server and MTP routing.
///
/// Use the builder methods to customise behaviour. All fields have sensible
/// defaults so `WebServerConfig::new()` gives a usable production-ready
@ -34,6 +37,9 @@ pub struct WebServerConfig {
pub(crate) mtp_path: String,
pub max_request_body: usize,
pub max_connections: usize,
pub serve_tcp_https: bool,
pub max_tcp_connections: usize,
pub tls_handshake_timeout: Duration,
pub request_timeout: Duration,
pub drain_timeout: Duration,
pub(crate) metrics: Option<Arc<dyn WebServerMetrics>>,
@ -52,6 +58,9 @@ impl WebServerConfig {
mtp_path: "/".to_string(),
max_request_body: 4 * 1024 * 1024,
max_connections: 256,
serve_tcp_https: true,
max_tcp_connections: 256,
tls_handshake_timeout: Duration::from_secs(10),
request_timeout: Duration::from_secs(30),
drain_timeout: Duration::from_secs(5),
metrics: None,
@ -60,8 +69,8 @@ impl WebServerConfig {
pub fn route<F, Fut>(mut self, path: impl Into<String>, handler: F) -> Result<Self, RouterError>
where
F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Http3Response> + Send + 'static,
F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = HttpResponse> + Send + 'static,
{
self.router = self.router.route(path, handler)?;
Ok(self)
@ -74,8 +83,8 @@ impl WebServerConfig {
handler: F,
) -> Result<Self, RouterError>
where
F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Http3Response> + Send + 'static,
F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = HttpResponse> + Send + 'static,
{
self.router = self.router.route_method(method, path, handler)?;
Ok(self)
@ -89,8 +98,8 @@ impl WebServerConfig {
handler: F,
) -> Result<Self, RouterError>
where
F: Fn(Http3Request, Http3Response, crate::RouteParams) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Http3Response> + Send + 'static,
F: Fn(HttpRequest, HttpResponse, crate::RouteParams) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = HttpResponse> + Send + 'static,
{
self.router = self.router.route_pattern(pattern, handler)?;
Ok(self)
@ -104,8 +113,8 @@ impl WebServerConfig {
handler: F,
) -> Result<Self, RouterError>
where
F: Fn(Http3Request, Http3Response, crate::RouteParams) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Http3Response> + Send + 'static,
F: Fn(HttpRequest, HttpResponse, crate::RouteParams) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = HttpResponse> + Send + 'static,
{
self.router = self.router.route_pattern_method(method, pattern, handler)?;
Ok(self)
@ -113,8 +122,8 @@ impl WebServerConfig {
pub fn fallback<F, Fut>(mut self, handler: F) -> Result<Self, RouterError>
where
F: Fn(Http3Request, Http3Response) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = Http3Response> + Send + 'static,
F: Fn(HttpRequest, HttpResponse) -> Fut + Send + Sync + 'static,
Fut: std::future::Future<Output = HttpResponse> + Send + 'static,
{
self.router = self.router.fallback(handler)?;
Ok(self)
@ -135,6 +144,21 @@ impl WebServerConfig {
self
}
pub fn serve_tcp_https(mut self, enabled: bool) -> Self {
self.serve_tcp_https = enabled;
self
}
pub fn max_tcp_connections(mut self, max: usize) -> Self {
self.max_tcp_connections = max;
self
}
pub fn tls_handshake_timeout(mut self, timeout: Duration) -> Self {
self.tls_handshake_timeout = timeout;
self
}
pub fn request_timeout(mut self, timeout: Duration) -> Self {
self.request_timeout = timeout;
self
@ -151,7 +175,7 @@ impl WebServerConfig {
}
}
/// An HTTP/3 server bound to MTP's configured address and certificate.
/// A combined HTTPS, HTTP/3, and WebTransport server.
///
/// One task owns the Quinn endpoint and dispatches all HTTP/3 requests. This
/// is the required ownership model for adding WebTransport MTP sessions on the
@ -160,52 +184,98 @@ pub struct MTPWebServer {
endpoint: quinn::Endpoint,
mtp_incoming: tokio::sync::mpsc::Receiver<Result<WebMTPConnection, mtp_host::AcceptError>>,
shutdown_tx: watch::Sender<()>,
_driver: Option<tokio::task::JoinHandle<()>>,
quic_driver: Option<tokio::task::JoinHandle<()>>,
tcp_driver: Option<tokio::task::JoinHandle<()>>,
local_addr: SocketAddr,
}
impl MTPWebServer {
pub async fn new(
host_config: HostConfig,
mut host_config: HostConfig,
web_config: WebServerConfig,
) -> Result<Self, CommunicationError> {
mtp_crypto::ensure_crypto_provider();
let certificates =
rustls::pki_types::CertificateDer::pem_slice_iter(&host_config.tls_fullchain)
.collect::<Result<Vec<_>, _>>()
.map_err(|_| CommunicationError::CertificateLoadFailed)?;
let key = PrivateKeyDer::from_pem_slice(&host_config.tls_key)
.map_err(|_| CommunicationError::CertificateParseFailed)?;
let tcp_listener = if web_config.serve_tcp_https {
let listener = TcpListener::bind(SocketAddr::new(host_config.ip, host_config.port))
.await
.map_err(|error| CommunicationError::Other(error.to_string()))?;
host_config.port = listener
.local_addr()
.map_err(|error| CommunicationError::Other(error.to_string()))?
.port();
Some(listener)
} else {
None
};
let tcp_tls = tcp_listener
.as_ref()
.map(|_| build_tcp_tls(&certificates, key.clone_key()))
.transpose()?;
let endpoint = build_endpoint(&host_config, certificates, key)?;
let local_addr = endpoint
.local_addr()
.map_err(|error| CommunicationError::Other(error.to_string()))?;
let host_config = Arc::new(host_config);
let endpoint = build_endpoint(&host_config)?;
let driver_endpoint = endpoint.clone();
// A completed MTP handshake must never block the endpoint driver just
// because the application is briefly slow to call `accept()`.
let (mtp_tx, mtp_incoming) = tokio::sync::mpsc::channel(web_config.max_connections.max(1));
let (shutdown_tx, shutdown_rx) = watch::channel(());
let connection_semaphore = Arc::new(Semaphore::new(web_config.max_connections));
let router = web_config.router.clone();
let metrics = web_config.metrics.clone();
let driver_config = DriverConfig {
router: web_config.router,
mtp_path: web_config.mtp_path,
router: web_config.router.clone(),
mtp_path: web_config.mtp_path.clone(),
max_request_body: web_config.max_request_body,
request_timeout: web_config.request_timeout,
drain_timeout: web_config.drain_timeout,
send_pongs: host_config.send_pongs,
policy: host_config.policy,
host_config,
metrics: web_config.metrics,
metrics: web_config.metrics.clone(),
};
let driver = tokio::spawn(run_driver(
let quic_driver = tokio::spawn(run_driver(
driver_endpoint,
driver_config,
mtp_tx,
connection_semaphore,
shutdown_rx,
shutdown_rx.clone(),
));
let tcp_driver = tcp_listener.zip(tcp_tls).map(|(listener, tls)| {
tokio::spawn(run_tcp_driver(
listener,
tls,
TcpDriverConfig {
router,
max_request_body: web_config.max_request_body,
request_timeout: web_config.request_timeout,
tls_handshake_timeout: web_config.tls_handshake_timeout,
drain_timeout: web_config.drain_timeout,
max_connections: web_config.max_tcp_connections,
metrics,
},
shutdown_rx,
))
});
Ok(Self {
endpoint,
mtp_incoming,
shutdown_tx,
_driver: Some(driver),
quic_driver: Some(quic_driver),
tcp_driver,
local_addr,
})
}
pub fn local_addr(&self) -> SocketAddr {
self.endpoint
.local_addr()
.expect("endpoint has a local address")
self.local_addr
}
pub async fn accept(&mut self) -> Result<Option<WebMTPConnection>, mtp_host::AcceptError> {
@ -222,7 +292,10 @@ impl MTPWebServer {
/// in-flight requests to complete before closing the endpoint.
pub async fn shutdown(mut self) {
let _ = self.shutdown_tx.send(());
if let Some(driver) = self._driver.take() {
if let Some(driver) = self.quic_driver.take() {
let _ = driver.await;
}
if let Some(driver) = self.tcp_driver.take() {
let _ = driver.await;
}
self.endpoint
@ -233,40 +306,33 @@ impl MTPWebServer {
pub async fn close(mut self) {
self.endpoint
.close(quinn::VarInt::from_u32(0), b"mtp-webserver shutdown");
if let Some(driver) = self._driver.take() {
if let Some(driver) = self.quic_driver.take() {
driver.abort();
let _ = driver.await;
}
if let Some(driver) = self.tcp_driver.take() {
driver.abort();
let _ = driver.await;
}
}
}
impl Drop for MTPWebServer {
fn drop(&mut self) {
if let Some(driver) = self._driver.take() {
if let Some(driver) = self.quic_driver.take() {
driver.abort();
}
if let Some(driver) = self.tcp_driver.take() {
driver.abort();
}
}
}
struct DriverConfig {
router: Router,
mtp_path: String,
max_request_body: usize,
request_timeout: Duration,
drain_timeout: Duration,
send_pongs: bool,
policy: mtp_transport::Policy,
host_config: Arc<HostConfig>,
metrics: Option<Arc<dyn WebServerMetrics>>,
}
fn build_endpoint(config: &HostConfig) -> Result<quinn::Endpoint, CommunicationError> {
mtp_crypto::ensure_crypto_provider();
let certificates = rustls::pki_types::CertificateDer::pem_slice_iter(&config.tls_fullchain)
.collect::<Result<Vec<_>, _>>()
.map_err(|_| CommunicationError::CertificateLoadFailed)?;
let key = PrivateKeyDer::from_pem_slice(&config.tls_key)
.map_err(|_| CommunicationError::CertificateParseFailed)?;
fn build_endpoint(
config: &HostConfig,
certificates: Vec<rustls::pki_types::CertificateDer<'static>>,
key: PrivateKeyDer<'static>,
) -> Result<quinn::Endpoint, CommunicationError> {
let mut tls = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certificates, key)
@ -281,373 +347,14 @@ fn build_endpoint(config: &HostConfig) -> Result<quinn::Endpoint, CommunicationE
.map_err(|error| CommunicationError::Other(error.to_string()))
}
async fn run_driver(
endpoint: quinn::Endpoint,
config: DriverConfig,
mtp_tx: tokio::sync::mpsc::Sender<Result<WebMTPConnection, mtp_host::AcceptError>>,
connection_semaphore: Arc<Semaphore>,
mut shutdown_rx: watch::Receiver<()>,
) {
let DriverConfig {
router,
mtp_path,
max_request_body,
request_timeout,
drain_timeout,
send_pongs,
policy,
host_config,
metrics,
} = config;
let mut connection_tasks = tokio::task::JoinSet::new();
loop {
tokio::select! {
biased;
_ = shutdown_rx.changed() => {
break;
}
incoming = endpoint.accept() => {
let Some(incoming) = incoming else {
break;
};
// Do not await capacity here: doing so would prevent this loop
// from observing shutdown while all connection slots are in use.
let permit = match connection_semaphore.clone().try_acquire_owned() {
Ok(permit) => permit,
Err(_) => {
tracing::debug!("rejecting QUIC connection at configured connection limit");
continue;
}
};
let router = router.clone();
let mtp_path = mtp_path.clone();
let mtp_tx = mtp_tx.clone();
let metrics = metrics.clone();
let host_config = host_config.clone();
connection_tasks.spawn(async move {
let _permit = permit;
let connect_start = std::time::Instant::now();
let connection = match incoming.await {
Ok(connection) => connection,
Err(error) => {
tracing::debug!(%error, "QUIC connection failed during handshake");
return;
}
};
let mut builder = h3::server::builder();
builder.enable_extended_connect(true);
builder.enable_webtransport(true);
builder.enable_datagram(true);
builder.max_webtransport_sessions(16);
let mut h3 = match builder
.build(h3_quinn::Connection::new(connection.clone()))
.await
{
Ok(connection) => connection,
Err(error) => {
tracing::debug!(%error, "HTTP/3 connection setup failed");
return;
}
};
if let Some(ref m) = metrics {
m.connection_accepted();
}
let remote_addr = connection.remote_address();
let mut tasks = tokio::task::JoinSet::new();
loop {
let resolver = match h3.accept().await {
Ok(Some(resolver)) => resolver,
Ok(None) => break,
Err(error) => {
tracing::debug!(%error, "HTTP/3 request accept failed");
break;
}
};
let (request, mut stream) = match resolver.resolve_request().await {
Ok(request) => request,
Err(error) => {
tracing::debug!(%error, "HTTP/3 request parse failed");
continue;
}
};
if request.method() == http::Method::CONNECT && request.uri().path() == mtp_path {
if request.extensions().get::<h3::ext::Protocol>()
!= Some(&h3::ext::Protocol::WEB_TRANSPORT)
{
let _ = stream
.send_response(
Response::builder()
.status(StatusCode::METHOD_NOT_ALLOWED)
.body(())
.unwrap(),
)
.await;
let _ = stream.finish().await;
continue;
}
let session = match h3_webtransport::server::WebTransportSession::accept(
request, stream, h3,
)
.await
{
Ok(session) => Arc::new(session),
Err(error) => {
tracing::debug!(%error, "WebTransport session accept failed");
return;
}
};
// The WebTransport session request driver must outlive this
// endpoint request task. Keep it detached so handing the MTP
// connection to the application does not wait for the session
// (which is intentionally an open-ended accept loop).
tokio::spawn(run_session_requests(
session.clone(),
router.clone(),
max_request_body,
request_timeout,
metrics.clone(),
remote_addr,
));
let result =
accept_web_connection(session, mtp_path, connection, send_pongs, policy, host_config.clone())
.await;
match mtp_tx.try_send(result) {
Ok(()) => {
// The detached session driver remains active while the
// delivered MTP connection keeps the session alive.
}
Err(tokio::sync::mpsc::error::TrySendError::Full(result)) => {
tracing::warn!("MTP connection backlog is full; dropping connection");
if let Ok(connection) = result {
connection.sender.close();
}
}
Err(tokio::sync::mpsc::error::TrySendError::Closed(result)) => {
if let Ok(connection) = result {
connection.sender.close();
}
}
}
return;
}
let router = router.clone();
let metrics = metrics.clone();
tasks.spawn(async move {
let path = request.uri().path().to_string();
if let Some(ref m) = metrics {
m.request_started(&path);
}
let req_start = std::time::Instant::now();
let (response, status) = match tokio::time::timeout(
request_timeout,
handle_http_request(request, &mut stream, &router, max_request_body, remote_addr),
)
.await
{
Ok(Ok((resp, status))) => (resp, status),
Ok(Err(_)) => (
Http3Response::new(StatusCode::BAD_GATEWAY),
StatusCode::BAD_GATEWAY,
),
Err(_) => (
Http3Response::new(StatusCode::REQUEST_TIMEOUT),
StatusCode::REQUEST_TIMEOUT,
),
};
let _ = write_response(&mut stream, response).await;
if let Some(ref m) = metrics {
m.request_completed(&path, status.as_u16(), req_start.elapsed());
}
});
}
tasks.join_all().await;
if let Some(ref m) = metrics {
m.connection_closed(connect_start.elapsed(), "normal");
}
});
}
}
}
// --- Drain phase: wait for in-flight connections ---
endpoint.close(quinn::VarInt::from_u32(0), b"mtp-webserver shutdown");
let drain_start = std::time::Instant::now();
while !connection_tasks.is_empty() {
tokio::select! {
Some(result) = connection_tasks.join_next() => {
if let Err(e) = result {
tracing::warn!("Connection task panicked: {}", e);
}
}
_ = tokio::time::sleep(drain_timeout.saturating_sub(drain_start.elapsed())) => {
tracing::warn!(
"Drain timeout expired with {} connections still in flight",
connection_tasks.len()
);
break;
}
}
}
connection_tasks.shutdown().await;
}
async fn handle_http_request<S>(
request: Request<()>,
stream: &mut h3::server::RequestStream<S, Bytes>,
router: &Router,
max_request_body: usize,
remote_addr: SocketAddr,
) -> Result<(Http3Response, StatusCode), WebServerError>
where
S: h3::quic::BidiStream<Bytes>,
{
let (request, too_large) = read_request(request, stream, max_request_body, remote_addr)
.await
.map_err(|e| WebServerError::Http(format!("request body read failed: {e}")))?;
if too_large {
return Ok((
Http3Response::new(StatusCode::PAYLOAD_TOO_LARGE),
StatusCode::PAYLOAD_TOO_LARGE,
));
}
let path = request.uri.path().to_string();
match router.handler(&request.method, &path) {
Some(handler) => {
let response = handler(request, Http3Response::default()).await;
let status = response.status;
Ok((response, status))
}
None => match router.pattern_handler(&request.method, &path) {
Some((handler, params)) => {
let response = handler(request, Http3Response::default(), params).await;
let status = response.status;
Ok((response, status))
}
None => match router.fallback_handler() {
Some(handler) => {
let response = handler(request, Http3Response::default()).await;
let status = response.status;
Ok((response, status))
}
None => Ok((
Http3Response::new(StatusCode::NOT_FOUND),
StatusCode::NOT_FOUND,
)),
},
},
}
}
async fn run_session_requests(
session: Arc<h3_webtransport::server::WebTransportSession<h3_quinn::Connection, Bytes>>,
router: Router,
max_request_body: usize,
request_timeout: Duration,
metrics: Option<Arc<dyn WebServerMetrics>>,
remote_addr: SocketAddr,
) {
loop {
match session.accept_bi().await {
Ok(Some(h3_webtransport::server::AcceptedBi::Request(request, mut stream))) => {
let router = router.clone();
let metrics = metrics.clone();
tokio::spawn(async move {
let path = request.uri().path().to_string();
if let Some(ref m) = metrics {
m.request_started(&path);
}
let req_start = std::time::Instant::now();
let (response, status) = match tokio::time::timeout(
request_timeout,
handle_http_request(
request,
&mut stream,
&router,
max_request_body,
remote_addr,
),
)
.await
{
Ok(Ok((resp, status))) => (resp, status),
Ok(Err(_)) => (
Http3Response::new(StatusCode::BAD_GATEWAY),
StatusCode::BAD_GATEWAY,
),
Err(_) => (
Http3Response::new(StatusCode::REQUEST_TIMEOUT),
StatusCode::REQUEST_TIMEOUT,
),
};
let _ = write_response(&mut stream, response).await;
if let Some(ref m) = metrics {
m.request_completed(&path, status.as_u16(), req_start.elapsed());
}
});
}
Ok(Some(h3_webtransport::server::AcceptedBi::BidiStream(_, _))) => {}
Ok(None) | Err(_) => break,
}
}
}
async fn read_request<S>(
request: Request<()>,
stream: &mut h3::server::RequestStream<S, Bytes>,
max_body: usize,
remote_addr: SocketAddr,
) -> Result<(Http3Request, bool), h3::error::StreamError>
where
S: h3::quic::BidiStream<Bytes>,
{
let (parts, _) = request.into_parts();
let mut body = Vec::new();
let mut too_large = false;
while let Some(chunk) = stream.recv_data().await? {
if body.len().saturating_add(chunk.remaining()) > max_body {
too_large = true;
break;
}
body.extend_from_slice(chunk.chunk());
}
Ok((
Http3Request {
method: parts.method,
uri: parts.uri,
headers: parts.headers,
body: (!body.is_empty()).then(|| Bytes::from(body)),
remote_addr,
},
too_large,
))
}
async fn write_response<S>(
stream: &mut h3::server::RequestStream<S, Bytes>,
response: Http3Response,
) -> Result<(), h3::error::StreamError>
where
S: h3::quic::BidiStream<Bytes>,
{
let mut builder = Response::builder().status(response.status);
for (name, value) in &response.headers {
builder = builder.header(name, value);
}
stream
.send_response(builder.body(()).expect("valid HTTP response"))
.await?;
for chunk in response.body {
stream.send_data(chunk).await?;
}
if let Some(mut chunks) = response.stream {
while let Some(chunk) = chunks.recv().await {
stream.send_data(chunk).await?;
}
}
stream.finish().await
fn build_tcp_tls(
certificates: &[rustls::pki_types::CertificateDer<'static>],
key: PrivateKeyDer<'static>,
) -> Result<Arc<rustls::ServerConfig>, CommunicationError> {
let mut tls = rustls::ServerConfig::builder()
.with_no_client_auth()
.with_single_cert(certificates.to_vec(), key)
.map_err(|_| CommunicationError::CertificateLoadFailed)?;
tls.alpn_protocols = vec![b"h2".to_vec(), b"http/1.1".to_vec()];
Ok(Arc::new(tls))
}