diff --git a/Cargo.lock b/Cargo.lock index 63ab9d3ca9797..cbb6d26e41966 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2006,6 +2006,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" @@ -3028,7 +3034,7 @@ dependencies = [ "libc", "option-ext", "redox_users", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -3442,14 +3448,15 @@ dependencies = [ [[package]] name = "eventsource-client" -version = "0.16.0" +version = "0.17.1" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "c26451361cde19fe2322835b6e684f4825902edf3139ce9d6a70e6542566a0b2" +checksum = "f08d78ef3c0bd7cef8f9fdf7822eceb44fe2300689906b9984c577f549bc80dd" 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", @@ -3997,11 +4004,22 @@ 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", +] + [[package]] name = "gimli" version = "0.32.3" @@ -4424,6 +4442,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 +4498,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 +4557,7 @@ dependencies = [ "libc", "percent-encoding", "pin-project-lite", - "socket2 0.6.4", + "socket2 0.5.10", "system-configuration", "tokio", "tower-service", @@ -4751,7 +4776,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 +4870,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46" dependencies = [ "hermit-abi", "libc", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -4932,7 +4957,7 @@ dependencies = [ "portable-atomic", "portable-atomic-util", "serde_core", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -5128,7 +5153,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 +5234,43 @@ dependencies = [ "tracing", ] +[[package]] +name = "launchdarkly-sdk-transport" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1fe83622d04dfcaaeac0b5e3aaa1cc156eb1e70c8b68dfcaffaee4365faa00d3" +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.0.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc7d03f7f2557dfebc24147d90b2080313fd5a60a4de5219dce1b574956f1c4e" 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 +5286,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 +5891,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 +6185,7 @@ dependencies = [ "prometheus", "proxy-header", "reqwest 0.12.28", + "rustls", "semver", "tempfile", "tokio", @@ -6438,6 +6487,7 @@ dependencies = [ "mz-txn-wal", "nix 0.31.3", "num_cpus", + "rustls", "serde", "tokio", "tower 0.5.3", @@ -6761,7 +6811,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 +6937,7 @@ dependencies = [ "regex", "reqwest 0.12.28", "rlimit", + "rustls", "semver", "sentry-tracing", "serde", @@ -8425,6 +8476,7 @@ dependencies = [ "postgres-protocol", "regex", "reqwest 0.12.28", + "rustls", "serde_json", "shell-words", "tempfile", @@ -8838,6 +8890,7 @@ dependencies = [ "rdkafka", "regex", "reqwest 0.12.28", + "rustls", "semver", "serde", "serde_json", @@ -9047,6 +9100,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 +10485,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 +10506,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 +10731,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" @@ -11373,7 +11441,7 @@ dependencies = [ "errno", "libc", "linux-raw-sys 0.12.1", - "windows-sys 0.61.1", + "windows-sys 0.59.0", ] [[package]] @@ -11385,6 +11453,7 @@ dependencies = [ "aws-lc-rs", "log", "once_cell", + "ring", "rustls-pki-types", "rustls-webpki", "subtle", @@ -12492,10 +12561,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 +12583,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 +12860,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 +13067,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 +13132,7 @@ dependencies = [ "pin-project-lite", "tower-layer", "tower-service", + "tracing", ] [[package]] diff --git a/Cargo.toml b/Cargo.toml index 04ccfa3b98d27..bdc21876cd6dd 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -407,7 +407,12 @@ 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 } +# 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" lru = "0.16.3" @@ -653,9 +658,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/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/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.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 bc65fb7a293bd..71b1b454212cf 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 { @@ -82,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`]. @@ -347,47 +354,100 @@ pub struct ClusterEvalContext { pub cluster: ClusterScopeContext, } -fn ld_config(api_key: &str, metrics: &Metrics) -> ld::Config { - 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())) - .build() - .expect("valid 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, + 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(read_timeout) + .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); + + let mut config = ld::ConfigBuilder::new(api_key) + .event_processor(&event_processor) + .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))?; + let ld_client = ld::Client::build(ld_config(api_key, base_uri, 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 +639,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 +692,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..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. /// @@ -657,6 +662,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 @@ -1115,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/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/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/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) 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"]))