From da5a8c4269c96631e2f19e3bf13317ee96af23f2 Mon Sep 17 00:00:00 2001 From: Jason Hernandez <7144515+jasonhernandez@users.noreply.github.com> Date: Mon, 6 Jul 2026 08:47:28 -0700 Subject: [PATCH 1/3] adapter: migrate LaunchDarkly SDK off fork to upstream 3.1.1 Move launchdarkly-server-sdk from the MaterializeInc/rust-server-sdk fork back to upstream crates.io 3.1.1, restoring the launchdarkly-sdk-transport + MetricsTransport setup and dropping the [patch.crates-io] override. The fork existed for launchdarkly/rust-server-sdk#116: a StreamingDataSource /eventsource StreamClosed bug where a non-Eof stream error left the data source stuck with no reconnect, silently breaking LD sync. A prior upgrade to upstream 3.0.1 had to be reverted (incident-984) because that bug was still unfixed upstream. The fixes have since landed, rust-server-sdk#168 and rust-eventsource-client#134/#135, and 3.1.1 resolves eventsource-client to 0.17.5, which carries them. Use the rustls + aws-lc-rs features (hyper-rustls-native-roots, crypto-aws-lc-rs), now the upstream defaults, instead of the prior attempt's native-tls/crypto-openssl, avoiding the OpenSSL path. The transport build_https() call is identical either way. Because the SDK now builds a rustls client, the workspace links both rustls provider features (aws_lc_rs via our features, ring transitively via the hyper-rustls chain). With both enabled rustls cannot select a process-default provider on its own and panics on first client build. Install aws-lc-rs explicitly (idempotently) at the top of the environmentd, clusterd, balancerd, sqllogictest, and testdrive entrypoints. orchestratord already installs it. deny.toml gains skips for the duplicate versions the transport stack pulls (older tower/rustls-native-certs; newer rand/rand_core/getrandom/cpufeatures) and re-adds the launchdarkly-sdk-transport wrapper. Adds MetricsTransport unit tests, including test_metric_frozen_on_midstream_ error, modeling the exact incident-984 failure mode (200 OK then a mid-stream timeout): they assert the last_sse_time_seconds gauge freezes so the staleness alert can detect a stuck data source. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 202 ++++++++++---- Cargo.toml | 6 +- deny.toml | 10 + src/adapter/Cargo.toml | 2 +- src/adapter/src/config/frontend.rs | 315 +++++++++++++++++++--- src/balancerd/Cargo.toml | 1 + src/balancerd/src/bin/balancerd.rs | 8 + src/clusterd/Cargo.toml | 1 + src/clusterd/src/lib.rs | 8 + src/dyncfg-launchdarkly/Cargo.toml | 2 +- src/dyncfg-launchdarkly/src/lib.rs | 21 +- src/environmentd/Cargo.toml | 1 + src/environmentd/src/environmentd/main.rs | 9 + src/sqllogictest/Cargo.toml | 1 + src/sqllogictest/src/bin/sqllogictest.rs | 8 + src/testdrive/Cargo.toml | 1 + src/testdrive/src/bin/testdrive.rs | 8 + 17 files changed, 506 insertions(+), 98 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 63ab9d3ca9797..c8cb45094f58b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,7 +31,7 @@ checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", "cipher", - "cpufeatures", + "cpufeatures 0.2.17", ] [[package]] @@ -1939,6 +1939,17 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" +[[package]] +name = "chacha20" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" +dependencies = [ + "cfg-if", + "cpufeatures 0.3.0", + "rand_core 0.10.1", +] + [[package]] name = "chrono" version = "0.4.41" @@ -2006,6 +2017,12 @@ dependencies = [ "half 1.6.0", ] +[[package]] +name = "cidr" +version = "0.3.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "579504560394e388085d0c080ea587dfa5c15f7e251b4d5247d1e1a61d1d6928" + [[package]] name = "cipher" version = "0.4.4" @@ -2349,6 +2366,15 @@ dependencies = [ "libc", ] +[[package]] +name = "cpufeatures" +version = "0.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" +dependencies = [ + "libc", +] + [[package]] name = "crc" version = "3.3.0" @@ -3028,7 +3054,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -3442,17 +3468,18 @@ dependencies = [ [[package]] name = "eventsource-client" -version = "0.16.0" +version = "0.17.5" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c26451361cde19fe2322835b6e684f4825902edf3139ce9d6a70e6542566a0b2" +checksum = "96df8cfa11d3c8e4e1a48b81c9c600ecbe8c11d1326f41e1524cc4d42f7bf6d9" dependencies = [ "base64 0.22.1", + "bytes", "futures", - "hyper 0.14.32", - "hyper-timeout 0.4.1", + "http 1.4.2", + "launchdarkly-sdk-transport", "log", "pin-project", - "rand 0.8.5", + "rand 0.10.2", "tokio", ] @@ -3997,11 +4024,23 @@ dependencies = [ "cfg-if", "js-sys", "libc", - "r-efi", + "r-efi 5.3.0", "wasi 0.14.2+wasi-0.2.4", "wasm-bindgen", ] +[[package]] +name = "getrandom" +version = "0.4.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "300e883d756b2e4ec94e02791f39b04b522276138852cfc41d9fb7e904106099" +dependencies = [ + "cfg-if", + "libc", + "r-efi 6.0.0", + "rand_core 0.10.1", +] + [[package]] name = "gimli" version = "0.32.3" @@ -4424,6 +4463,25 @@ dependencies = [ "want", ] +[[package]] +name = "hyper-http-proxy" +version = "1.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "8021e0ae20c08eadc94d0bdafdeda66d4f0858541c146ae6e46b219bfe58497e" +dependencies = [ + "bytes", + "futures-util", + "headers", + "http 1.4.2", + "hyper 1.9.0", + "hyper-rustls", + "hyper-util", + "pin-project-lite", + "tokio", + "tokio-rustls", + "tower-service", +] + [[package]] name = "hyper-openssl" version = "0.10.2" @@ -4461,18 +4519,6 @@ dependencies = [ "tower-service", ] -[[package]] -name = "hyper-timeout" -version = "0.4.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bbb958482e8c7be4bc3cf272a766a2b0bf1a6755e7a6ae777f017a31d11b13b1" -dependencies = [ - "hyper 0.14.32", - "pin-project-lite", - "tokio", - "tokio-io-timeout", -] - [[package]] name = "hyper-timeout" version = "0.5.1" @@ -4532,7 +4578,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "system-configuration", "tokio", "tower-service", @@ -4751,7 +4797,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "4b0f83760fb341a774ed326568e19f5a863af4a952def8c39f9ab92fd95b88e5" dependencies = [ "equivalent", - "hashbrown 0.16.1", + "hashbrown 0.15.3", "serde", "serde_core", ] @@ -4845,7 +4891,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -4932,7 +4978,7 @@ dependencies = [ "portable-atomic", "portable-atomic-util", "serde_core", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -5128,7 +5174,7 @@ dependencies = [ "http-body-util", "hyper 1.9.0", "hyper-openssl", - "hyper-timeout 0.5.1", + "hyper-timeout", "hyper-util", "jiff", "jsonpath-rust", @@ -5209,20 +5255,43 @@ dependencies = [ "tracing", ] +[[package]] +name = "launchdarkly-sdk-transport" +version = "0.1.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "032d66802c461d7254ed1de887be6ca98ebbe126867d9a60c754baae81c4d10d" +dependencies = [ + "bytes", + "futures", + "http 1.4.2", + "http-body-util", + "hyper 1.9.0", + "hyper-http-proxy", + "hyper-rustls", + "hyper-timeout", + "hyper-util", + "log", + "no-proxy", + "tower 0.4.13", +] + [[package]] name = "launchdarkly-server-sdk" -version = "2.6.2" -source = "git+https://github.com/MaterializeInc/rust-server-sdk?rev=3e0a0b98b09a2970f292577a07e1c9382b65b5da#3e0a0b98b09a2970f292577a07e1c9382b65b5da" +version = "3.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "0440c21aaa0670948071d9cbd8142e9d057a4b40feb3e6293e0efc6de9879b69" dependencies = [ "aws-lc-rs", + "bitflags 2.11.0", + "bytes", "chrono", "crossbeam-channel", "data-encoding", "eventsource-client", "futures", - "hyper 0.14.32", + "http 1.4.2", + "launchdarkly-sdk-transport", "launchdarkly-server-sdk-evaluation", - "lazy_static", "log", "lru", "moka", @@ -5238,9 +5307,9 @@ dependencies = [ [[package]] name = "launchdarkly-server-sdk-evaluation" -version = "2.0.1" +version = "2.1.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63706c23ee67f699563e5c52c7542361eccc966cba0430b6a7862c0ecaee9432" +checksum = "4bc97a52681b9860197ad81a2c1a34d96bb647459bcdf067ac1fae42c9fe8c30" dependencies = [ "base16ct", "chrono", @@ -5843,10 +5912,10 @@ dependencies = [ "hex", "http 1.4.2", "humantime", - "hyper-tls 0.5.0", "imbl", "ipnet", "itertools 0.14.0", + "launchdarkly-sdk-transport", "launchdarkly-server-sdk", "maplit", "mz-adapter-types", @@ -6137,6 +6206,7 @@ dependencies = [ "prometheus", "proxy-header", "reqwest 0.12.28", + "rustls", "semver", "tempfile", "tokio", @@ -6438,6 +6508,7 @@ dependencies = [ "mz-txn-wal", "nix 0.31.3", "num_cpus", + "rustls", "serde", "tokio", "tower 0.5.3", @@ -6761,7 +6832,7 @@ version = "0.0.0" dependencies = [ "anyhow", "humantime", - "hyper-tls 0.5.0", + "launchdarkly-sdk-transport", "launchdarkly-server-sdk", "mz-build-info", "mz-dyncfg", @@ -6887,6 +6958,7 @@ dependencies = [ "regex", "reqwest 0.12.28", "rlimit", + "rustls", "semver", "sentry-tracing", "serde", @@ -8425,6 +8497,7 @@ dependencies = [ "postgres-protocol", "regex", "reqwest 0.12.28", + "rustls", "serde_json", "shell-words", "tempfile", @@ -8838,6 +8911,7 @@ dependencies = [ "rdkafka", "regex", "reqwest 0.12.28", + "rustls", "semver", "serde", "serde_json", @@ -9047,6 +9121,15 @@ dependencies = [ "libc", ] +[[package]] +name = "no-proxy" +version = "0.3.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7dd3f71e44f009513a82ac2612c0c2da7e2924fad776931164317baa9695e150" +dependencies = [ + "cidr", +] + [[package]] name = "nom" version = "7.1.3" @@ -10423,7 +10506,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042" dependencies = [ "heck", - "itertools 0.13.0", + "itertools 0.10.5", "log", "multimap", "petgraph", @@ -10444,7 +10527,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf" dependencies = [ "anyhow", - "itertools 0.13.0", + "itertools 0.10.5", "proc-macro2", "quote", "syn 2.0.117", @@ -10669,6 +10752,12 @@ version = "5.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" +[[package]] +name = "r-efi" +version = "6.0.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf" + [[package]] name = "radium" version = "0.7.0" @@ -10710,6 +10799,17 @@ dependencies = [ "rand_core 0.9.3", ] +[[package]] +name = "rand" +version = "0.10.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" +dependencies = [ + "chacha20", + "getrandom 0.4.3", + "rand_core 0.10.1", +] + [[package]] name = "rand_chacha" version = "0.2.2" @@ -10768,6 +10868,12 @@ dependencies = [ "getrandom 0.3.3", ] +[[package]] +name = "rand_core" +version = "0.10.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" + [[package]] name = "rand_distr" version = "0.5.1" @@ -11373,7 +11479,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -11385,6 +11491,7 @@ dependencies = [ "aws-lc-rs", "log", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -11951,7 +12058,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", ] @@ -11968,7 +12075,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures", + "cpufeatures 0.2.17", "digest", "sha2-asm", ] @@ -12492,10 +12599,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd" dependencies = [ "fastrand 2.3.0", - "getrandom 0.3.3", + "getrandom 0.4.3", "once_cell", "rustix 1.1.4", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -12514,7 +12621,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "230a1b821ccbd75b185820a1f1ff7b14d21da1e442e22c0863ea5f08771a8874" dependencies = [ "rustix 1.1.4", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -12791,16 +12898,6 @@ dependencies = [ "windows-sys 0.61.1", ] -[[package]] -name = "tokio-io-timeout" -version = "1.1.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "90c49f106be240de154571dd31fbe48acb10ba6c6dd6f6517ad603abffa42de9" -dependencies = [ - "pin-project-lite", - "tokio", -] - [[package]] name = "tokio-io-utility" version = "0.7.6" @@ -13008,7 +13105,7 @@ dependencies = [ "http-body 1.0.1", "http-body-util", "hyper 1.9.0", - "hyper-timeout 0.5.1", + "hyper-timeout", "hyper-util", "percent-encoding", "pin-project", @@ -13073,6 +13170,7 @@ dependencies = [ "pin-project-lite", "tower-layer", "tower-service", + "tracing", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 04ccfa3b98d27..11a53e97957b3 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -407,7 +407,8 @@ junit-report = "0.8.3" k8s-controller = "0.11.0" k8s-openapi = { version = "0.27.0", features = ["schemars", "v1_32"] } kube = { version = "3.1.0", default-features = false, features = ["client", "derive", "openssl-tls", "runtime", "ws"] } -launchdarkly-server-sdk = { version = "2.6.2", default-features = false } +launchdarkly-server-sdk = { version = "3.1.1", default-features = false, features = ["hyper-rustls-native-roots", "crypto-aws-lc-rs"] } +launchdarkly-sdk-transport = "0.1" lgalloc = "0.6.0" libc = "0.2.186" lru = "0.16.3" @@ -653,9 +654,6 @@ postgres_array = { git = "https://github.com/MaterializeInc/rust-postgres-array" # Waiting on https://github.com/MaterializeInc/serde-value/pull/35. serde-value = { git = "https://github.com/MaterializeInc/serde-value.git" } -# Waiting for resolution of https://github.com/launchdarkly/rust-server-sdk/issues/116 -launchdarkly-server-sdk = { git = "https://github.com/MaterializeInc/rust-server-sdk", rev = "3e0a0b98b09a2970f292577a07e1c9382b65b5da" } - # Waiting on https://github.com/edenhill/librdkafka/pull/4051. rdkafka = { git = "https://github.com/MaterializeInc/rust-rdkafka.git" } rdkafka-sys = { git = "https://github.com/MaterializeInc/rust-rdkafka.git" } diff --git a/deny.toml b/deny.toml index ca75af3011106..81034ea2ba2e2 100644 --- a/deny.toml +++ b/deny.toml @@ -150,6 +150,15 @@ skip = [ { name = "hashlink", version = "0.9.1" }, # held back by owo-colors 4.3 (mz-deploy terminal styling) { name = "supports-color", version = "2.1.0" }, + + # Pulled by launchdarkly-server-sdk 3.x via launchdarkly-sdk-transport / + # eventsource-client (proxy/timeout/rustls stack and the SDK's RNG path). + # NB: tower 0.4.13 is already skipped above (mz-deploy). + { name = "rustls-native-certs", version = "0.7.3" }, + { name = "rand", version = "0.10.1" }, + { name = "rand_core", version = "0.10.1" }, + { name = "getrandom", version = "0.4.2" }, + { name = "cpufeatures", version = "0.3.0" }, ] [[bans.deny]] @@ -206,6 +215,7 @@ wrappers = [ "globset", "launchdarkly-server-sdk", "launchdarkly-server-sdk-evaluation", + "launchdarkly-sdk-transport", "native-tls", "opendal", "os_info", diff --git a/src/adapter/Cargo.toml b/src/adapter/Cargo.toml index 76b1c2b2a4b17..cccc24a57cfdb 100644 --- a/src/adapter/Cargo.toml +++ b/src/adapter/Cargo.toml @@ -29,10 +29,10 @@ hex.workspace = true humantime.workspace = true imbl.workspace = true http.workspace = true -hyper-tls = "0.5.0" ipnet.workspace = true itertools.workspace = true launchdarkly-server-sdk.workspace = true +launchdarkly-sdk-transport.workspace = true maplit.workspace = true mz-adapter-types = { path = "../adapter-types" } mz-audit-log = { path = "../audit-log" } diff --git a/src/adapter/src/config/frontend.rs b/src/adapter/src/config/frontend.rs index bc65fb7a293bd..3a942d9602870 100644 --- a/src/adapter/src/config/frontend.rs +++ b/src/adapter/src/config/frontend.rs @@ -10,16 +10,18 @@ use std::collections::BTreeMap; use std::fs; use std::path::PathBuf; -use std::sync::Arc; use std::time::Duration; +use bytes::Bytes; use derivative::Derivative; -use hyper_tls::HttpsConnector; +use futures::TryStreamExt; +use launchdarkly_sdk_transport::{ByteStream, HttpTransport, ResponseFuture}; use launchdarkly_server_sdk as ld; use mz_build_info::BuildInfo; use mz_cloud_provider::CloudProvider; use mz_cluster_client::ReplicaId; use mz_controller_types::ClusterId; +use mz_ore::metrics::UIntGauge; use mz_ore::now::NowFn; use mz_sql::catalog::EnvironmentId; use serde_json::Value as JsonValue; @@ -71,7 +73,7 @@ impl SystemParameterFrontend { /// Create a new [SystemParameterFrontend] initialize. /// /// This will create and initialize an [ld::Client] instance. The - /// [ld::Client::initialized_async] call will be attempted in a loop with an + /// [ld::Client::wait_for_initialization] call will be attempted in a loop with an /// exponential backoff with power `2s` and max duration `60s`. pub async fn from(sync_config: &SystemParameterSyncConfig) -> Result { match &sync_config.backend_config { @@ -347,25 +349,68 @@ pub struct ClusterEvalContext { pub cluster: ClusterScopeContext, } -fn ld_config(api_key: &str, metrics: &Metrics) -> ld::Config { +/// An [`HttpTransport`] wrapper that records timestamps on successful HTTP +/// responses. Used to populate Prometheus metrics that track LaunchDarkly +/// connectivity health. +/// +/// Two instances are created — one for the event processor (CSE metric, tracks +/// outbound event sends) and one for the streaming data source (SSE metric, +/// tracks inbound SSE events). +#[derive(Clone)] +struct MetricsTransport { + inner: T, + last_success_gauge: UIntGauge, + now_fn: NowFn, +} + +impl HttpTransport for MetricsTransport { + fn request(&self, request: http::Request>) -> ResponseFuture { + let inner_fut = self.inner.request(request); + let gauge = self.last_success_gauge.clone(); + let now_fn = self.now_fn.clone(); + Box::pin(async move { + let resp = inner_fut.await?; + if resp.status().is_success() { + gauge.set(now_fn() / 1000); + let (parts, body) = resp.into_parts(); + let wrapped: ByteStream = Box::pin(body.inspect_ok(move |_| { + gauge.set(now_fn() / 1000); + })); + Ok(http::Response::from_parts(parts, wrapped)) + } else { + Ok(resp) + } + }) + } +} + +fn ld_config(api_key: &str, metrics: &Metrics, now_fn: &NowFn) -> ld::Config { + let transport = launchdarkly_sdk_transport::HyperTransport::builder() + .connect_timeout(Duration::from_secs(10)) + .read_timeout(Duration::from_secs(300)) + .build_https() + .expect("failed to create HTTPS transport"); + + let cse_transport = MetricsTransport { + inner: transport.clone(), + last_success_gauge: metrics.last_cse_time_seconds.clone(), + now_fn: now_fn.clone(), + }; + let data_source_transport = MetricsTransport { + inner: transport, + last_success_gauge: metrics.last_sse_time_seconds.clone(), + now_fn: now_fn.clone(), + }; + + let mut event_processor = ld::EventProcessorBuilder::new(); + event_processor.transport(cse_transport); + + let mut data_source = ld::StreamingDataSourceBuilder::new(); + data_source.transport(data_source_transport); + ld::ConfigBuilder::new(api_key) - .event_processor( - ld::EventProcessorBuilder::new() - .https_connector(HttpsConnector::new()) - .on_success({ - let last_cse_time_seconds = metrics.last_cse_time_seconds.clone(); - Arc::new(move |result| { - if let Ok(ts) = u64::try_from(result.time_from_server / 1000) { - last_cse_time_seconds.set(ts); - } else { - tracing::warn!( - "Cannot convert time_from_server / 1000 from u128 to u64" - ); - } - }) - }), - ) - .data_source(ld::StreamingDataSourceBuilder::new().https_connector(HttpsConnector::new())) + .event_processor(&event_processor) + .data_source(&data_source) .build() .expect("valid config") } @@ -375,19 +420,9 @@ async fn ld_client( metrics: &Metrics, now_fn: &NowFn, ) -> Result { - let ld_client = ld::Client::build(ld_config(api_key, metrics))?; + let ld_client = ld::Client::build(ld_config(api_key, metrics, now_fn))?; tracing::info!("waiting for SystemParameterFrontend to initialize"); - // Start and initialize LD client for the frontend. The callback passed - // will export the last time when an SSE event from the LD server was - // received in a Prometheus metric. - ld_client.start_with_default_executor_and_callback({ - let last_sse_time_seconds = metrics.last_sse_time_seconds.clone(); - let now_fn = now_fn.clone(); - Arc::new(move |_ev| { - let ts = now_fn() / 1000; - last_sse_time_seconds.set(ts); - }) - }); + ld_client.start_with_default_executor(); let max_backoff = Duration::from_secs(60); let mut backoff = Duration::from_secs(5); @@ -579,7 +614,13 @@ fn ld_ctx( #[cfg(test)] mod tests { + use std::sync::Arc; + use std::sync::atomic::{AtomicU64, Ordering}; + + use futures::StreamExt; + use launchdarkly_sdk_transport::{ByteStream, TransportError}; use mz_build_info::DUMMY_BUILD_INFO; + use mz_ore::metrics::MetricsRegistry; use super::*; @@ -626,4 +667,212 @@ mod tests { fn environment_wide_context_is_unscoped() { ld_ctx(&env_id(), &DUMMY_BUILD_INFO, None, None).expect("environment-wide context builds"); } + + /// A fake transport that simulates a long-lived SSE streaming connection: + /// returns 200 OK immediately, then delivers multiple SSE events as body + /// chunks (exactly how LaunchDarkly's streaming data source works). + #[derive(Clone)] + struct FakeSseTransport; + + impl HttpTransport for FakeSseTransport { + fn request(&self, _request: http::Request>) -> ResponseFuture { + let body: ByteStream = Box::pin(futures::stream::iter(vec![ + Ok(Bytes::from("event: put\ndata: {\"flags\":{}}\n\n")), + Ok(Bytes::from("event: patch\ndata: {\"key\":\"flag1\"}\n\n")), + Ok(Bytes::from("event: patch\ndata: {\"key\":\"flag2\"}\n\n")), + ])); + Box::pin(async move { + http::Response::builder() + .status(200) + .body(body) + .map_err(|e| TransportError::new(std::io::Error::other(e))) + }) + } + } + + /// A fake transport that returns an error, simulating a failed connection. + #[derive(Clone)] + struct FailingTransport; + + impl HttpTransport for FailingTransport { + fn request(&self, _request: http::Request>) -> ResponseFuture { + Box::pin(async move { + Err(TransportError::new(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "connection refused", + ))) + }) + } + } + + /// A fake transport that returns 200 OK, delivers one event, then errors + /// mid-stream with a timeout: the non-Eof stream error a dropped long-lived + /// SSE connection surfaces. + #[derive(Clone)] + struct MidStreamFailureTransport; + + impl HttpTransport for MidStreamFailureTransport { + fn request(&self, _request: http::Request>) -> ResponseFuture { + let body: ByteStream = Box::pin(futures::stream::iter(vec![ + Ok(Bytes::from("event: put\ndata: {\"flags\":{}}\n\n")), + Err(TransportError::new(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "body timed out", + ))), + ])); + Box::pin(async move { + http::Response::builder() + .status(200) + .body(body) + .map_err(|e| TransportError::new(std::io::Error::other(e))) + }) + } + } + + fn test_gauge(registry: &MetricsRegistry, name: &str) -> UIntGauge { + registry.register(mz_ore::metric!( + name: name, + help: "test gauge", + )) + } + + /// Verifies that MetricsTransport updates the gauge on each body chunk, + /// not just on the initial HTTP 200 response head. This matters for + /// long-lived streaming connections where SSE events arrive as body chunks. + #[mz_ore::test(tokio::test)] + async fn test_metric_updated_on_body_chunks() -> Result<(), anyhow::Error> { + let time = Arc::new(AtomicU64::new(1_000_000)); + let time_clone = Arc::clone(&time); + let now_fn = NowFn::from(move || time_clone.load(Ordering::SeqCst)); + + let registry = MetricsRegistry::new(); + let gauge = test_gauge(®istry, "test_sse_gauge"); + + let transport = MetricsTransport { + inner: FakeSseTransport, + last_success_gauge: gauge.clone(), + now_fn, + }; + + assert_eq!(gauge.get(), 0); + + let request = http::Request::builder() + .uri("https://stream.launchdarkly.com/all") + .body(None)?; + let response = transport.request(request).await?; + + assert_eq!(gauge.get(), 1000); + + time.store(2_800_000, Ordering::SeqCst); + + let mut body = response.into_body(); + let mut event_count = 0; + while let Some(Ok(_chunk)) = body.next().await { + event_count += 1; + } + assert_eq!(event_count, 3); + + assert_eq!(gauge.get(), 2800); + Ok(()) + } + + #[mz_ore::test(tokio::test)] + async fn test_cse_metric_updates_correctly_per_request() -> Result<(), anyhow::Error> { + let time = Arc::new(AtomicU64::new(1_000_000)); + let time_clone = Arc::clone(&time); + let now_fn = NowFn::from(move || time_clone.load(Ordering::SeqCst)); + + let registry = MetricsRegistry::new(); + let gauge = test_gauge(®istry, "test_cse_gauge"); + + let transport = MetricsTransport { + inner: FakeSseTransport, + last_success_gauge: gauge.clone(), + now_fn, + }; + + let req = || -> Result>, http::Error> { + http::Request::builder() + .uri("https://events.launchdarkly.com/bulk") + .body(None) + }; + + let _ = transport.request(req()?).await?; + assert_eq!(gauge.get(), 1000); + + time.store(2_000_000, Ordering::SeqCst); + let _ = transport.request(req()?).await?; + assert_eq!(gauge.get(), 2000); + + time.store(3_000_000, Ordering::SeqCst); + let _ = transport.request(req()?).await?; + assert_eq!(gauge.get(), 3000); + Ok(()) + } + + #[mz_ore::test(tokio::test)] + async fn test_metric_not_updated_on_failed_request() -> Result<(), anyhow::Error> { + let now_fn = NowFn::from(|| 5_000_000u64); + + let registry = MetricsRegistry::new(); + let gauge = test_gauge(®istry, "test_fail_gauge"); + + let transport = MetricsTransport { + inner: FailingTransport, + last_success_gauge: gauge.clone(), + now_fn, + }; + + let request = http::Request::builder() + .uri("https://stream.launchdarkly.com/all") + .body(None)?; + let result = transport.request(request).await; + assert!(result.is_err()); + assert_eq!(gauge.get(), 0, "gauge must not update on transport error"); + Ok(()) + } + + /// Verifies that when an SSE connection returns 200 OK and then dies + /// mid-stream, `last_sse_time_seconds` advances only for the events that + /// arrived and then freezes — the frozen timestamp is what lets the + /// staleness alert detect a stuck data source. + #[mz_ore::test(tokio::test)] + async fn test_metric_frozen_on_midstream_error() -> Result<(), anyhow::Error> { + let time = Arc::new(AtomicU64::new(1_000_000)); + let time_clone = Arc::clone(&time); + let now_fn = NowFn::from(move || time_clone.load(Ordering::SeqCst)); + + let registry = MetricsRegistry::new(); + let gauge = test_gauge(®istry, "test_midstream_gauge"); + + let transport = MetricsTransport { + inner: MidStreamFailureTransport, + last_success_gauge: gauge.clone(), + now_fn, + }; + + // The 200 OK response head updates the gauge. + let request = http::Request::builder() + .uri("https://stream.launchdarkly.com/all") + .body(None)?; + let response = transport.request(request).await?; + assert_eq!(gauge.get(), 1000); + + // The first event arrives and advances the gauge. + time.store(2_000_000, Ordering::SeqCst); + let mut body = response.into_body(); + assert!(matches!(body.next().await, Some(Ok(_)))); + assert_eq!(gauge.get(), 2000); + + // The stream then errors mid-flight. Time has moved forward, but the + // gauge must stay frozen at the last successful event. + time.store(9_000_000, Ordering::SeqCst); + assert!(matches!(body.next().await, Some(Err(_)))); + assert_eq!( + gauge.get(), + 2000, + "gauge must freeze on mid-stream error so the staleness alert can fire" + ); + Ok(()) + } } diff --git a/src/balancerd/Cargo.toml b/src/balancerd/Cargo.toml index eaeb2f93e2ae1..608f743c2fa7d 100644 --- a/src/balancerd/Cargo.toml +++ b/src/balancerd/Cargo.toml @@ -42,6 +42,7 @@ num_cpus.workspace = true openssl.workspace = true prometheus.workspace = true proxy-header.workspace = true +rustls.workspace = true semver.workspace = true tokio.workspace = true tokio-openssl.workspace = true diff --git a/src/balancerd/src/bin/balancerd.rs b/src/balancerd/src/bin/balancerd.rs index 28104b2e63c42..03caa10fc43e5 100644 --- a/src/balancerd/src/bin/balancerd.rs +++ b/src/balancerd/src/bin/balancerd.rs @@ -172,6 +172,14 @@ pub struct ServiceArgs { fn main() { let args: Args = cli::parse_args(CliConfig::default()); + // Pin the rustls crypto provider to aws-lc-rs. The LaunchDarkly SDK uses + // hyper-rustls, so building its client resolves the process-default rustls + // provider. The workspace also links rustls' `ring` feature (pulled by + // other hyper-rustls chains), and with both provider features enabled + // rustls cannot choose a default on its own and panics. The call is + // idempotent, so ignore the result. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + // Mirror the tokio Runtime configuration in our production binaries. let ncpus_useful = usize::max(1, std::cmp::min(num_cpus::get(), num_cpus::get_physical())); let runtime = tokio::runtime::Builder::new_multi_thread() diff --git a/src/clusterd/Cargo.toml b/src/clusterd/Cargo.toml index dd4830573c3cf..e89a4cadd1a3f 100644 --- a/src/clusterd/Cargo.toml +++ b/src/clusterd/Cargo.toml @@ -40,6 +40,7 @@ mz-timely-util = { path = "../timely-util" } mz-txn-wal = { path = "../txn-wal" } nix.workspace = true num_cpus.workspace = true +rustls.workspace = true serde.workspace = true tokio.workspace = true tower.workspace = true diff --git a/src/clusterd/src/lib.rs b/src/clusterd/src/lib.rs index ed97cf19b2fc2..09236abf5707c 100644 --- a/src/clusterd/src/lib.rs +++ b/src/clusterd/src/lib.rs @@ -190,6 +190,14 @@ fn process_ordinal_from_hostname(hostname: &str) -> Option<&str> { pub fn main() { mz_ore::panic::install_enhanced_handler(); + // Pin the rustls crypto provider to aws-lc-rs. The LaunchDarkly SDK uses + // hyper-rustls, so building its client resolves the process-default rustls + // provider. The workspace also links rustls' `ring` feature (pulled by + // other hyper-rustls chains), and with both provider features enabled + // rustls cannot choose a default on its own and panics. The call is + // idempotent, so ignore the result. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + // Derive `CLUSTERD_PROCESS` (the process ordinal) from the pod hostname // when running under Kubernetes and it was not set explicitly. The // distroless image has no shell entrypoint to do this, so clusterd does it diff --git a/src/dyncfg-launchdarkly/Cargo.toml b/src/dyncfg-launchdarkly/Cargo.toml index 42ff83496db0e..b1efcbece2523 100644 --- a/src/dyncfg-launchdarkly/Cargo.toml +++ b/src/dyncfg-launchdarkly/Cargo.toml @@ -13,8 +13,8 @@ workspace = true [dependencies] anyhow.workspace = true humantime.workspace = true -hyper-tls = "0.5.0" launchdarkly-server-sdk.workspace = true +launchdarkly-sdk-transport.workspace = true mz-build-info = { path = "../build-info" } mz-dyncfg = { path = "../dyncfg" } mz-ore = { path = "../ore", default-features = false, features = ["async"] } diff --git a/src/dyncfg-launchdarkly/src/lib.rs b/src/dyncfg-launchdarkly/src/lib.rs index 3e25eb5b41bb4..50f412f777763 100644 --- a/src/dyncfg-launchdarkly/src/lib.rs +++ b/src/dyncfg-launchdarkly/src/lib.rs @@ -11,7 +11,6 @@ use std::time::Duration; -use hyper_tls::HttpsConnector; use launchdarkly_server_sdk as ld; use mz_build_info::BuildInfo; use mz_dyncfg::{ConfigSet, ConfigUpdates, ConfigVal}; @@ -50,13 +49,21 @@ where let _ = dyn_into_flag(entry.val())?; } let ld_client = if let Some(key) = launchdarkly_sdk_key { + let transport = launchdarkly_sdk_transport::HyperTransport::builder() + .connect_timeout(Duration::from_secs(10)) + .read_timeout(Duration::from_secs(300)) + .build_https() + .expect("failed to create HTTPS transport"); + + let mut data_source = ld::StreamingDataSourceBuilder::new(); + data_source.transport(transport.clone()); + + let mut event_processor = ld::EventProcessorBuilder::new(); + event_processor.transport(transport); + let config = ld::ConfigBuilder::new(key) - .event_processor( - ld::EventProcessorBuilder::new().https_connector(HttpsConnector::new()), - ) - .data_source( - ld::StreamingDataSourceBuilder::new().https_connector(HttpsConnector::new()), - ) + .data_source(&data_source) + .event_processor(&event_processor) .build() .expect("valid config"); let client = ld::Client::build(config)?; diff --git a/src/environmentd/Cargo.toml b/src/environmentd/Cargo.toml index 7fe2599be2168..1872038b12cac 100644 --- a/src/environmentd/Cargo.toml +++ b/src/environmentd/Cargo.toml @@ -96,6 +96,7 @@ rand.workspace = true regex = { workspace = true, optional = true } reqwest.workspace = true rlimit.workspace = true +rustls.workspace = true semver.workspace = true sentry-tracing.workspace = true serde.workspace = true diff --git a/src/environmentd/src/environmentd/main.rs b/src/environmentd/src/environmentd/main.rs index 489b63464507e..17a082e86edfd 100644 --- a/src/environmentd/src/environmentd/main.rs +++ b/src/environmentd/src/environmentd/main.rs @@ -657,6 +657,15 @@ pub fn main() { fn run(mut args: Args) -> Result<(), anyhow::Error> { mz_ore::panic::install_enhanced_handler(); + + // Pin the rustls crypto provider to aws-lc-rs. The LaunchDarkly SDK uses + // hyper-rustls, so building its client resolves the process-default rustls + // provider. The workspace also links rustls' `ring` feature (pulled by + // other hyper-rustls chains), and with both provider features enabled + // rustls cannot choose a default on its own and panics. The call is + // idempotent, so ignore the result. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let envd_start = Instant::now(); // Configure signal handling as soon as possible. We want signals to be diff --git a/src/sqllogictest/Cargo.toml b/src/sqllogictest/Cargo.toml index c5276285203cc..da6264f1dca92 100644 --- a/src/sqllogictest/Cargo.toml +++ b/src/sqllogictest/Cargo.toml @@ -46,6 +46,7 @@ mz-tracing = { path = "../tracing" } postgres-protocol.workspace = true regex.workspace = true reqwest.workspace = true +rustls.workspace = true shell-words.workspace = true serde_json.workspace = true tempfile.workspace = true diff --git a/src/sqllogictest/src/bin/sqllogictest.rs b/src/sqllogictest/src/bin/sqllogictest.rs index f610dada2d21e..93301c6b5d6db 100644 --- a/src/sqllogictest/src/bin/sqllogictest.rs +++ b/src/sqllogictest/src/bin/sqllogictest.rs @@ -114,6 +114,14 @@ struct Args { async fn main() -> ExitCode { mz_ore::panic::install_enhanced_handler(); + // Pin the rustls crypto provider to aws-lc-rs. The LaunchDarkly SDK uses + // hyper-rustls, so building its client resolves the process-default rustls + // provider. The workspace also links rustls' `ring` feature (pulled by + // other hyper-rustls chains), and with both provider features enabled + // rustls cannot choose a default on its own and panics. The call is + // idempotent, so ignore the result. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + let args: Args = cli::parse_args(CliConfig { env_prefix: Some("MZ_"), enable_version_flag: false, diff --git a/src/testdrive/Cargo.toml b/src/testdrive/Cargo.toml index c3654b101d7b8..5d2913820c2bb 100644 --- a/src/testdrive/Cargo.toml +++ b/src/testdrive/Cargo.toml @@ -66,6 +66,7 @@ rand.workspace = true rdkafka.workspace = true regex.workspace = true reqwest.workspace = true +rustls.workspace = true semver.workspace = true serde.workspace = true serde_json = { workspace = true, features = ["raw_value"] } diff --git a/src/testdrive/src/bin/testdrive.rs b/src/testdrive/src/bin/testdrive.rs index 4acbb284c2dfa..9e2b028f0dac2 100644 --- a/src/testdrive/src/bin/testdrive.rs +++ b/src/testdrive/src/bin/testdrive.rs @@ -300,6 +300,14 @@ struct Args { async fn main() { let args: Args = cli::parse_args(CliConfig::default()); + // Pin the rustls crypto provider to aws-lc-rs. The LaunchDarkly SDK uses + // hyper-rustls, so building its client resolves the process-default rustls + // provider. The workspace also links rustls' `ring` feature (pulled by + // other hyper-rustls chains), and with both provider features enabled + // rustls cannot choose a default on its own and panics. The call is + // idempotent, so ignore the result. + let _ = rustls::crypto::aws_lc_rs::default_provider().install_default(); + tracing_subscriber::fmt() .with_env_filter(EnvFilter::from(args.log_filter)) .with_writer(io::stdout) From aaf2f277c7aa291afba88dc03df2c22084612857 Mon Sep 17 00:00:00 2001 From: Jason Hernandez <7144515+jasonhernandez@users.noreply.github.com> Date: Mon, 6 Jul 2026 08:50:38 -0700 Subject: [PATCH 2/3] adapter: add LaunchDarkly reconnect integration test incident-984 was a runtime failure: the LaunchDarkly data source stopped reconnecting after its streaming connection died, silently wedging flag sync. The existing test/launchdarkly nightly covers value sync, persistence, targeting, and the kill switch, but nothing exercises reconnect after a mid-stream failure. This adds test/launchdarkly-reconnect, which reproduces the incident deterministically against a mock and needs no real LaunchDarkly credentials. The failure mode matters. The incident signature was a mid-body read timeout on a silently-dead connection, hyper::Error(Body, Kind(TimedOut)), which the eventsource client surfaces to the data source as a stream error. A TCP RST or a clean FIN does NOT reproduce it: those are retried inside the eventsource client on every SDK version and never reach the data source (verified empirically, nightly #17018 stayed green with an RST-based mock even on the exact incident-era dependency stack). The mock serves an initial flag value (2 GiB) on the first TWO streaming connections and then goes silent holding each open, so the transport read timeout ends the stream. Two, because environmentd creates a short-lived bootstrap LD client at boot (load_remote_system_parameters) before the long-lived sync client, and the stall must hit the sync client (verified empirically, nightly #17020: with only the first connection stalled, the sync client landed on connection two and got the updated value without any reconnect being exercised). Every later (reconnecting) client gets the updated value (3 GiB) plus heartbeats. Production changes, both hidden test knobs: - --launchdarkly-base-uri (env MZ_LAUNCHDARKLY_BASE_URI) overrides the SDK's streaming/polling/events endpoints with a single base URL via the SDK's ServiceEndpointsBuilder::relay_proxy, letting tests point the SDK at the mock. Also generally useful for relay-proxy setups. - MZ_LAUNCHDARKLY_READ_TIMEOUT overrides the transport's streaming read timeout (default 300s) so the test can trigger the timeout path in seconds. mzcompose.py boots environmentd against the mock with a 5s read timeout and asserts SHOW max_result_size reaches 3GB, which can only happen if the sync client's data source reconnected after the timeout. A regressed SDK stays stuck at 2GB and the assertion times out. Wired into the nightly pipeline. Co-Authored-By: Claude Opus 4.8 (1M context) --- ci/nightly/pipeline.template.yml | 11 ++ src/adapter/src/config.rs | 5 + src/adapter/src/config/frontend.rs | 43 ++++-- src/environmentd/src/environmentd/main.rs | 6 + src/environmentd/src/lib.rs | 4 + src/environmentd/src/test_util.rs | 1 + src/sqllogictest/src/runner.rs | 1 + test/launchdarkly-reconnect/mock_ld.py | 162 ++++++++++++++++++++++ test/launchdarkly-reconnect/mzcompose.py | 93 +++++++++++++ 9 files changed, 317 insertions(+), 9 deletions(-) create mode 100644 test/launchdarkly-reconnect/mock_ld.py create mode 100644 test/launchdarkly-reconnect/mzcompose.py diff --git a/ci/nightly/pipeline.template.yml b/ci/nightly/pipeline.template.yml index ad2b69637b4b3..0e3e705e35547 100644 --- a/ci/nightly/pipeline.template.yml +++ b/ci/nightly/pipeline.template.yml @@ -1492,6 +1492,17 @@ steps: composition: launchdarkly-flag-consistency branches: "main" + - id: launchdarkly-reconnect + label: "LaunchDarkly reconnect" + depends_on: build-aarch64 + timeout_in_minutes: 30 + agents: + # Uses a mock LaunchDarkly server; needs no real credentials. + queue: hetzner-aarch64-4cpu-8gb + plugins: + - ./ci/plugins/mzcompose: + composition: launchdarkly-reconnect + - group: E2E key: e2e steps: diff --git a/src/adapter/src/config.rs b/src/adapter/src/config.rs index 46ee79ce1ed69..82fdfe03a5709 100644 --- a/src/adapter/src/config.rs +++ b/src/adapter/src/config.rs @@ -109,6 +109,11 @@ pub enum SystemParameterSyncClientConfig { LaunchDarkly { /// The LaunchDarkly SDK key sdk_key: String, + /// Overrides the LaunchDarkly streaming, polling, and events endpoints + /// with a single base URL (as for a relay proxy). `None` uses + /// LaunchDarkly's default endpoints. Primarily for pointing the SDK at + /// a mock server in tests. + base_uri: Option, /// Function to return the current time. now_fn: NowFn, }, diff --git a/src/adapter/src/config/frontend.rs b/src/adapter/src/config/frontend.rs index 3a942d9602870..71b1b454212cf 100644 --- a/src/adapter/src/config/frontend.rs +++ b/src/adapter/src/config/frontend.rs @@ -84,9 +84,14 @@ impl SystemParameterFrontend { build_info: sync_config.build_info, metrics: sync_config.metrics.clone(), }), - SystemParameterSyncClientConfig::LaunchDarkly { sdk_key, now_fn } => Ok(Self { + SystemParameterSyncClientConfig::LaunchDarkly { + sdk_key, + base_uri, + now_fn, + } => Ok(Self { client: SystemParameterFrontendClient::LaunchDarkly { - client: ld_client(sdk_key, &sync_config.metrics, now_fn).await?, + client: ld_client(sdk_key, base_uri.as_deref(), &sync_config.metrics, now_fn) + .await?, // The environment-wide context carries no cluster/replica // scope. Scoped evaluation passes a `cluster` or `replica` // context per pass via [`ld_ctx`]. @@ -384,10 +389,25 @@ impl HttpTransport for MetricsTransport { } } -fn ld_config(api_key: &str, metrics: &Metrics, now_fn: &NowFn) -> ld::Config { +fn ld_config( + api_key: &str, + base_uri: Option<&str>, + metrics: &Metrics, + now_fn: &NowFn, +) -> ld::Config { + // How long a body read on the streaming connection may stay idle before + // the transport surfaces a timeout error to the data source. This is the + // error class of incident-984 (a silently-dead connection). Overridable + // via a hidden env var so tests can trigger the timeout path in seconds + // instead of minutes (see test/launchdarkly-reconnect). + let read_timeout = std::env::var("MZ_LAUNCHDARKLY_READ_TIMEOUT") + .ok() + .and_then(|v| humantime::parse_duration(&v).ok()) + .unwrap_or(Duration::from_secs(300)); + let transport = launchdarkly_sdk_transport::HyperTransport::builder() .connect_timeout(Duration::from_secs(10)) - .read_timeout(Duration::from_secs(300)) + .read_timeout(read_timeout) .build_https() .expect("failed to create HTTPS transport"); @@ -408,19 +428,24 @@ fn ld_config(api_key: &str, metrics: &Metrics, now_fn: &NowFn) -> ld::Config { let mut data_source = ld::StreamingDataSourceBuilder::new(); data_source.transport(data_source_transport); - ld::ConfigBuilder::new(api_key) + let mut config = ld::ConfigBuilder::new(api_key) .event_processor(&event_processor) - .data_source(&data_source) - .build() - .expect("valid config") + .data_source(&data_source); + if let Some(base_uri) = base_uri { + let mut endpoints = ld::ServiceEndpointsBuilder::new(); + endpoints.relay_proxy(base_uri); + config = config.service_endpoints(&endpoints); + } + config.build().expect("valid config") } async fn ld_client( api_key: &str, + base_uri: Option<&str>, metrics: &Metrics, now_fn: &NowFn, ) -> Result { - let ld_client = ld::Client::build(ld_config(api_key, metrics, now_fn))?; + let ld_client = ld::Client::build(ld_config(api_key, base_uri, metrics, now_fn))?; tracing::info!("waiting for SystemParameterFrontend to initialize"); ld_client.start_with_default_executor(); diff --git a/src/environmentd/src/environmentd/main.rs b/src/environmentd/src/environmentd/main.rs index 17a082e86edfd..60a744a57d0d2 100644 --- a/src/environmentd/src/environmentd/main.rs +++ b/src/environmentd/src/environmentd/main.rs @@ -434,6 +434,11 @@ pub struct Args { /// configuration parameters. #[clap(long, env = "LAUNCHDARKLY_SDK_KEY")] launchdarkly_sdk_key: Option, + /// Overrides the LaunchDarkly streaming, polling, and events endpoints with + /// a single base URL, as for a relay proxy. Primarily intended for pointing + /// the SDK at a mock LaunchDarkly server in tests. + #[clap(long, env = "LAUNCHDARKLY_BASE_URI", value_name = "URL")] + launchdarkly_base_uri: Option, /// A list of PARAM_NAME=KEY_NAME pairs from system parameter names to /// LaunchDarkly feature keys. /// @@ -1124,6 +1129,7 @@ fn run(mut args: Args) -> Result<(), anyhow::Error> { segment_client_side: args.segment_client_side, test_only_dummy_segment_client: args.test_only_dummy_segment_client, launchdarkly_sdk_key: args.launchdarkly_sdk_key, + launchdarkly_base_uri: args.launchdarkly_base_uri, launchdarkly_key_map: args .launchdarkly_key_map .into_iter() diff --git a/src/environmentd/src/lib.rs b/src/environmentd/src/lib.rs index d04e8487c0202..246ed69497c75 100644 --- a/src/environmentd/src/lib.rs +++ b/src/environmentd/src/lib.rs @@ -166,6 +166,9 @@ pub struct Config { /// An SDK key for LaunchDarkly. Enables system parameter synchronization /// with LaunchDarkly. pub launchdarkly_sdk_key: Option, + /// Overrides the LaunchDarkly service endpoints with a single base URL, as + /// for a relay proxy or a mock server in tests. + pub launchdarkly_base_uri: Option, /// An invertible map from system parameter names to LaunchDarkly feature /// keys to use when propagating values from the latter to the former. pub launchdarkly_key_map: BTreeMap, @@ -480,6 +483,7 @@ impl Listeners { config.launchdarkly_key_map, SystemParameterSyncClientConfig::LaunchDarkly { sdk_key: key, + base_uri: config.launchdarkly_base_uri, now_fn: config.now.clone(), }, )), diff --git a/src/environmentd/src/test_util.rs b/src/environmentd/src/test_util.rs index ec878b33b206e..b29bd530d32c3 100644 --- a/src/environmentd/src/test_util.rs +++ b/src/environmentd/src/test_util.rs @@ -903,6 +903,7 @@ impl Listeners { aws_account_id: None, aws_privatelink_availability_zones: None, launchdarkly_sdk_key: None, + launchdarkly_base_uri: None, launchdarkly_key_map: Default::default(), config_sync_file_path: None, config_sync_timeout: Duration::from_secs(30), diff --git a/src/sqllogictest/src/runner.rs b/src/sqllogictest/src/runner.rs index 9033442ffccdd..a86b2caf09e29 100644 --- a/src/sqllogictest/src/runner.rs +++ b/src/sqllogictest/src/runner.rs @@ -1281,6 +1281,7 @@ impl<'a> RunnerInner<'a> { aws_account_id: None, aws_privatelink_availability_zones: None, launchdarkly_sdk_key: None, + launchdarkly_base_uri: None, launchdarkly_key_map: Default::default(), config_sync_file_path: None, config_sync_timeout: Duration::from_secs(30), diff --git a/test/launchdarkly-reconnect/mock_ld.py b/test/launchdarkly-reconnect/mock_ld.py new file mode 100644 index 0000000000000..87e118132b7c6 --- /dev/null +++ b/test/launchdarkly-reconnect/mock_ld.py @@ -0,0 +1,162 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +""" +A minimal mock of the LaunchDarkly streaming API, used to exercise the SDK's +reconnect behavior (see mzcompose.py). + +The first STALL_CONNECTIONS streaming clients receive an initial flag value +and then the connection goes silent: the mock holds it open without sending +any further bytes and without closing it. The SDK's transport read timeout +then fires mid-body, surfacing the exact error class of incident-984 (a +silently-dead connection producing `hyper::Error(Body, Kind(TimedOut))`). A +TCP RST or a clean FIN would not do: those are retried inside the eventsource +client on every SDK version and never reach the data source. + +Every subsequent (reconnecting) client receives an updated value and periodic +heartbeats. The test asserts that the value changed, which can only happen if +the long-lived sync client reconnected after the timeout (see +STALL_CONNECTIONS for why the first stall alone is not enough). +""" + +import json +import sys +import threading +import time +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer + +PORT = 8080 +FLAG_KEY = "reconnect-test" + +# Number variations served before and after the forced reconnect. The SDK maps +# these onto the `max_result_size` system parameter (2 GiB and 3 GiB). +INITIAL_VALUE = 2147483648 +RECONNECT_VALUE = 3221225472 + +# How many streaming connections receive the initial value and then stall. +# environmentd creates TWO LaunchDarkly clients at boot: a short-lived +# bootstrap loader (load_remote_system_parameters, dropped right after +# initialization) and the long-lived sync client. Stalling only the first +# connection would waste the stall on the bootstrap loader and hand the sync +# client the updated value with no reconnect exercised at all. Stalling the +# first two covers both: the sync client always receives at least one +# stall-then-timeout cycle, even if the bootstrap connection ever goes away. +STALL_CONNECTIONS = 2 + +# Seconds to hold a stalled connection open. Must comfortably exceed the SDK's +# read timeout (MZ_LAUNCHDARKLY_READ_TIMEOUT in mzcompose.py) so the timeout, +# not a connection close, ends the stream. +STALL_SECONDS = 600.0 + +# Seconds between heartbeats on healthy (reconnected) streams. Must be well +# below the SDK's read timeout so a healthy stream never trips it. +HEARTBEAT_SECONDS = 1.0 + +_lock = threading.Lock() +_stream_connections = 0 + + +def put_event(value: int, version: int) -> bytes: + """Serialize a LaunchDarkly streaming `put` event carrying a single flag + that, with targeting off, evaluates to `value`.""" + flag = { + "key": FLAG_KEY, + "version": version, + "on": False, + "targets": [], + "rules": [], + "prerequisites": [], + "fallthrough": {"variation": 0}, + "offVariation": 0, + "variations": [value], + "salt": "reconnect-test-salt", + "clientSideAvailability": { + "usingMobileKey": False, + "usingEnvironmentId": False, + }, + } + payload = {"path": "/", "data": {"flags": {FLAG_KEY: flag}, "segments": {}}} + return f"event: put\ndata: {json.dumps(payload)}\n\n".encode() + + +class Handler(BaseHTTPRequestHandler): + protocol_version = "HTTP/1.1" + + def log_message(self, format: str, *args: object) -> None: + sys.stderr.write("mock-ld: " + (format % args) + "\n") + + def _send_stream_headers(self) -> None: + self.send_response(200) + self.send_header("Content-Type", "text/event-stream") + self.send_header("Cache-Control", "no-cache") + self.end_headers() + + def do_GET(self) -> None: + if self.path == "/health": + self.send_response(200) + self.send_header("Content-Length", "0") + self.end_headers() + return + + if self.path != "/all": + self.send_response(404) + self.send_header("Content-Length", "0") + self.end_headers() + return + + global _stream_connections + with _lock: + _stream_connections += 1 + n = _stream_connections + + self._send_stream_headers() + + if n <= STALL_CONNECTIONS: + # Send the initial value, then go silent while holding the + # connection open. No further bytes, no FIN, no RST: the SDK's + # read timeout must be what ends the stream, surfacing the + # incident-984 error class to the data source. + self.wfile.write(put_event(INITIAL_VALUE, 1)) + self.wfile.flush() + self.log_message("stalling streaming connection %d", n) + time.sleep(STALL_SECONDS) + self.close_connection = True + return + + # Reconnecting client: send the updated value, then hold the connection + # open with periodic heartbeats so the SDK stays connected. + self.wfile.write(put_event(RECONNECT_VALUE, 2)) + self.wfile.flush() + try: + while True: + time.sleep(HEARTBEAT_SECONDS) + self.wfile.write(b":heartbeat\n\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, OSError): + return + + def do_POST(self) -> None: + # The event processor POSTs analytics events to `/bulk`; accept and + # discard them so it doesn't log errors. + length = int(self.headers.get("Content-Length", 0)) + if length: + self.rfile.read(length) + self.send_response(202) + self.send_header("Content-Length", "0") + self.end_headers() + + +def main() -> None: + server = ThreadingHTTPServer(("0.0.0.0", PORT), Handler) + sys.stderr.write(f"mock-ld: listening on {PORT}\n") + server.serve_forever() + + +if __name__ == "__main__": + main() diff --git a/test/launchdarkly-reconnect/mzcompose.py b/test/launchdarkly-reconnect/mzcompose.py new file mode 100644 index 0000000000000..85887488d95d6 --- /dev/null +++ b/test/launchdarkly-reconnect/mzcompose.py @@ -0,0 +1,93 @@ +# Copyright Materialize, Inc. and contributors. All rights reserved. +# +# Use of this software is governed by the Business Source License +# included in the LICENSE file at the root of this repository. +# +# As of the Change Date specified in that file, in accordance with +# the Business Source License, use of this software will be governed +# by the Apache License, Version 2.0. + +""" +Regression test for incident-984: the LaunchDarkly data source must reconnect +after its streaming connection dies with a mid-body read timeout, so that flag +updates keep syncing. + +A mock LaunchDarkly server (mock_ld.py) serves an initial flag value on the +first two streaming connections and then goes silent while holding each +connection open. Two, because environmentd creates a short-lived bootstrap LD +client at boot before the long-lived sync client, and the stall must hit the +sync client. The SDK's transport read timeout (shortened to seconds via +MZ_LAUNCHDARKLY_READ_TIMEOUT) then fires mid-body, reproducing the exact +incident error class: `hyper::Error(Body, Kind(TimedOut))` on a silently-dead +connection. Note that a TCP RST or clean FIN would not reproduce it, those are +retried inside the eventsource client on every SDK version. + +Every reconnecting client receives an updated value. environmentd is pointed +at the mock via MZ_LAUNCHDARKLY_BASE_URI, so the updated value can only reach +it if the data source reconnected after the timeout. A regressed SDK gets +stuck on the initial value and the assertion below times out. + +Unlike test/launchdarkly, this needs no real LaunchDarkly credentials. +""" + +from materialize.mzcompose.composition import Composition, Service +from materialize.mzcompose.service import Service as DockerService +from materialize.mzcompose.services.materialized import Materialized +from materialize.mzcompose.services.testdrive import Testdrive + +FLAG_KEY = "reconnect-test" +MOCK_HOST = "mock-launchdarkly" +MOCK_PORT = 8080 + +SERVICES = [ + DockerService( + name=MOCK_HOST, + config={ + "image": "python:3.12-slim", + "volumes": ["./mock_ld.py:/app/mock_ld.py"], + "command": ["python3", "-u", "/app/mock_ld.py"], + "ports": [MOCK_PORT], + "healthcheck": { + "test": [ + "CMD", + "python3", + "-c", + "import urllib.request; urllib.request.urlopen('http://localhost:8080/health')", + ], + "interval": "1s", + "start_period": "30s", + }, + }, + ), + Materialized( + environment_extra=[ + "MZ_LAUNCHDARKLY_SDK_KEY=sdk-mock-key", + f"MZ_LAUNCHDARKLY_BASE_URI=http://{MOCK_HOST}:{MOCK_PORT}", + f"MZ_LAUNCHDARKLY_KEY_MAP=max_result_size={FLAG_KEY}", + "MZ_CONFIG_SYNC_LOOP_INTERVAL=1s", + # Shorten the streaming read timeout (default 300s) so the mock's + # stalled connection times out in seconds. Must be well above the + # mock's 1s heartbeat interval so healthy streams never trip it. + "MZ_LAUNCHDARKLY_READ_TIMEOUT=5s", + ], + additional_system_parameter_defaults={ + "log_filter": "mz_adapter::config=debug,launchdarkly_server_sdk=debug,info", + }, + external_metadata_store=True, + ), + # The reconnect (eventsource backoff) plus the 1s sync loop means the + # updated value can take several seconds to land; give it ample room. + Testdrive(no_reset=True, seed=1, default_timeout="120s"), +] + + +def workflow_default(c: Composition) -> None: + c.up(MOCK_HOST, "materialized", Service("testdrive", idle=True)) + + # The mock serves 2 GiB on the first two streaming connections and stalls + # each until the SDK's read timeout fires, then serves 3 GiB to every + # later (reconnecting) client. Reaching 3 GiB therefore proves the sync + # client's data source reconnected after the mid-body timeout; a regressed + # SDK stays stuck at 2 GiB and this assertion times out. We don't assert + # the transient 2 GiB value, as the timeout can race startup. + c.testdrive("\n".join(["> SHOW max_result_size", "3GB"])) From 8156f3fba7830027b37d0c91f99a05ed50d3f73b Mon Sep 17 00:00:00 2001 From: Jason Hernandez <7144515+jasonhernandez@users.noreply.github.com> Date: Mon, 6 Jul 2026 08:51:08 -0700 Subject: [PATCH 3/3] adapter: pin LaunchDarkly stack to incident-era versions to prove reconnect test fails DO NOT MERGE. This is a throwaway verification layer on top of the upstream migration and its reconnect test. It downgrades the LaunchDarkly dependency stack to the exact pre-fix versions from incident-984 to prove the reconnect test catches the regression by going red. Pinned (matching the lockfile of the reverted #35903, the incident build): - launchdarkly-server-sdk = 3.0.1 (predates rust-server-sdk#168, in 3.0.3) - eventsource-client = 0.17.1 (predates the reconnect fixes #134/#135, which shipped in 0.17.4) - launchdarkly-sdk-transport = 0.1.1 Pinning only the SDK to 3.0.1 is NOT sufficient: a nightly run of that configuration came back green, because the lock still resolved eventsource-client to 0.17.5, whose reconnect fix recovers the stream beneath the pre-fix SDK. The regression only reproduces with the pre-fix eventsource-client as well. The reconnect test cuts the first streaming connection with a TCP RST, the non-Eof error class those fixes address. On the incident-era stack the data source gives up and never reconnects, so the mock's updated value never arrives and `SHOW max_result_size` stays at 2GB until the assertion times out. A red `LaunchDarkly reconnect` nightly is the expected, desired outcome. The SDK is pinned as exactly "=3.0.1" in Cargo.toml. eventsource-client and launchdarkly-sdk-transport are transitive, so their pins live only in Cargo.lock (via cargo update --precise). Do not regenerate the lockfile on this branch, that would silently float them back to the fixed versions. Co-Authored-By: Claude Opus 4.8 (1M context) --- Cargo.lock | 58 ++++++++++-------------------------------------------- Cargo.toml | 6 +++++- 2 files changed, 15 insertions(+), 49 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index c8cb45094f58b..cbb6d26e41966 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -31,7 +31,7 @@ checksum = "b169f7a6d4742236a0a00c541b845991d0ac43e546831af1249753ab4c3aa3a0" dependencies = [ "cfg-if", "cipher", - "cpufeatures 0.2.17", + "cpufeatures", ] [[package]] @@ -1939,17 +1939,6 @@ version = "0.2.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "613afe47fcd5fac7ccf1db93babcb082c5994d996f20b8b159f2ad1658eb5724" -[[package]] -name = "chacha20" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "d524456ba66e72eb8b115ff89e01e497f8e6d11d78b70b1aa13c0fbd97540a81" -dependencies = [ - "cfg-if", - "cpufeatures 0.3.0", - "rand_core 0.10.1", -] - [[package]] name = "chrono" version = "0.4.41" @@ -2366,15 +2355,6 @@ dependencies = [ "libc", ] -[[package]] -name = "cpufeatures" -version = "0.3.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "8b2a41393f66f16b0823bb79094d54ac5fbd34ab292ddafb9a0456ac9f87d201" -dependencies = [ - "libc", -] - [[package]] name = "crc" version = "3.3.0" @@ -3468,9 +3448,9 @@ dependencies = [ [[package]] name = "eventsource-client" -version = "0.17.5" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96df8cfa11d3c8e4e1a48b81c9c600ecbe8c11d1326f41e1524cc4d42f7bf6d9" +checksum = "f08d78ef3c0bd7cef8f9fdf7822eceb44fe2300689906b9984c577f549bc80dd" dependencies = [ "base64 0.22.1", "bytes", @@ -3479,7 +3459,7 @@ dependencies = [ "launchdarkly-sdk-transport", "log", "pin-project", - "rand 0.10.2", + "rand 0.8.5", "tokio", ] @@ -4038,7 +4018,6 @@ dependencies = [ "cfg-if", "libc", "r-efi 6.0.0", - "rand_core 0.10.1", ] [[package]] @@ -5257,9 +5236,9 @@ dependencies = [ [[package]] name = "launchdarkly-sdk-transport" -version = "0.1.4" +version = "0.1.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "032d66802c461d7254ed1de887be6ca98ebbe126867d9a60c754baae81c4d10d" +checksum = "1fe83622d04dfcaaeac0b5e3aaa1cc156eb1e70c8b68dfcaffaee4365faa00d3" dependencies = [ "bytes", "futures", @@ -5277,9 +5256,9 @@ dependencies = [ [[package]] name = "launchdarkly-server-sdk" -version = "3.1.1" +version = "3.0.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0440c21aaa0670948071d9cbd8142e9d057a4b40feb3e6293e0efc6de9879b69" +checksum = "fc7d03f7f2557dfebc24147d90b2080313fd5a60a4de5219dce1b574956f1c4e" dependencies = [ "aws-lc-rs", "bitflags 2.11.0", @@ -10799,17 +10778,6 @@ dependencies = [ "rand_core 0.9.3", ] -[[package]] -name = "rand" -version = "0.10.2" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c7f5fa3a058cd35567ef9bfa5e75732bee0f9e4c55fa90477bef2dfcdbc4be80" -dependencies = [ - "chacha20", - "getrandom 0.4.3", - "rand_core 0.10.1", -] - [[package]] name = "rand_chacha" version = "0.2.2" @@ -10868,12 +10836,6 @@ dependencies = [ "getrandom 0.3.3", ] -[[package]] -name = "rand_core" -version = "0.10.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "63b8176103e19a2643978565ca18b50549f6101881c443590420e4dc998a3c69" - [[package]] name = "rand_distr" version = "0.5.1" @@ -12058,7 +12020,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e3bf829a2d51ab4a5ddf1352d8470c140cadc8301b2ae1789db023f01cedd6ba" dependencies = [ "cfg-if", - "cpufeatures 0.2.17", + "cpufeatures", "digest", ] @@ -12075,7 +12037,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a7507d819769d01a365ab707794a4084392c824f54a7a6a7862f8c3d0892b283" dependencies = [ "cfg-if", - "cpufeatures 0.2.17", + "cpufeatures", "digest", "sha2-asm", ] diff --git a/Cargo.toml b/Cargo.toml index 11a53e97957b3..bdc21876cd6dd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -407,7 +407,11 @@ junit-report = "0.8.3" k8s-controller = "0.11.0" k8s-openapi = { version = "0.27.0", features = ["schemars", "v1_32"] } kube = { version = "3.1.0", default-features = false, features = ["client", "derive", "openssl-tls", "runtime", "ws"] } -launchdarkly-server-sdk = { version = "3.1.1", default-features = false, features = ["hyper-rustls-native-roots", "crypto-aws-lc-rs"] } +# NB: pinned to EXACTLY 3.0.1 (not ^3.0.1). This is the pre-fix SDK that +# predates rust-server-sdk#168, and this throwaway branch exists to prove the +# reconnect test goes red against it. A caret range would let cargo resolve to +# 3.1.1 (which carries the fix) and silently turn the test green. +launchdarkly-server-sdk = { version = "=3.0.1", default-features = false, features = ["hyper-rustls-native-roots", "crypto-aws-lc-rs"] } launchdarkly-sdk-transport = "0.1" lgalloc = "0.6.0" libc = "0.2.186"