You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by hg...@apache.org on 2026/07/16 10:31:06 UTC
(iggy) branch feat/server-ng-http-tls updated (47b49a215 -> 4bf4f5ac0)
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a change to branch feat/server-ng-http-tls
in repository https://gitbox.apache.org/repos/asf/iggy.git
omit 47b49a215 merge master
omit 4f84bccaa feat(server-ng): serve HTTPS on the REST listener
add e98476210 feat(simulator): deterministic future executor and server-ng dispatch shell (#3671)
new 4bf4f5ac0 feat(server-ng): serve HTTPS on the REST listener
This update added new revisions after undoing existing revisions.
That is to say, some revisions that were in the old version of the
branch are not in the new version. This situation occurs
when a user --force pushes a change and generates a repository
containing something like this:
* -- * -- B -- O -- O -- O (47b49a215)
\
N -- N -- N refs/heads/feat/server-ng-http-tls (4bf4f5ac0)
You should already have received notification emails for all of the O
revisions, and so the following emails describe only the N revisions
from the common base, B.
Any revisions marked "omit" are not gone; other references still
refer to them. Any revisions marked "discard" are gone forever.
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
Cargo.lock | 5 +
core/configs/src/server_ng_config/validators.rs | 46 +-
core/consensus/Cargo.toml | 1 +
core/consensus/src/impls.rs | 234 ++++-
core/message_bus/src/installer/mod.rs | 19 +-
core/message_bus/src/lib.rs | 95 ++
core/metadata/Cargo.toml | 9 +
core/metadata/src/impls/metadata.rs | 27 +-
core/metadata/src/stm/mux.rs | 6 +-
core/metadata/src/stm/snapshot.rs | 17 +-
core/metadata/src/stm/stream.rs | 54 +
core/partitions/Cargo.toml | 9 +
core/partitions/src/iggy_partition.rs | 1 +
core/partitions/src/iggy_partitions.rs | 43 +-
core/server-ng/src/auth.rs | 96 +-
core/server-ng/src/bootstrap.rs | 148 ++-
core/server-ng/src/consumer_group.rs | 52 +-
core/server-ng/src/dispatch.rs | 503 ++++++---
core/server-ng/src/dispatch/authz.rs | 133 ++-
core/server-ng/src/partition_reconciler.rs | 5 +-
core/server-ng/src/responses.rs | 222 +++-
core/server-ng/src/users.rs | 17 +-
core/server_common/Cargo.toml | 8 +
core/server_common/src/crypto.rs | 39 +-
core/server_common/src/sharding/namespace.rs | 2 +-
core/shard/src/lib.rs | 136 ++-
core/shard/src/metrics.rs | 16 +
core/shard/src/router.rs | 33 +-
core/simulator/Cargo.toml | 12 +-
core/simulator/src/bus.rs | 256 ++++-
core/simulator/src/client.rs | 200 +++-
core/simulator/src/deps.rs | 86 +-
core/simulator/src/executor/mod.rs | 802 ++++++++++++++
core/simulator/src/executor/time.rs | 282 +++++
core/simulator/src/lib.rs | 1281 +++++++++++++++++++++--
core/simulator/src/replica.rs | 233 ++++-
core/simulator/src/workload/oracle.rs | 4 +-
37 files changed, 4549 insertions(+), 583 deletions(-)
create mode 100644 core/simulator/src/executor/mod.rs
create mode 100644 core/simulator/src/executor/time.rs
(iggy) 01/01: feat(server-ng): serve HTTPS on the REST listener
Posted by hg...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
hubcio pushed a commit to branch feat/server-ng-http-tls
in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 4bf4f5ac020842534e8409dd9cde4dce1a87ba7e
Author: Hubert Gruszecki <h....@gmail.com>
AuthorDate: Tue Jul 14 14:50:30 2026 +0200
feat(server-ng): serve HTTPS on the REST listener
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()]
+ );
+ }
+}