SQL-522: serve HTTP/2 on environmentd and balancerd HTTP endpoints

Cargo.toml

@@ -308,7 +308,9 @@ aws-smithy-runtime = { version = "1.9.8", features = ["connector-hyper-0-14-x"]
 aws-smithy-runtime-api = "1.10.0"
 aws-smithy-types = { version = "1.1.8", features = ["byte-stream-poll-next"] }
 aws-types = "1.3.9"
-axum = { version = "0.8.8", features = ["ws"] }
+# 0.8.9 minimum: 0.8.8 has a method-routing bug that registers CONNECT
+# handlers (needed for WebSockets over HTTP/2) under OPTIONS.
+axum = { version = "0.8.9", features = ["http2", "ws"] }
 axum-extra = { version = "0.12.5", features = ["typed-header"] }
 axum-server = { version = "0.8.0", features = ["tls-rustls"] }
 azure_core = "0.21.0"
@@ -386,12 +388,12 @@ http = "1.4.0"
 http-body-util = "0.1.3"
 httparse = "1.8.0"
 humantime = "2.3.0"
-hyper = { version = "1.9.0", features = ["http1", "server"] }
+hyper = { version = "1.9.0", features = ["client", "http1", "http2", "server"] }
 # hyper 0.14 is used by AWS SDK. Used to override the DNS resolver used by the SDK,
 # which can only be done by constructing the HTTP Client
 hyper-0-14 = { package = "hyper", version = "0.14", features = ["client", "tcp"] }
 hyper-openssl = "0.10.2"
-hyper-util = "0.1.20"
+hyper-util = { version = "0.1.20", features = ["server-auto", "tokio"] }
 tower-service = "0.3.3"
 iceberg = "0.9.0"
 iceberg-catalog-rest = "0.9.0"
@@ -482,7 +484,9 @@ rdkafka = { version = "0.29.0", features = ["cmake-build", "libz-static", "ssl-v
 rdkafka-sys = { version = "4.3.0", features = ["cmake-build", "libz-static", "ssl-vendored", "zstd"] }
 regex = "1.12.3"
 regex-syntax = "0.8.10"
-reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-vendored", "stream"] }
+# `native-tls-alpn` is required for reqwest to negotiate HTTP/2 via ALPN with
+# the native-tls backend; without it clients silently stay on HTTP/1.1.
+reqwest = { version = "0.12.28", features = ["blocking", "charset", "cookies", "default-tls", "http2", "json", "native-tls-alpn", "native-tls-vendored", "stream"] }
 reqwest-middleware = { version = "0.4.2", features = ["json"] }
 reqwest-retry = "0.8.0"
 rlimit = "0.11.0"
@@ -543,6 +547,7 @@ tokio-openssl = "0.6.5"
 tokio-postgres = "0.7.15"
 tokio-stream = "0.1.18"
 tokio-test = "0.4.5"
+tokio-tungstenite = "0.29.0"
 tokio-util = "0.7.18"
 toml = "0.8.22"
 toml_edit = { version = "0.22.26", features = ["serde"] }
@@ -559,7 +564,7 @@ tracing-capture = "0.1.0"
 tracing-core = "0.1.35"
 tracing-opentelemetry = "0.32.1"
 tracing-subscriber = { version = "0.3.23", features = ["env-filter", "fmt", "json", "tracing-log"] }
-tungstenite = "0.28.0"
+tungstenite = "0.29.0"
 turmoil = "0.7.2"
 uncased = "0.9.10"
 unicode-normalization = "0.1.25"

deny.toml

@@ -159,6 +159,10 @@ skip = [
     { name = "rand_core", version = "0.10.1" },
     { name = "getrandom", version = "0.4.2" },
     { name = "cpufeatures", version = "0.3.0" },
+    # Held back by kube-client; axum 0.8.9 uses tungstenite 0.29 for
+    # WebSockets.
+    { name = "tungstenite", version = "0.28.0" },
+    { name = "tokio-tungstenite", version = "0.28.0" },
 ]
 
 [[bans.deny]]

misc/python/materialize/mzcompose/__init__.py

@@ -665,6 +665,7 @@ def get_default_system_parameters(
     "balancerd_sigterm_connection_wait",
     "balancerd_sigterm_listen_wait",
     "balancerd_inject_proxy_protocol_header_http",
+    "balancerd_https_enable_http2_alpn",
     "balancerd_log_filter",
     "balancerd_opentelemetry_filter",
     "balancerd_log_filter_defaults",

misc/python/materialize/parallel_workload/action.py

@@ -2042,6 +2042,7 @@ def __init__(
             "balancerd_sigterm_connection_wait",
             "balancerd_sigterm_listen_wait",
             "balancerd_inject_proxy_protocol_header_http",
+            "balancerd_https_enable_http2_alpn",
             "balancerd_log_filter",
             "balancerd_opentelemetry_filter",
             "balancerd_log_filter_defaults",

src/balancerd/src/dyncfgs.rs

@@ -43,6 +43,26 @@ pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config<bool> = Config::new(
     "Whether to inject tcp proxy protocol headers to downstream http servers.",
 );
 
+/// Whether to advertise HTTP/2 via ALPN on the HTTPS listener.
+///
+/// balancerd is a byte proxy: it terminates TLS and forwards the decrypted
+/// stream to environmentd. ALPN is answered during the client handshake,
+/// before balancerd has connected upstream, so it cannot discover whether
+/// environmentd speaks HTTP/2 in time. Enabling this while environmentd is
+/// still HTTP/1.1-only makes clients negotiate h2 and send frames environmentd
+/// rejects with "invalid HTTP version parsed (found HTTP2 preface)". Enable
+/// only after every environmentd instance supports HTTP/2.
+///
+/// NOTE: read once when the TLS context is built at startup, so a change only
+/// takes effect after balancerd restarts.
+pub const HTTPS_ENABLE_HTTP2_ALPN: Config<bool> = Config::new(
+    "balancerd_https_enable_http2_alpn",
+    false,
+    "Whether to advertise HTTP/2 via ALPN on the HTTPS listener. \
+    Enable only after all environmentd instances support HTTP/2. \
+    Takes effect on balancerd restart.",
+);
+
 /// Sets the filter to apply to stderr logging.
 pub const LOGGING_FILTER: Config<&str> = Config::new(
     "balancerd_log_filter",
@@ -98,6 +118,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&SIGTERM_CONNECTION_WAIT)
         .add(&SIGTERM_LISTEN_WAIT)
         .add(&INJECT_PROXY_PROTOCOL_HEADER_HTTP)
+        .add(&HTTPS_ENABLE_HTTP2_ALPN)
         .add(&LOGGING_FILTER)
         .add(&OPENTELEMETRY_FILTER)
         .add(&LOGGING_FILTER_DEFAULTS)
@@ -124,6 +145,11 @@ pub(crate) fn set_defaults(
                 INJECT_PROXY_PROTOCOL_HEADER_HTTP.name(),
                 mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?),
             )
+        } else if k.as_str() == HTTPS_ENABLE_HTTP2_ALPN.name() {
+            config_updates.add_dynamic(
+                HTTPS_ENABLE_HTTP2_ALPN.name(),
+                mz_dyncfg::ConfigVal::Bool(bool::from_str(v)?),
+            )
         } else {
             return Err(anyhow!("Invalid default config value {k}"));
         }

src/balancerd/src/lib.rs

@@ -30,7 +30,6 @@ use anyhow::Context;
 use axum::response::IntoResponse;
 use axum::{Router, routing};
 use bytes::BytesMut;
-use futures::TryFutureExt;
 use futures::stream::BoxStream;
 use hickory_resolver::config::LookupIpStrategy;
 use hickory_resolver::lookup_ip::LookupIp;
@@ -39,7 +38,7 @@ use hickory_resolver::proto::rr::{RData, RecordType};
 use hickory_resolver::system_conf::read_system_conf;
 use hickory_resolver::{Resolver, TokioResolver};
 use hyper::StatusCode;
-use hyper_util::rt::TokioIo;
+use hyper_util::rt::{TokioExecutor, TokioIo};
 use launchdarkly_server_sdk as ld;
 use mz_build_info::{BuildInfo, build_info};
 use mz_dyncfg::ConfigSet;
@@ -77,8 +76,8 @@ use uuid::Uuid;
 
 use crate::codec::{BackendMessage, FramedConn};
 use crate::dyncfgs::{
-    INJECT_PROXY_PROTOCOL_HEADER_HTTP, SIGTERM_CONNECTION_WAIT, SIGTERM_LISTEN_WAIT,
-    has_tracing_config_update, tracing_config,
+    HTTPS_ENABLE_HTTP2_ALPN, INJECT_PROXY_PROTOCOL_HEADER_HTTP, SIGTERM_CONNECTION_WAIT,
+    SIGTERM_LISTEN_WAIT, has_tracing_config_update, tracing_config,
 };
 
 /// Balancer build information.
@@ -302,7 +301,12 @@ impl BalancerService {
     pub async fn serve(self) -> Result<(), anyhow::Error> {
         let (pgwire_tls, https_tls) = match &self.cfg.tls {
             Some(tls) => {
-                let context = tls.reloading_context(self.cfg.reload_certs)?;
+                // Controlled by dyncfg: only advertise HTTP/2 via ALPN when the
+                // upstream environmentd is known to support it. balancerd is a
+                // byte proxy, so if we advertise h2 before environmentd supports
+                // it, clients send h2 frames that environmentd cannot parse.
+                let enable_http2_alpn = HTTPS_ENABLE_HTTP2_ALPN.get(&self.configs);
+                let context = tls.reloading_context(self.cfg.reload_certs, enable_http2_alpn)?;
                 (
                     Some(ReloadingTlsConfig {
                         context: context.clone(),
@@ -472,8 +476,11 @@ impl mz_server_core::Server for InternalHttpServer {
         let conn = TokioIo::new(conn);
 
         Box::pin(async {
-            let http = hyper::server::conn::http1::Builder::new();
-            http.serve_connection(conn, service).err_into().await
+            // Serve HTTP/1.1 or HTTP/2 (h2c via preface sniffing).
+            let http = hyper_util::server::conn::auto::Builder::new(TokioExecutor::new());
+            http.serve_connection(conn, service)
+                .await
+                .map_err(|e| anyhow::anyhow!(e))
         })
     }
 }
@@ -1209,28 +1216,32 @@ impl mz_server_core::Server for HttpsBalancer {
             let active_guard = inner_metrics.active_connections();
             let result: Result<_, anyhow::Error> = Box::pin(async move {
                 let peer_addr = peer_addr.context("fetching peer addr")?;
-                let (mut client_stream, servername): (Box<dyn ClientStream>, Option<String>) =
-                    match tls_context {
-                        Some(tls_context) => {
-                            let mut ssl_stream =
-                                SslStream::new(Ssl::new(&tls_context.get())?, conn)?;
-                            if let Err(e) = Pin::new(&mut ssl_stream).accept().await {
-                                let _ = ssl_stream.get_mut().shutdown().await;
-                                return Err(e.into());
-                            }
-                            let servername: Option<String> =
-                                ssl_stream.ssl().servername(NameType::HOST_NAME).map(|sn| {
-                                    match sn.split_once('.') {
-                                        Some((left, _right)) => left,
-                                        None => sn,
-                                    }
-                                    .into()
-                                });
-                            debug!("Found sni servername: {servername:?} (https)");
-                            (Box::new(ssl_stream), servername)
+                let (mut client_stream, servername, client_h2): (
+                    Box<dyn ClientStream>,
+                    Option<String>,
+                    bool,
+                ) = match tls_context {
+                    Some(tls_context) => {
+                        let mut ssl_stream = SslStream::new(Ssl::new(&tls_context.get())?, conn)?;
+                        if let Err(e) = Pin::new(&mut ssl_stream).accept().await {
+                            let _ = ssl_stream.get_mut().shutdown().await;
+                            return Err(e.into());
                         }
-                        _ => (Box::new(conn), None),
-                    };
+                        let servername: Option<String> =
+                            ssl_stream.ssl().servername(NameType::HOST_NAME).map(|sn| {
+                                match sn.split_once('.') {
+                                    Some((left, _right)) => left,
+                                    None => sn,
+                                }
+                                .into()
+                            });
+                        debug!("Found sni servername: {servername:?} (https)");
+                        let client_h2 =
+                            ssl_stream.ssl().selected_alpn_protocol() == Some(b"h2".as_slice());
+                        (Box::new(ssl_stream), servername, client_h2)
+                    }
+                    _ => (Box::new(conn), None, false),
+                };
                 let resolved =
                     Self::resolve(&resolver, &resolve_template, port, servername.as_deref())
                         .await?;
@@ -1242,23 +1253,28 @@ impl mz_server_core::Server for HttpsBalancer {
                     Ok(stream) => stream,
                     Err(e) => {
                         error!("failed to connect to upstream server: {e}");
-                        let body = "upstream server not available";
                         // We know this is an HTTPs stream (see name
                         // HttpsBalancer), but we actually don't care what type
                         // of traffic it is and we only use raw tcp streams.In
                         // order to respond with HTTP we have to write this as a
-                        // raw http message.
-                        let response = format!(
-                            "HTTP/1.1 502 Bad Gateway\r\n\
-                             Content-Type: text/plain\r\n\
-                             Content-Length: {}\r\n\
-                             Connection: close\r\n\
-                             \r\n\
-                             {}",
-                            body.len(),
-                            body
-                        );
-                        let _ = client_stream.write_all(response.as_bytes()).await;
+                        // raw http message. This raw message is only
+                        // intelligible to HTTP/1 clients, though: clients that
+                        // negotiated HTTP/2 via ALPN just get a closed
+                        // connection.
+                        if !client_h2 {
+                            let body = "upstream server not available";
+                            let response = format!(
+                                "HTTP/1.1 502 Bad Gateway\r\n\
+                                 Content-Type: text/plain\r\n\
+                                 Content-Length: {}\r\n\
+                                 Connection: close\r\n\
+                                 \r\n\
+                                 {}",
+                                body.len(),
+                                body
+                            );
+                            let _ = client_stream.write_all(response.as_bytes()).await;
+                        }
                         let _ = client_stream.shutdown().await;
                         return Ok(());
                     }