You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2026/07/16 10:54:39 UTC
(iggy) branch master updated: feat(server-ng): serve HTTPS on the REST listener (#3672)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2e56b5fc0 feat(server-ng): serve HTTPS on the REST listener (#3672)
2e56b5fc0 is described below
commit 2e56b5fc0b6d2776936d995190544e5b3062e2b6
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Thu Jul 16 12:54:35 2026 +0200
feat(server-ng): serve HTTPS on the REST listener (#3672)
http.tls.enabled was read but never acted on: the REST listener
always bound plaintext, so bearer tokens and passwords crossed
the wire in clear despite TLS being configured.
Terminate TLS with the compio-native rustls substrate the binary
transports already use, rather than pulling tokio via axum-server.
Handshakes run off the accept path behind a bounded channel so a
slow peer cannot stall accept. Connections are served through a
hand-rolled hyper-util auto loop over HyperStream::new_tls;
cyper_axum::serve only wraps plaintext streams, and splitting the
TLS stream deadlocks full-duplex HTTP on compio's BiLock. ALPN
advertises h2 and http/1.1 for parity with the legacy server.
Credential errors surface as ListenerCredentials instead of a
panic, and enabling TLS with an empty cert_file or key_file now
fails config validation at boot.
An integration test spawns iggy-server-ng with http.tls enabled
and asserts /ping and /users/login return 200 over HTTP/2,
proving certificate loading, the handshake pump, and ALPN end to
end. reqwest's http2 feature is now pinned instead of enabled
transitively via testcontainers.
---
Cargo.lock | 3 +
Cargo.toml | 3 +
core/configs/src/server_ng_config/validators.rs | 33 +++
core/integration/Cargo.toml | 2 +-
core/integration/tests/server/http_tls.rs | 158 ++++++++++++++
core/integration/tests/server/mod.rs | 4 +
core/server-ng/Cargo.toml | 3 +
core/server-ng/src/http.rs | 41 ++--
core/server-ng/src/http/tls.rs | 266 ++++++++++++++++++++++++
9 files changed, 498 insertions(+), 15 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index ef4a301b1..22cbf3cfe 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -11790,6 +11790,7 @@ dependencies = [
"ctrlc",
"cyper",
"cyper-axum",
+ "cyper-core",
"dashmap",
"dotenvy",
"err_trail",
@@ -11800,6 +11801,8 @@ dependencies = [
"hash32 1.0.0",
"human-repr",
"hwlocality",
+ "hyper",
+ "hyper-util",
"iggy",
"iggy_binary_protocol",
"iggy_common",
diff --git a/Cargo.toml b/Cargo.toml
index 537ae8102..87a96add3 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -148,6 +148,7 @@ ctrlc = { version = "3.5", features = ["termination"] }
cucumber = "0.23"
cyper = { version = "0.9.0", features = ["rustls"], default-features = false }
cyper-axum = { version = "0.9.0" }
+cyper-core = { version = "0.9.0", default-features = false }
darling = "0.23"
dashmap = "6.2.1"
deltalake = { version = "0.32.4", features = ["azure", "gcs", "s3"] }
@@ -190,6 +191,8 @@ http = "1.4.2"
human-repr = "1.1.0"
humantime = "2.3.0"
hwlocality = "1.0.0-alpha.12"
+hyper = "1.10.1"
+hyper-util = { version = "0.1.20", features = ["server-auto", "service"] }
iceberg = "0.9.1"
iceberg-catalog-rest = "0.9.1"
iceberg-storage-opendal = "0.9.1"
diff --git a/core/configs/src/server_ng_config/validators.rs b/core/configs/src/server_ng_config/validators.rs
index 427a081e8..0ab7939d3 100644
--- a/core/configs/src/server_ng_config/validators.rs
+++ b/core/configs/src/server_ng_config/validators.rs
@@ -124,6 +124,16 @@ impl Validatable<ConfigurationError> for ServerNgConfig {
return Err(ConfigurationError::InvalidConfigurationValue);
}
+ if self.http.enabled
+ && self.http.tls.enabled
+ && (self.http.tls.cert_file.is_empty() || self.http.tls.key_file.is_empty())
+ {
+ eprintln!(
+ "http.tls.enabled=true requires non-empty http.tls.cert_file and http.tls.key_file"
+ );
+ return Err(ConfigurationError::InvalidConfigurationValue);
+ }
+
if topic_size < self.system.segment.size.as_bytes_u64() {
eprintln!(
"system.topic.max_size ({} B) must be >= system.segment.size ({} B)",
@@ -373,4 +383,27 @@ mod tests {
defaults.message_saver.enforce_fsync
);
}
+
+ // http.enabled needs a non-ServerDefault JWT expiry to clear the sibling
+ // check above; ServerNgConfig::default() already satisfies that.
+ fn https_config(cert_file: &str, key_file: &str) -> ServerNgConfig {
+ let mut cfg = ServerNgConfig::default();
+ cfg.http.enabled = true;
+ cfg.http.tls.enabled = true;
+ cfg.http.tls.cert_file = cert_file.to_string();
+ cfg.http.tls.key_file = key_file.to_string();
+ cfg
+ }
+
+ #[test]
+ fn validate_rejects_tls_enabled_with_empty_cert_file() {
+ let cfg = https_config("", "key.pem");
+ assert!(cfg.validate().is_err());
+ }
+
+ #[test]
+ fn validate_accepts_tls_enabled_with_both_files_set() {
+ let cfg = https_config("cert.pem", "key.pem");
+ assert!(cfg.validate().is_ok());
+ }
}
diff --git a/core/integration/Cargo.toml b/core/integration/Cargo.toml
index 62168b3f2..9569d4d1d 100644
--- a/core/integration/Cargo.toml
+++ b/core/integration/Cargo.toml
@@ -66,7 +66,7 @@ mongodb = { workspace = true }
predicates = { workspace = true }
rand = { workspace = true }
rcgen = { workspace = true }
-reqwest = { workspace = true, features = ["json"] }
+reqwest = { workspace = true, features = ["http2", "json"] }
reqwest-middleware = { workspace = true }
reqwest-retry = { workspace = true }
rmcp = { workspace = true, features = [
diff --git a/core/integration/tests/server/http_tls.rs b/core/integration/tests/server/http_tls.rs
new file mode 100644
index 000000000..80bd58910
--- /dev/null
+++ b/core/integration/tests/server/http_tls.rs
@@ -0,0 +1,158 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! End-to-end HTTPS proof for the server-ng shard-0 REST listener. The TLS
+//! accept pump ([`server_ng::http::tls`]) and the `start()` HTTPS branch
+//! unit-test in isolation; this is the only path that drives a live rustls
+//! client over the wire and proves the response was actually served over
+//! HTTP/2, negotiated via ALPN. A failure here (h1 fallback, handshake
+//! rejection, or a plaintext bind) is a real defect in that path, not a test
+//! artifact.
+
+use std::path::PathBuf;
+use std::time::{Duration, Instant};
+
+use iggy::prelude::*;
+use integration::harness::TestHarness;
+use reqwest::{Certificate, StatusCode, Version};
+use serde_json::json;
+use tokio::time::sleep;
+
+/// The harness readiness gate fires on the dumped runtime config, which the
+/// server writes before the HTTPS listener finishes binding; a bounded
+/// connect-retry loop bridges that gap.
+const READY_TIMEOUT: Duration = Duration::from_secs(15);
+const READY_RETRY_INTERVAL: Duration = Duration::from_millis(50);
+const REQUEST_TIMEOUT: Duration = Duration::from_secs(30);
+
+/// Absolute path to a repo loopback cert asset. The spawned server's CWD is a
+/// temp dir, so relative paths break; `CARGO_MANIFEST_DIR` is the integration
+/// crate, whose sibling `../certs` holds the checked-in loopback material.
+fn cert_asset(file: &str) -> PathBuf {
+ std::fs::canonicalize(format!("{}/../certs/{file}", env!("CARGO_MANIFEST_DIR")))
+ .unwrap_or_else(|error| panic!("canonicalize repo cert asset {file}: {error}"))
+}
+
+/// Boot iggy-server-ng with `[http.tls]` enabled against the repo loopback
+/// cert, then prove a real HTTPS request is served over HTTP/2. Single node:
+/// the HTTP listener is shard-0-only and cluster formation is irrelevant to
+/// what we are proving, so a standalone node keeps the signal clean.
+#[tokio::test]
+#[serial_test::parallel]
+async fn given_http_tls_enabled_when_pinging_should_serve_https_over_http2() {
+ let mut harness = TestHarness::builder()
+ .default_server()
+ .cluster_nodes(1)
+ .build()
+ .expect("build TLS harness");
+
+ let cert = cert_asset("iggy_cert.pem");
+ let key = cert_asset("iggy_key.pem");
+ harness
+ .server_mut()
+ .add_env("IGGY_HTTP_TLS_ENABLED", "true");
+ harness
+ .server_mut()
+ .add_env("IGGY_HTTP_TLS_CERT_FILE", cert.display().to_string());
+ harness
+ .server_mut()
+ .add_env("IGGY_HTTP_TLS_KEY_FILE", key.display().to_string());
+
+ harness
+ .start()
+ .await
+ .expect("start server-ng with HTTPS enabled");
+
+ let addr = harness
+ .server()
+ .http_addr()
+ .expect("HTTP transport not configured on test server");
+ // The loopback leaf's SAN is `DNS:localhost` only (no IP), so the client
+ // must dial by hostname for hostname verification to pass.
+ let base_url = format!("https://localhost:{}", addr.port());
+
+ // Real chain + hostname verification: trust the repo CA that signed the
+ // loopback leaf and let ALPN auto-negotiate (no prior-knowledge, which
+ // would bypass ALPN). rustls is the only TLS backend compiled in.
+ let ca =
+ Certificate::from_pem(&std::fs::read(cert_asset("iggy_ca_cert.pem")).expect("read CA"))
+ .expect("parse CA pem");
+ let client = reqwest::Client::builder()
+ .timeout(REQUEST_TIMEOUT)
+ .use_rustls_tls()
+ .add_root_certificate(ca)
+ .build()
+ .expect("build HTTPS client");
+
+ // GET /ping, retrying only connect failures while the listener finishes
+ // binding after readiness; any other error is a real handshake/protocol
+ // defect and must surface.
+ let deadline = Instant::now() + READY_TIMEOUT;
+ let ping = loop {
+ match client.get(format!("{base_url}/ping")).send().await {
+ Ok(response) => break response,
+ Err(error) if error.is_connect() => {
+ assert!(
+ Instant::now() < deadline,
+ "HTTPS /ping never connected within {READY_TIMEOUT:?}: {error}"
+ );
+ sleep(READY_RETRY_INTERVAL).await;
+ }
+ Err(error) => panic!("HTTPS /ping failed (not a connect error): {error}"),
+ }
+ };
+ assert_eq!(
+ ping.status(),
+ StatusCode::OK,
+ "ping must answer 200 over HTTPS"
+ );
+ assert_eq!(
+ ping.version(),
+ Version::HTTP_2,
+ "ALPN must negotiate HTTP/2 (server advertises h2 first), got {:?}",
+ ping.version()
+ );
+
+ // A real request/response body over TLS, not just a static probe: root
+ // login must round-trip a token.
+ let login = client
+ .post(format!("{base_url}/users/login"))
+ .json(&json!({
+ "username": DEFAULT_ROOT_USERNAME,
+ "password": DEFAULT_ROOT_PASSWORD,
+ }))
+ .send()
+ .await
+ .expect("root login request over HTTPS");
+ assert_eq!(
+ login.status(),
+ StatusCode::OK,
+ "root login must answer 200 over HTTPS"
+ );
+ assert_eq!(
+ login.version(),
+ Version::HTTP_2,
+ "login response must also be HTTP/2"
+ );
+ let identity: IdentityInfo = login.json().await.expect("decode IdentityInfo over HTTPS");
+ assert!(
+ identity
+ .access_token
+ .is_some_and(|token| !token.token.is_empty()),
+ "login over HTTPS must return a non-empty access token"
+ );
+}
diff --git a/core/integration/tests/server/mod.rs b/core/integration/tests/server/mod.rs
index 015df6978..25cac503d 100644
--- a/core/integration/tests/server/mod.rs
+++ b/core/integration/tests/server/mod.rs
@@ -37,6 +37,10 @@ mod http_vsr;
// bodies); the RBAC matrix lives in permissions_scenario.
#[cfg(feature = "vsr")]
mod http_rbac;
+// End-to-end HTTPS: server-ng serves the REST listener over TLS and negotiates
+// HTTP/2 via ALPN.
+#[cfg(feature = "vsr")]
+mod http_tls;
// Binary GetClusterMetadata must serve the real roster from a VSR cluster.
#[cfg(feature = "vsr")]
mod cluster_metadata_vsr;
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index 6601ef23f..b175aff16 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -108,6 +108,7 @@ crossfire = { workspace = true }
ctrlc = { workspace = true }
cyper = { workspace = true }
cyper-axum = { workspace = true }
+cyper-core = { workspace = true }
dashmap = { workspace = true }
dotenvy = { workspace = true }
err_trail = { workspace = true }
@@ -117,6 +118,8 @@ fs2 = { workspace = true }
futures = { workspace = true }
hash32 = { workspace = true }
human-repr = { workspace = true }
+hyper = { workspace = true }
+hyper-util = { workspace = true }
iggy_binary_protocol = { workspace = true }
iggy_common = { workspace = true }
journal = { workspace = true }
diff --git a/core/server-ng/src/http.rs b/core/server-ng/src/http.rs
index 2d4f53766..26aab3afc 100644
--- a/core/server-ng/src/http.rs
+++ b/core/server-ng/src/http.rs
@@ -33,6 +33,7 @@ mod reply;
mod session;
mod state;
mod submit;
+mod tls;
mod wire;
use std::cell::{Cell, RefCell};
@@ -76,7 +77,9 @@ use crate::http::state::{HttpInner, HttpState, insert_view_header};
use crate::server_error::ServerNgError;
/// Bind the shard-0 HTTP listener and spawn the `cyper-axum` serve loop as a
-/// background task on shard 0's compio runtime.
+/// background task on shard 0's compio runtime. Serves HTTPS when
+/// `http.tls.enabled` (a TLS accept pump feeds handshaken streams to the
+/// serve loop, see [`mod@tls`]), plain HTTP otherwise.
///
/// The caller gates this to shard 0 and to `http.enabled`; the listener stops
/// when the bus shutdown token fires.
@@ -84,8 +87,8 @@ use crate::server_error::ServerNgError;
/// # Errors
///
/// Returns [`ServerNgError`] if the JWT manager cannot be built from
-/// `http_config.jwt`, the `[http.cors]` config is invalid, or the listener
-/// cannot bind to `addr`.
+/// `http_config.jwt`, the `[http.cors]` config is invalid, the `[http.tls]`
+/// credentials cannot be loaded, or the listener cannot bind to `addr`.
pub async fn start(
shard: &Rc<ServerNgShard>,
addr: SocketAddr,
@@ -103,7 +106,6 @@ pub async fn start(
.then(|| configure_cors(&http_config.cors))
.transpose()?;
let (listener, bound_addr) = client_listener::tcp::bind(addr).await?;
- info!(address = %bound_addr, "server-ng HTTP listener started");
let state: HttpState = SendWrapper::new(Rc::new(HttpInner {
shard: Rc::clone(shard),
@@ -135,16 +137,27 @@ pub async fn start(
usize::try_from(http_config.max_request_size.as_bytes_u64()).unwrap_or(usize::MAX);
let router = router(state, max_request_size, cors, http_config.web_ui);
- let shutdown = shard.bus.token();
- let handle = compio::runtime::spawn(async move {
- if let Err(error) = cyper_axum::serve(listener, router)
- .with_graceful_shutdown(async move { shutdown.wait().await })
- .await
- {
- error!(%error, "server-ng HTTP listener terminated with error");
- }
- });
- shard.bus.track_background(handle);
+ if http_config.tls.enabled {
+ let server_config = tls::load_http_tls_server_config(&http_config.tls)?;
+ let (connections, pump) =
+ tls::spawn_accept_pump(listener, server_config, shard.bus.token());
+ shard.bus.track_background(pump);
+ info!(address = %bound_addr, "server-ng HTTPS listener started");
+ let handle = compio::runtime::spawn(tls::serve(connections, router, shard.bus.token()));
+ shard.bus.track_background(handle);
+ } else {
+ info!(address = %bound_addr, "server-ng HTTP listener started");
+ let shutdown = shard.bus.token();
+ let handle = compio::runtime::spawn(async move {
+ if let Err(error) = cyper_axum::serve(listener, router)
+ .with_graceful_shutdown(async move { shutdown.wait().await })
+ .await
+ {
+ error!(%error, "server-ng HTTP listener terminated with error");
+ }
+ });
+ shard.bus.track_background(handle);
+ }
Ok(())
}
diff --git a/core/server-ng/src/http/tls.rs b/core/server-ng/src/http/tls.rs
new file mode 100644
index 000000000..dd974c897
--- /dev/null
+++ b/core/server-ng/src/http/tls.rs
@@ -0,0 +1,266 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! HTTPS for the shard-0 REST listener.
+//!
+//! server-ng is thread-per-core `compio/io_uring`, so it cannot reuse the
+//! legacy `axum-server` TLS acceptor. The plain-HTTP path uses
+//! `cyper_axum::serve`; the TLS path cannot, because that serve loop wraps
+//! its IO in a `compio` `Split` (one `BiLock`): hyper parks a pending read
+//! holding the lock, so the response write blocks on the same lock forever
+//! and 0 bytes are returned. Instead this module drives each connection on a
+//! hand-rolled hyper loop over an UNSPLIT [`TlsStream`] via
+//! [`cyper_core::HyperStream::new_tls`], which hyper's single-task loop polls
+//! read-then-write on one `&mut` - no split, no deadlock.
+//!
+//! A separate accept pump owns the raw TCP listener and the `compio-tls`
+//! acceptor: it accepts plaintext sockets and spawns one handshake task per
+//! connection, so a slow peer cannot block the accept loop (the
+//! cross-transport invariant in `message_bus::client_listener`). Handshaken
+//! streams flow to the serve loop over a bounded channel.
+
+use std::future::Future;
+use std::net::SocketAddr;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_channel::{Receiver, Sender};
+use axum::Router;
+use compio::net::{TcpListener, TcpStream};
+use compio::runtime::JoinHandle;
+use compio::tls::{TlsAcceptor, TlsStream};
+use configs::http::HttpTlsConfig;
+use cyper_core::HyperStream;
+use futures::{FutureExt, pin_mut};
+use hyper::rt::Executor;
+use hyper_util::server::conn::auto::Builder;
+use hyper_util::service::TowerToHyperService;
+use message_bus::ShutdownToken;
+use message_bus::transports::tls::{
+ TlsServerCredentials, install_default_crypto_provider, load_pem,
+};
+use tracing::{debug, error};
+
+use crate::server_error::ServerNgError;
+
+/// hyper's auto-builder serves whichever protocol the client selects via
+/// ALPN; advertise the same pair `axum-server` negotiates by default on the
+/// legacy HTTP server (there inherited from its default, hand-rolled here):
+/// HTTP/2 preferred, HTTP/1.1 fallback.
+const ALPN_H2: &[u8] = b"h2";
+const ALPN_HTTP11: &[u8] = b"http/1.1";
+
+/// Handshake wall-clock bound for the accept pump. The HTTP listener has no
+/// `MessageBusConfig` to source it from, so it mirrors the binary
+/// transports' `DEFAULT_HANDSHAKE_GRACE` directly.
+const HTTP_TLS_HANDSHAKE_GRACE: Duration = Duration::from_secs(10);
+
+/// Depth of the handshaken-connection channel between the accept pump and
+/// the serve loop. The serve loop drains one per accept and immediately
+/// spawns a per-connection hyper task, so it rarely fills; the bound keeps
+/// a handshake burst from queuing without limit.
+const ACCEPT_CHANNEL_DEPTH: usize = 256;
+
+/// A handshaken TLS stream plus the peer it came from, carried from the
+/// accept pump to the serve loop.
+type Handshaken = (TlsStream<TcpStream>, SocketAddr);
+
+/// Load the HTTPS `ServerConfig` from the operator's PEM cert + key.
+///
+/// # Errors
+///
+/// [`ServerNgError::ListenerCredentials`] with `transport: "http.tls"` if
+/// the PEM files cannot be read or the certificate / key pair is rejected.
+pub fn load_http_tls_server_config(
+ tls: &HttpTlsConfig,
+) -> Result<Arc<rustls::ServerConfig>, ServerNgError> {
+ let credentials =
+ load_pem(Path::new(&tls.cert_file), Path::new(&tls.key_file)).map_err(|source| {
+ ServerNgError::ListenerCredentials {
+ transport: "http.tls",
+ source,
+ }
+ })?;
+ build_server_config(credentials)
+}
+
+/// Bind the accept pump: build the acceptor, spawn the pump task, and hand
+/// back the handshaken-connection receiver for [`serve`] plus the pump's
+/// join handle for the caller to track.
+pub fn spawn_accept_pump(
+ listener: TcpListener,
+ config: Arc<rustls::ServerConfig>,
+ shutdown: ShutdownToken,
+) -> (Receiver<Handshaken>, JoinHandle<()>) {
+ let (connections_tx, connections_rx) =
+ async_channel::bounded::<Handshaken>(ACCEPT_CHANNEL_DEPTH);
+ let acceptor = TlsAcceptor::from(config);
+ let pump = compio::runtime::spawn(accept_pump(listener, acceptor, connections_tx, shutdown));
+ (connections_rx, pump)
+}
+
+/// Drive the HTTPS serve loop until the accept pump closes the channel at
+/// shutdown. Each handshaken stream gets its own detached hyper task; on
+/// shutdown the pump drops its sender, this loop ends, and the in-flight
+/// connection tasks drain via their own shutdown clone.
+pub async fn serve(connections: Receiver<Handshaken>, router: Router, shutdown: ShutdownToken) {
+ while let Ok((tls, peer)) = connections.recv().await {
+ let router = router.clone();
+ let shutdown = shutdown.clone();
+ compio::runtime::spawn(serve_connection(tls, peer, router, shutdown)).detach();
+ }
+}
+
+/// Serve one HTTPS connection: wrap the unsplit [`TlsStream`] in a duplex-safe
+/// [`HyperStream`] and run hyper's auto (h1/h2) connection to completion,
+/// switching to a graceful shutdown when the bus shutdown fires.
+async fn serve_connection(
+ tls: TlsStream<TcpStream>,
+ peer: SocketAddr,
+ router: Router,
+ shutdown: ShutdownToken,
+) {
+ let io = HyperStream::new_tls(tls);
+ // `Router<()>` already maps the incoming body to axum's `Body` in its own
+ // `Service` impl, so it serves hyper's `Request<Incoming>` directly - no
+ // `map_request` shim. `with_state(())` finalizes the routes eagerly.
+ let service = TowerToHyperService::new(router.with_state(()));
+ let builder = Builder::new(LocalExecutor);
+ // `serve_connection_with_upgrades` borrows `builder`, so it must outlive
+ // `conn`; keep it bound rather than inlined.
+ let conn = builder.serve_connection_with_upgrades(Box::pin(io), service);
+ pin_mut!(conn);
+
+ // Fire the graceful-shutdown transition exactly once: a fresh
+ // `shutdown.wait()` per loop turn would re-resolve immediately (recv on a
+ // closed channel), so pin one fused future and poll it by `&mut`.
+ let shutdown = shutdown.wait().fuse();
+ pin_mut!(shutdown);
+ loop {
+ futures::select! {
+ result = conn.as_mut().fuse() => {
+ if let Err(error) = result {
+ debug!(%peer, %error, "server-ng HTTPS connection terminated with error");
+ }
+ break;
+ }
+ () = &mut shutdown => conn.as_mut().graceful_shutdown(),
+ }
+ }
+}
+
+/// Single-thread executor for hyper's per-connection tasks. Unlike
+/// `cyper_core::CompioExecutor` it puts no `Send` bound on the spawned
+/// future, so the shard-local (non-`Send`) axum service future drives
+/// directly without a `SendWrapper` shim. Spawns onto the current shard's
+/// compio runtime.
+#[derive(Clone)]
+struct LocalExecutor;
+
+impl<F: Future<Output = ()> + 'static> Executor<F> for LocalExecutor {
+ fn execute(&self, fut: F) {
+ compio::runtime::spawn(fut).detach();
+ }
+}
+
+/// Turn credentials into an ALPN-configured `ServerConfig`. Mirrors the
+/// binary TCP-TLS path (`with_no_client_auth`, `max_early_data_size = 0`);
+/// the only HTTP-specific step is the ALPN advertisement.
+fn build_server_config(
+ credentials: TlsServerCredentials,
+) -> Result<Arc<rustls::ServerConfig>, ServerNgError> {
+ install_default_crypto_provider();
+ let mut config = rustls::ServerConfig::builder()
+ .with_no_client_auth()
+ .with_single_cert(credentials.cert_chain, credentials.key_der)
+ .map_err(|error| ServerNgError::ListenerCredentials {
+ transport: "http.tls",
+ source: std::io::Error::other(format!(
+ "http TLS server config rejected credentials: {error}"
+ )),
+ })?;
+ config.max_early_data_size = 0;
+ config.alpn_protocols = vec![ALPN_H2.to_vec(), ALPN_HTTP11.to_vec()];
+ Ok(Arc::new(config))
+}
+
+/// Accept plaintext sockets and dispatch each to its own handshake task
+/// until shutdown. On shutdown the sole sender drops, closing the channel.
+async fn accept_pump(
+ listener: TcpListener,
+ acceptor: TlsAcceptor,
+ connections: Sender<Handshaken>,
+ shutdown: ShutdownToken,
+) {
+ loop {
+ futures::select! {
+ () = shutdown.wait().fuse() => {
+ debug!("server-ng HTTPS accept pump shutting down");
+ break;
+ }
+ result = listener.accept().fuse() => match result {
+ Ok((stream, peer)) => spawn_handshake(&acceptor, &connections, stream, peer),
+ Err(error) => error!(%error, "server-ng HTTPS accept failed"),
+ },
+ }
+ }
+}
+
+/// Run one connection's TLS handshake in a detached, time-bounded task and,
+/// on success, forward the stream to the serve loop. A handshake that times
+/// out or errors is logged at debug and dropped; it never blocks the accept
+/// loop.
+fn spawn_handshake(
+ acceptor: &TlsAcceptor,
+ connections: &Sender<Handshaken>,
+ stream: TcpStream,
+ peer: SocketAddr,
+) {
+ let acceptor = acceptor.clone();
+ let connections = connections.clone();
+ compio::runtime::spawn(async move {
+ match compio::time::timeout(HTTP_TLS_HANDSHAKE_GRACE, acceptor.accept(stream)).await {
+ Ok(Ok(tls)) => {
+ // Drop on send error: the channel is closed only at
+ // shutdown, when the serve loop is already tearing down.
+ let _ = connections.send((tls, peer)).await;
+ }
+ Ok(Err(error)) => debug!(%peer, %error, "server-ng HTTPS handshake failed"),
+ Err(_elapsed) => {
+ debug!(%peer, grace = ?HTTP_TLS_HANDSHAKE_GRACE, "server-ng HTTPS handshake timed out");
+ }
+ }
+ })
+ .detach();
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use message_bus::transports::tls::self_signed_for_loopback;
+
+ #[test]
+ fn server_config_advertises_h2_and_http11_alpn() {
+ let config =
+ build_server_config(self_signed_for_loopback()).expect("self-signed cert builds");
+ assert_eq!(
+ config.alpn_protocols,
+ vec![b"h2".to_vec(), b"http/1.1".to_vec()]
+ );
+ }
+}