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

Cargo.toml

@@ -299,7 +299,7 @@ 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.9", features = ["ws"] }
+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"
@@ -377,12 +377,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.10.1"
 iceberg-catalog-rest = "0.10.1"
@@ -536,6 +536,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"] }

misc/python/materialize/mzcompose/__init__.py

@@ -749,6 +749,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

@@ -3312,6 +3312,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

@@ -46,6 +46,27 @@ pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config<bool> = Config::new(
     ParameterScope::Environment,
 );
 
+/// 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.",
+    ParameterScope::Environment,
+);
+
 /// Sets the filter to apply to stderr logging.
 pub const LOGGING_FILTER: Config<&str> = Config::new(
     "balancerd_log_filter",
@@ -106,6 +127,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)
@@ -132,6 +154,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(());
                     }

src/balancerd/tests/server.rs

@@ -39,7 +39,7 @@ use mz_ore::retry::Retry;
 use mz_ore::tracing::TracingHandle;
 use mz_ore::{assert_contains, assert_err, assert_ok, task};
 use mz_server_core::TlsCertConfig;
-use openssl::ssl::{SslConnectorBuilder, SslVerifyMode};
+use openssl::ssl::{SslConnector, SslConnectorBuilder, SslMethod, SslVerifyMode};
 use openssl::x509::X509;
 use tokio::sync::oneshot;
 use uuid::Uuid;
@@ -201,7 +201,13 @@ async fn test_balancer() {
             None,
             None,
             TracingHandle::disabled(),
-            vec![],
+            // Advertise HTTP/2 via ALPN. This defaults off in production so a
+            // balancerd that upgrades ahead of environmentd does not offer h2
+            // to clients before environmentd can parse it.
+            vec![(
+                "balancerd_https_enable_http2_alpn".to_string(),
+                "true".to_string(),
+            )],
         );
         let balancer_server = BalancerService::new(balancer_cfg).await.unwrap();
         let balancer_pgwire_listen = balancer_server.pgwire.0.local_addr();
@@ -288,6 +294,43 @@ async fn test_balancer() {
         let resp_x509 = X509::from_der(tlsinfo.peer_certificate().unwrap()).unwrap();
         let server_x509 = X509::from_pem(&std::fs::read(&server_cert).unwrap()).unwrap();
         assert_eq!(resp_x509, server_x509);
+        assert_eq!(resp.version(), reqwest::Version::HTTP_11);
+        assert_contains!(resp.text().await.unwrap(), "12234");
+
+        // With `balancerd_https_enable_http2_alpn` set, balancerd offers h2 to
+        // clients that ask for it. reqwest's native-tls backend does not, hence
+        // the HTTP/1.1 responses either side of this.
+        assert_eq!(
+            alpn_selected(balancer_https_listen, b"\x02h2\x08http/1.1")
+                .await
+                .as_deref(),
+            Some(&b"h2"[..])
+        );
+        assert_eq!(
+            alpn_selected(balancer_https_listen, b"\x08http/1.1")
+                .await
+                .as_deref(),
+            Some(&b"http/1.1"[..])
+        );
+
+        // HTTP/1.1-only clients are still served.
+        let http1_client = reqwest::Client::builder()
+            .add_root_certificate(
+                reqwest::Certificate::from_pem(&ca.cert.to_pem().unwrap()).unwrap(),
+            )
+            .pool_max_idle_per_host(0)
+            .http1_only()
+            .build()
+            .unwrap();
+        let resp = http1_client
+            .post(&https_url)
+            .header("Content-Type", "application/json")
+            .basic_auth(frontegg_user, Some(&frontegg_password))
+            .body(body)
+            .send()
+            .await
+            .unwrap();
+        assert_eq!(resp.version(), reqwest::Version::HTTP_11);
         assert_contains!(resp.text().await.unwrap(), "12234");
 
         // Generate new certs. Install only the key, reload, and make sure the old cert is still in
@@ -406,5 +449,39 @@ async fn test_balancer() {
             })
             .await
             .unwrap();
+
+        // The internal HTTP server serves h2c (HTTP/2 with prior knowledge)
+        // alongside HTTP/1.1.
+        let h2c_client = reqwest::Client::builder()
+            .http2_prior_knowledge()
+            .build()
+            .unwrap();
+        let resp = h2c_client.get(&metrics_url).send().await.unwrap();
+        assert_eq!(resp.version(), reqwest::Version::HTTP_2);
+        assert!(resp.status().is_success());
     }
 }
+
+/// Returns the protocol the TLS server at `addr` selects for a client offering
+/// `alpn`, in OpenSSL wire format (length-prefixed protocol names).
+async fn alpn_selected(addr: SocketAddr, alpn: &'static [u8]) -> Option<Vec<u8>> {
+    // The handshake is blocking, and the server shares this runtime.
+    mz_ore::task::spawn_blocking(
+        || "alpn_probe",
+        move || {
+            let mut connector = SslConnector::builder(SslMethod::tls()).unwrap();
+            connector.set_verify(SslVerifyMode::NONE);
+            connector.set_alpn_protos(alpn).unwrap();
+            let stream = connector
+                .build()
+                .configure()
+                .unwrap()
+                .verify_hostname(false)
+                .use_server_name_indication(false)
+                .connect("", std::net::TcpStream::connect(addr).unwrap())
+                .unwrap();
+            stream.ssl().selected_alpn_protocol().map(<[u8]>::to_vec)
+        },
+    )
+    .await
+}

src/environmentd/Cargo.toml

@@ -145,6 +145,7 @@ serde_urlencoded.workspace = true
 similar-asserts.workspace = true
 timely.workspace = true
 tokio-postgres = { workspace = true, features = ["with-chrono-0_4", "with-serde_json-1"] }
+tokio-tungstenite.workspace = true
 uuid.workspace = true
 
 [build-dependencies]