Upgrade most HTTP/Hyper crates and make mas-listener work

Author: sandhose

Cargo.lock

@@ -68,6 +68,15 @@ version = "0.1.80"
 [workspace.dependencies.anyhow]
 version = "1.0.86"
 
+# HTTP router
+[workspace.dependencies.axum]
+version = "0.7.5"
+
+# Extra utilities for Axum
+[workspace.dependencies.axum-extra]
+version = "0.9.3"
+features = ["cookie-private", "cookie-key-expansion", "typed-header"]
+
 # UTF-8 paths
 [workspace.dependencies.camino]
 version = "1.1.7"
@@ -94,20 +103,25 @@ version = "0.3.9"
 
 # HTTP request/response
 [workspace.dependencies.http]
-version = "0.2.12"
+version = "1.1.0"
 
 # HTTP body trait
 [workspace.dependencies.http-body]
-version = "0.4.6"
+version = "1.0.0"
 
-# HTTP client
+# HTTP client and server
 [workspace.dependencies.hyper]
-version = "0.14.29"
-features = ["client", "http1", "http2", "stream", "runtime"]
+version = "1.4.0"
+features = ["client", "http1", "http2"]
+
+# Additional Hyper utilties
+[workspace.dependencies.hyper-util]
+version = "0.1.6"
+features = ["server", "server-auto", "service", "http1", "http2", "tokio"]
 
 # Hyper Rustls support
 [workspace.dependencies.hyper-rustls]
-version = "0.25.0"
+version = "0.27.2"
 features = ["http1", "http2"]
 default-features = false
 

Cargo.toml

@@ -13,8 +13,8 @@ workspace = true
 
 [dependencies]
 async-trait.workspace = true
-axum = { version = "0.6.20", features = ["headers"] }
-axum-extra = { version = "0.8.0", features = ["cookie-private", "cookie-key-expansion"] }
+axum.workspace = true
+axum-extra.workspace = true
 chrono.workspace = true
 data-encoding = "2.6.0"
 futures-util = "0.3.30"

crates/axum-utils/Cargo.toml

@@ -13,7 +13,7 @@ workspace = true
 
 [dependencies]
 anyhow.workspace = true
-axum = "0.6.20"
+axum.workspace = true
 camino.workspace = true
 clap.workspace = true
 console = "0.15.8"

crates/cli/Cargo.toml

@@ -31,9 +31,9 @@ sentry = { version = "0.31.8", default-features = false }
 hyper.workspace = true
 tower.workspace = true
 tower-http = { version = "0.4.4", features = ["cors"] }
-axum = "0.6.20"
-axum-macros = "0.3.8"
-axum-extra = { version = "0.8.0", features = ["cookie-private"] }
+axum.workspace = true
+axum-macros = "0.4.1"
+axum-extra.workspace = true
 
 async-graphql.workspace = true
 
@@ -52,7 +52,12 @@ serde_urlencoded = "0.7.1"
 # Password hashing
 argon2 = { version = "0.5.3", features = ["password-hash", "std"] }
 bcrypt = "0.15.1"
-pbkdf2 = { version = "0.12.2", features = ["password-hash", "std", "simple", "parallel"] }
+pbkdf2 = { version = "0.12.2", features = [
+    "password-hash",
+    "std",
+    "simple",
+    "parallel",
+] }
 zeroize = "1.8.1"
 
 # Various data types and utilities

crates/handlers/Cargo.toml

@@ -40,13 +40,12 @@ tokio.workspace = true
 
 [features]
 client = [
-  "dep:mas-tower",
-  "dep:rustls",
-  "hyper/tcp",
-  "dep:hyper-rustls",
-  "dep:rustls-platform-verifier",
-  "tower/limit",
-  "tower-http/timeout",
-  "tower-http/follow-redirect",
-  "tower-http/set-header",
+    "dep:mas-tower",
+    "dep:rustls",
+    "dep:hyper-rustls",
+    "dep:rustls-platform-verifier",
+    "tower/limit",
+    "tower-http/timeout",
+    "tower-http/follow-redirect",
+    "tower-http/set-header",
 ]

crates/http/Cargo.toml

@@ -17,14 +17,15 @@ event-listener = "5.3.1"
 futures-util = "0.3.30"
 http-body.workspace = true
 hyper = { workspace = true, features = ["server"] }
+hyper-util.workspace = true
 libc = "0.2.155"
 pin-project-lite = "0.2.14"
 socket2 = "0.5.7"
 thiserror.workspace = true
 tokio.workspace = true
 tokio-rustls = "0.25.0"
-tower-http = { version = "0.4.4", features = ["add-extension"] }
-tower-service = "0.3.2"
+tower.workspace = true
+tower-http = { version = "0.5.2", features = ["add-extension"] }
 tracing.workspace = true
 
 [dev-dependencies]

crates/listener/Cargo.toml

@@ -21,16 +21,17 @@ use std::{
 };
 
 use anyhow::Context;
-use hyper::{service::service_fn, Request, Response};
+use hyper::{Request, Response};
 use mas_listener::{server::Server, shutdown::ShutdownStream, ConnectionInfo};
 use tokio::signal::unix::SignalKind;
 use tokio_rustls::rustls::{server::WebPkiClientVerifier, RootCertStore, ServerConfig};
+use tower::service_fn;
 
 static CA_CERT_PEM: &[u8] = include_bytes!("./certs/ca.pem");
 static SERVER_CERT_PEM: &[u8] = include_bytes!("./certs/server.pem");
 static SERVER_KEY_PEM: &[u8] = include_bytes!("./certs/server-key.pem");
 
-async fn handler(req: Request<hyper::Body>) -> Result<Response<String>, Infallible> {
+async fn handler(req: Request<hyper::body::Incoming>) -> Result<Response<String>, Infallible> {
     tracing::info!("Handling request");
     tokio::time::sleep(Duration::from_secs(3)).await;
     let info = req.extensions().get::<ConnectionInfo>().unwrap();

crates/listener/examples/demo/main.rs

@@ -22,14 +22,17 @@ use std::{
 
 use event_listener::{Event, EventListener};
 use futures_util::{stream::SelectAll, Stream, StreamExt};
-use http_body::Body;
-use hyper::{body::HttpBody, server::conn::Connection, Request, Response};
+use hyper::{Request, Response};
+use hyper_util::{
+    rt::{TokioExecutor, TokioIo},
+    server::conn::auto::Connection,
+    service::TowerToHyperService,
+};
 use pin_project_lite::pin_project;
 use thiserror::Error;
-use tokio::io::{AsyncRead, AsyncWrite};
 use tokio_rustls::rustls::ServerConfig;
+use tower::Service;
 use tower_http::add_extension::AddExtension;
-use tower_service::Service;
 use tracing::Instrument;
 
 use crate::{
@@ -91,10 +94,10 @@ impl<S> Server<S> {
     /// Run a single server
     pub async fn run<B, SD>(self, shutdown: SD)
     where
-        S: Service<Request<hyper::Body>, Response = Response<B>> + Clone + Send + 'static,
+        S: Service<Request<hyper::body::Incoming>, Response = Response<B>> + Clone + Send + 'static,
         S::Future: Send + 'static,
         S::Error: std::error::Error + Send + Sync + 'static,
-        B: Body + Send + 'static,
+        B: http_body::Body + Send + 'static,
         B::Data: Send,
         B::Error: std::error::Error + Send + Sync + 'static,
         SD: Stream + Unpin,
@@ -173,15 +176,20 @@ async fn accept<S, B>(
     stream: UnixOrTcpConnection,
     service: S,
 ) -> Result<
-    Connection<MaybeTlsStream<Rewind<UnixOrTcpConnection>>, AddExtension<S, ConnectionInfo>>,
+    Connection<
+        'static,
+        TokioIo<MaybeTlsStream<Rewind<UnixOrTcpConnection>>>,
+        TowerToHyperService<AddExtension<S, ConnectionInfo>>,
+        TokioExecutor,
+    >,
     AcceptError,
 >
 where
-    S: Service<Request<hyper::Body>, Response = Response<B>>,
+    S: Service<Request<hyper::body::Incoming>, Response = Response<B>> + Send + Clone + 'static,
     S::Error: std::error::Error + Send + Sync + 'static,
     S::Future: Send + 'static,
-    B: HttpBody + Send + 'static,
-    B::Data: Send + 'static,
+    B: http_body::Body + Send + 'static,
+    B::Data: Send,
     B::Error: std::error::Error + Send + Sync + 'static,
 {
     let span = tracing::Span::current();
@@ -219,18 +227,17 @@ where
             net_peer_addr: peer_addr.into_net(),
         };
 
-        let service = AddExtension::new(service, info);
-
-        let conn = if is_h2 {
-            hyper::server::conn::Http::new()
-                .http2_only(true)
-                .serve_connection(stream, service)
-        } else {
-            hyper::server::conn::Http::new()
-                .http1_only(true)
-                .http1_keep_alive(true)
-                .serve_connection(stream, service)
-        };
+        let mut builder = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
+        if is_h2 {
+            builder = builder.http2_only();
+        }
+        builder.http1().keep_alive(true);
+
+        let service = TowerToHyperService::new(AddExtension::new(service, info));
+
+        let conn = builder
+            .serve_connection(TokioIo::new(stream), service)
+            .into_owned();
 
         Ok(conn)
     })
@@ -270,18 +277,19 @@ impl<C> AbortableConnection<C> {
     }
 }
 
-impl<T, S, B> Future for AbortableConnection<Connection<T, S>>
+impl<T, S, B> Future
+    for AbortableConnection<Connection<'static, T, TowerToHyperService<S>, TokioExecutor>>
 where
-    Connection<T, S>: Future,
-    S: Service<Request<hyper::Body>, Response = Response<B>> + Send + 'static,
+    Connection<'static, T, TowerToHyperService<S>, TokioExecutor>: Future,
+    S: Service<Request<hyper::body::Incoming>, Response = Response<B>> + Send + Clone + 'static,
     S::Future: Send + 'static,
     S::Error: std::error::Error + Send + Sync,
-    B: HttpBody + Send + 'static,
+    T: hyper::rt::Read + hyper::rt::Write + Unpin,
+    B: http_body::Body + Send + 'static,
     B::Data: Send,
-    B::Error: std::error::Error + Send + Sync,
-    T: AsyncRead + AsyncWrite + Unpin,
+    B::Error: std::error::Error + Send + Sync + 'static,
 {
-    type Output = <Connection<T, S> as Future>::Output;
+    type Output = <Connection<'static, T, TowerToHyperService<S>, TokioExecutor> as Future>::Output;
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let mut this = self.project();
@@ -308,10 +316,10 @@ where
 #[allow(clippy::too_many_lines)]
 pub async fn run_servers<S, B, SD>(listeners: impl IntoIterator<Item = Server<S>>, mut shutdown: SD)
 where
-    S: Service<Request<hyper::Body>, Response = Response<B>> + Clone + Send + 'static,
+    S: Service<Request<hyper::body::Incoming>, Response = Response<B>> + Clone + Send + 'static,
     S::Future: Send + 'static,
     S::Error: std::error::Error + Send + Sync + 'static,
-    B: Body + Send + 'static,
+    B: http_body::Body + Send + 'static,
     B::Data: Send,
     B::Error: std::error::Error + Send + Sync + 'static,
     SD: Stream + Unpin,

crates/listener/src/server.rs

@@ -12,7 +12,7 @@ publish = false
 workspace = true
 
 [dependencies]
-axum = { version = "0.6.20", default-features = false }
+axum.workspace = true
 serde.workspace = true
 serde_urlencoded = "0.7.1"
 url.workspace = true