Files
buzz/crates/buzz-relay/src/feature_flags.rs
Luke Tornquist a17243d846 feat(relay): runtime-tunable history-read and AUTH concurrency caps via feature flags (#9170)
## Goal

Give operators a way to relieve database pressure in seconds, without
restarting relay pods. Today every concurrency limit is an environment
variable, so changing one means a rollout. During an incident, that
rollout's restart sets off a reconnect storm of its own.

## What changes

**Two runtime-tunable caps.** Each is read from an integer feature flag
in a new process-wide evaluation context, and re-read every 5 s:

| Flag | Caps | Ceiling |
|---|---|---|
| `buzz.relay.concurrency.history-reads` | subscription-history reads
holding a reader connection (the #9132 gate) | the existing `pool −
pool/5` share, which keeps readers free for authorization |
| `buzz.relay.concurrency.auth` | verified NIP-42 AUTH frames running
their database checks | none (unbounded today) |

- **Flags can only tighten.** A positive value is clamped to the
ceiling. Zero, a negative value, no value, or a provider that never
initialized leaves the gate at its ceiling, which is today's behavior.
Once LaunchDarkly has delivered values, they stay in force through a
later outage, so a cap tightened during an incident survives a provider
blip.
- **Lowering never cancels admitted work.** A lowered gate admits
nothing new until the work in flight falls under the new limit.
- **Visibility:** `buzz_concurrency_limit{gate}` reports the applied
limit (`0` means no limit), and each change logs a line.

**One resizable gate.** `HistoryReadGate` becomes
`buzz_db::ConcurrencyGate`, a FIFO semaphore whose limit can move
anywhere in `1..=ceiling`:
- Lowering it forgets free permits and records any shortfall as debt.
- A permit released while there is debt is retired instead of returned.
- Raising it repays debt first, then adds permits.

All of this, and admission itself, settles against the debt under one
lock, so a newer limit always supersedes an older one; there is no
background resize task to race. Tokio hands a released permit to a
queued waiter before it runs; if that waiter is then cancelled, tokio
returns the permit to the semaphore directly, so admission retires any
permit it gets while debt is outstanding. The AUTH cap uses the same
type.

**AUTH shedding.** The permit is taken after signature verification, so
only real logins queue for the database checks. The wait is bounded at 2
s. A shed login gets a bare close with no `OK`, exactly like an auth
timeout. I chose this from what each client does on main today:
- **`OK false`:** the desktop app retries on an unjittered 60 s ladder.
Mobile parks the session until the app resumes. The agent harness fails
at startup.
- **Bare close:** the desktop app, mobile, the agent harness and the CLI
all reconnect on their jittered, capped backoff.

On the relay, the shed is visible as access line
`denied/concurrency_limit` and as
`buzz_auth_outcomes_total{outcome="concurrency_limit"}`.

**The feature-flag provider is chosen at startup.**
`BUZZ_FEATURE_FLAGS_PROVIDER` takes `static` (the default),
`environment` or `launchdarkly`, with `BUZZ_LAUNCHDARKLY_SDK_KEY` and an
optional `BUZZ_LAUNCHDARKLY_RELAY_PROXY_URL`:
- **Exactly one provider.** The value is parsed into a single provider,
so there is no stacking and no fallback, and the evaluator never changes
for the life of the process. The variable sits outside the
`BUZZ_FEATURE_FLAG_` namespace, so it can never be read as a flag.
- **Startup failures.** Local misconfiguration fails startup: an unknown
provider, a missing key, or a malformed proxy URL. LaunchDarkly being
unreachable, or rejecting the key, does not. The relay starts on
declared defaults, the SDK keeps connecting, and
`buzz_feature_flags_ready{provider}` shows whether the provider has
initialized. A revoked key must not crash-loop every pod over a tuning
knob.
- **Every build carries the LaunchDarkly adapter.** It stays inert
unless selected, so one published image serves every deployment and each
deployment's configuration pins its provider.

## Design-doc amendments (`crates/buzz-feature-flags/README.md`)

- **Provider selection.** The compile-time "exactly one provider"
contract becomes the startup selection above. The only-one guarantee now
holds per process instead of per build.
- **Startup.** Failing closed now covers local configuration only.
- **Guardrails.** "Flags may choose between equivalent implementations
only" gains one exception: a flag may set an operational limit as long
as it can only tighten it, and shedding under that limit stays
retryable.
- **Evaluation context.** `EvaluationContext::for_process()` adds a
process context, which LaunchDarkly sees as a single `relay`-kind
context keyed `relay`.

## What got simpler

- **Removed the compile-time provider contract:** the nested fixture
crate (with its own 2.8k-line `Cargo.lock`), its 229-line contract test,
and the exact-one `compile_error!` guards it required consumers to copy.
One startup parse plus a unit test replaces them.
- **One gate type serves both caps.** The history-read gate became that
generic type instead of gaining a sibling.
- **Net change:** about −2.3k lines.

## Review-Proven Rules applied

- **#1:** shedding is retryable (`PoolTimedOut` → CLOSED for history, a
bare close for AUTH) and never an empty success. Provider failures are
logged and surfaced through a gauge.
- **#2:** the gate settles every limit change and permit release under
one lock. The table test covers lowering while idle, lowering below the
count held, lowering then raising before holders leave, lowering then
raising past the count held, and restoring the ceiling.
- **#3:** the tests drive the production call sites:
`Db::get_channel_window_with_session` under a lowered and raised cap
(Postgres), and `handle_auth` under a full AUTH gate. The controller
test runs the real refresh loop.
- **#4:** values are clamped, waits are bounded, and the refresh runs on
a fixed interval with skipped missed ticks.
- **#9:** AUTH shedding sends clients down their jittered reconnect
backoff rather than an unjittered or terminal path (see above).

## Review rework (`92a975c8`, `3f09cb33`)

- **Keys renamed** to `buzz.relay.concurrency.*`: LaunchDarkly keys
cannot be renamed later, and the flag project may be shared with other
services.
- **Shutdown:** LaunchDarkly's final flush has no deadline of its own,
so `close()` now runs it on a detached thread and waits at most 2 s,
after the audit drain. (The blocking pool would not bound it: the
runtime waits for blocking tasks when it drops.)
- **Refresh loop:** a panicking refresh is caught, logged and counted
(`buzz_concurrency_limit_refresh_panics_total`), and the next tick
retries, so the caps cannot freeze silently.
- **Runbook:** `docs/operator-concurrency-caps.md`, covering how to
tighten a cap and the cases in which a tightened cap returns to its
ceiling (mistyped or deleted flag, pod started during a provider
outage).
- **AUTH test:** asserts the second login passes the gate and returns
its permit; the gate permit is renamed so it no longer shares a name
with the NIP-FI effect permit.

Gated on a Linux workstation at `92a975c8` (`3f09cb33` is rustfmt only;
`cargo fmt --check` passes there): clippy `-D warnings`,
`buzz-feature-flags` (50), `buzz-relay --lib --bins` (1740), the new
modules under parallel `cargo test`, and the Postgres AUTH tests (27).
Each new test fails under its mutant: the refresh without
`catch_unwind`, `close()` with no deadline (hangs until killed), and the
AUTH handler with the gate skipped.

## Testing

Re-run in full at `3c289134` after review rework, on a Linux
workstation:

- `cargo fmt --check`; `cargo clippy -p buzz-feature-flags -p buzz-db -p
buzz-relay --all-targets --all-features -D warnings`.
- `buzz-feature-flags` with and without `launchdarkly` (50 and 24
tests), `buzz-db --lib` (209), and `buzz-relay --lib --bins` (1738),
plus a plain parallel `cargo test` of the new relay modules.
- Postgres: the `buzz-db` routed-read and history-gate tests (55), and
the `buzz-relay` AUTH and connection tests (38).
- `a_cancelled_waiter_cannot_reopen_a_lowered_gate` fails with the debt
check in `admit` removed.
- The new infra-free relay tests are added to the CI unit lane
(`Justfile` and `scripts/run-tests.sh`).
- E2E against a local relay at `06c39d64`: cap gauges, sequential logins
and history under cap 1, a 20-login burst (19 shed with a bare close,
all reconnected), and LaunchDarkly misconfiguration and unreachability.


🤖 Generated with [Claude Code](https://claude.com/claude-code)

---------

Signed-off-by: Luke Tornquist <tornquist@squareup.com>
Co-authored-by: Elrond <28d6302a099e5225b02c4155ac4236e4912603df2ab08dbfc2f4fef08ce598c8@buzz.block.builderlab.xyz>
Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
2026-11-06 14:48:06 -05:00

360 lines
14 KiB
Rust

//! The relay's feature-flag provider, chosen once at startup.
//!
//! `BUZZ_FEATURE_FLAGS_PROVIDER` selects exactly one evaluator for the life of
//! the process (see `crates/buzz-feature-flags/README.md`):
//!
//! | Value | Evaluator |
//! |-------|-----------|
//! | unset, `static` | every flag returns its declared default |
//! | `environment` | `BUZZ_FEATURE_FLAG_*` values, read once at startup |
//! | `launchdarkly` | LaunchDarkly, keyed by `BUZZ_LAUNCHDARKLY_SDK_KEY`, optionally through the Relay Proxy at `BUZZ_LAUNCHDARKLY_RELAY_PROXY_URL` |
//!
//! The variable is deliberately outside the `BUZZ_FEATURE_FLAG_` namespace the
//! environment evaluator reads, so it can never be mistaken for a flag.
//!
//! Local configuration errors (an unknown provider, a missing SDK key, a
//! malformed proxy URL) fail startup. LaunchDarkly itself being unreachable
//! does not: the relay starts on declared defaults and the SDK keeps
//! connecting in the background, because no safety invariant may depend on
//! remote flag availability. After LaunchDarkly first delivers data, a lost
//! connection leaves its last-known values in force until it reconnects.
use std::sync::Arc;
use std::time::Duration;
use buzz_feature_flags::launchdarkly::{LaunchDarklyEvaluator, LaunchDarklyRuntimeConfig};
use buzz_feature_flags::{EnvironmentEvaluator, FlagEvaluator, StaticEvaluator};
use tracing::{info, warn};
/// Environment variable naming the provider.
pub const PROVIDER_ENV: &str = "BUZZ_FEATURE_FLAGS_PROVIDER";
const SDK_KEY_ENV: &str = "BUZZ_LAUNCHDARKLY_SDK_KEY";
const RELAY_PROXY_ENV: &str = "BUZZ_LAUNCHDARKLY_RELAY_PROXY_URL";
/// How long startup waits for LaunchDarkly's first flag data before serving
/// on declared defaults.
const LAUNCHDARKLY_START_WAIT: Duration = Duration::from_secs(3);
/// How long shutdown waits for LaunchDarkly's final analytics flush. The flush
/// has no deadline of its own, and an unreachable events endpoint must not
/// spend the pod's termination grace.
const LAUNCHDARKLY_CLOSE_WAIT: Duration = Duration::from_secs(2);
/// The provider selected by [`PROVIDER_ENV`].
#[derive(Debug)]
enum Provider {
Static,
Environment,
LaunchDarkly(LaunchDarklyRuntimeConfig),
}
/// Parse the provider from `get`, which reads one environment variable.
fn provider_from(get: impl Fn(&str) -> Option<String>) -> anyhow::Result<Provider> {
let value = get(PROVIDER_ENV).unwrap_or_default();
match value.trim() {
"" | "static" => Ok(Provider::Static),
"environment" => Ok(Provider::Environment),
"launchdarkly" => {
let sdk_key = get(SDK_KEY_ENV)
.filter(|key| !key.trim().is_empty())
.ok_or_else(|| {
anyhow::anyhow!("{PROVIDER_ENV}=launchdarkly requires {SDK_KEY_ENV}")
})?;
let mut config = LaunchDarklyRuntimeConfig::new(sdk_key);
if let Some(endpoint) = get(RELAY_PROXY_ENV).filter(|url| !url.trim().is_empty()) {
config = config.with_relay_proxy_endpoint(endpoint);
}
Ok(Provider::LaunchDarkly(config))
}
other => anyhow::bail!(
"{PROVIDER_ENV}={other:?} is not a feature flag provider; \
use static, environment, or launchdarkly"
),
}
}
/// The process's one flag evaluator, plus the concrete handle its provider
/// needs for diagnostics and shutdown.
pub struct FeatureFlags {
evaluator: Arc<dyn FlagEvaluator>,
provider: &'static str,
environment: Option<Arc<EnvironmentEvaluator>>,
launchdarkly: Option<Arc<LaunchDarklyEvaluator>>,
}
impl FeatureFlags {
/// Build the provider [`PROVIDER_ENV`] selects. Fails only on local
/// configuration errors; see the module docs.
pub async fn from_env() -> anyhow::Result<Self> {
let provider = provider_from(|name| std::env::var(name).ok())?;
let flags = Self::build(provider, LAUNCHDARKLY_START_WAIT).await?;
info!(provider = flags.provider, "feature flag provider selected");
Ok(flags)
}
async fn build(provider: Provider, start_wait: Duration) -> anyhow::Result<Self> {
Ok(match provider {
Provider::Static => Self::from_evaluator("static", Arc::new(StaticEvaluator)),
Provider::Environment => {
let owner = Arc::new(EnvironmentEvaluator::from_process_environment());
Self {
evaluator: owner.clone(),
provider: "environment",
environment: Some(owner),
launchdarkly: None,
}
}
Provider::LaunchDarkly(config) => {
let owner = Arc::new(
LaunchDarklyEvaluator::from_runtime_config(config)
.map_err(|e| anyhow::anyhow!("LaunchDarkly configuration: {e}"))?,
);
if let Err(error) = owner.start_with_default_executor_and_wait(start_wait).await {
warn!(
%error,
"LaunchDarkly has not delivered flags; serving declared defaults until it does"
);
}
Self {
evaluator: owner.clone(),
provider: "launchdarkly",
environment: None,
launchdarkly: Some(owner),
}
}
})
}
/// Flags backed by `evaluator` alone (tests, embedding).
pub fn from_evaluator(provider: &'static str, evaluator: Arc<dyn FlagEvaluator>) -> Self {
Self {
evaluator,
provider,
environment: None,
launchdarkly: None,
}
}
/// The evaluator every consumer shares.
pub fn evaluator(&self) -> &dyn FlagEvaluator {
self.evaluator.as_ref()
}
/// Report provider health after a round of evaluations: log malformed
/// environment values once each, and publish whether the provider has
/// initialized (`buzz_feature_flags_ready`). It does not track the
/// connection afterwards: an initialized LaunchDarkly client keeps serving
/// its last-known values through an outage.
pub fn report(&self) {
if let Some(environment) = &self.environment {
for diagnostic in environment.take_diagnostics() {
warn!(
?diagnostic,
"malformed feature flag environment value; using its default"
);
}
}
let ready = self
.launchdarkly
.as_ref()
.is_none_or(|launchdarkly| launchdarkly.initialized());
metrics::gauge!("buzz_feature_flags_ready", "provider" => self.provider).set(if ready {
1.0
} else {
0.0
});
}
/// Stop the provider's background work, waiting at most
/// [`LAUNCHDARKLY_CLOSE_WAIT`].
pub async fn close(&self) {
self.close_within(LAUNCHDARKLY_CLOSE_WAIT).await;
}
/// LaunchDarkly's close blocks while it flushes analytics, so it runs on
/// its own detached thread: past `wait`, shutdown moves on and process
/// exit ends the flush. (The blocking pool would not do: the runtime
/// waits for its tasks when it drops.)
async fn close_within(&self, wait: Duration) {
let Some(launchdarkly) = self.launchdarkly.clone() else {
return;
};
let (closed_tx, closed_rx) = tokio::sync::oneshot::channel();
let spawned = std::thread::Builder::new()
.name("launchdarkly-close".into())
.spawn(move || {
launchdarkly.close();
let _ = closed_tx.send(());
});
if let Err(error) = spawned {
warn!(%error, "LaunchDarkly shutdown failed");
return;
}
match tokio::time::timeout(wait, closed_rx).await {
Ok(Ok(())) => {}
Ok(Err(_)) => warn!("LaunchDarkly shutdown failed"),
Err(_) => warn!(
?wait,
"LaunchDarkly did not finish its final flush; shutting down without it"
),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use std::collections::HashMap;
fn parse(pairs: &[(&str, &str)]) -> anyhow::Result<Provider> {
let env: HashMap<String, String> = pairs
.iter()
.map(|(k, v)| ((*k).to_owned(), (*v).to_owned()))
.collect();
provider_from(|name| env.get(name).cloned())
}
#[test]
fn provider_selection_is_one_explicit_choice() {
assert!(matches!(parse(&[]), Ok(Provider::Static)));
assert!(matches!(
parse(&[(PROVIDER_ENV, "static")]),
Ok(Provider::Static)
));
assert!(matches!(
parse(&[(PROVIDER_ENV, " environment ")]),
Ok(Provider::Environment)
));
// An SDK key alone never selects LaunchDarkly.
assert!(matches!(
parse(&[(SDK_KEY_ENV, "sdk-key")]),
Ok(Provider::Static)
));
let Ok(Provider::LaunchDarkly(config)) = parse(&[
(PROVIDER_ENV, "launchdarkly"),
(SDK_KEY_ENV, "sdk-key"),
(RELAY_PROXY_ENV, "http://ld-proxy.example.org:8030"),
]) else {
panic!("launchdarkly with a key selects LaunchDarkly");
};
assert_eq!(config.sdk_key, "sdk-key");
assert_eq!(
config.relay_proxy_endpoint.as_deref(),
Some("http://ld-proxy.example.org:8030")
);
let Ok(Provider::LaunchDarkly(config)) = parse(&[
(PROVIDER_ENV, "launchdarkly"),
(SDK_KEY_ENV, "sdk-key"),
(RELAY_PROXY_ENV, ""),
]) else {
panic!("an empty proxy URL means no proxy");
};
assert_eq!(config.relay_proxy_endpoint, None);
}
#[test]
fn provider_misconfiguration_fails_startup() {
for (pairs, needle) in [
(&[(PROVIDER_ENV, "launchdarkly")][..], SDK_KEY_ENV),
(
&[(PROVIDER_ENV, "launchdarkly"), (SDK_KEY_ENV, " ")][..],
SDK_KEY_ENV,
),
(
&[(PROVIDER_ENV, "LaunchDarkly")][..],
"not a feature flag provider",
),
(
&[(PROVIDER_ENV, "flagsmith")][..],
"not a feature flag provider",
),
] {
let error = parse(pairs)
.expect_err("misconfiguration must fail")
.to_string();
assert!(error.contains(needle), "{pairs:?}: {error}");
}
}
#[tokio::test]
async fn an_unreachable_launchdarkly_starts_on_defaults() {
// The relay installs its rustls provider before building flags; the
// SDK's HTTP client needs one, and several are compiled in.
let _already_installed = rustls::crypto::ring::default_provider().install_default();
// Nothing listens on the discard port, so the SDK never initializes.
let provider = parse(&[
(PROVIDER_ENV, "launchdarkly"),
(SDK_KEY_ENV, "sdk-key"),
(RELAY_PROXY_ENV, "http://127.0.0.1:9"),
])
.expect("valid configuration");
let flags = FeatureFlags::build(provider, Duration::from_millis(50))
.await
.expect("an unreachable provider must not fail startup");
let flag = buzz_feature_flags::IntegerFlag::new("buzz.relay.concurrency.auth", 0);
assert_eq!(
flags
.evaluator()
.evaluate_int(flag, &buzz_feature_flags::EvaluationContext::for_process()),
0,
"an uninitialized provider serves the declared default"
);
flags.close().await;
}
#[tokio::test]
async fn close_is_bounded_when_launchdarkly_hangs() {
let _already_installed = rustls::crypto::ring::default_provider().install_default();
// A proxy that accepts connections and never answers: every SDK
// request, the final analytics flush included, hangs.
let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
let address = listener.local_addr().expect("address");
std::thread::spawn(move || {
let mut held = Vec::new();
for stream in listener.incoming().flatten() {
held.push(stream);
}
});
let provider = parse(&[
(PROVIDER_ENV, "launchdarkly"),
(SDK_KEY_ENV, "sdk-key"),
(RELAY_PROXY_ENV, &format!("http://{address}")),
])
.expect("valid configuration");
let flags = FeatureFlags::build(provider, Duration::from_millis(50))
.await
.expect("a hanging provider must not fail startup");
// An evaluation queues an analytics event for the final flush.
let flag = buzz_feature_flags::IntegerFlag::new("buzz.relay.concurrency.auth", 0);
flags
.evaluator()
.evaluate_int(flag, &buzz_feature_flags::EvaluationContext::for_process());
let started = std::time::Instant::now();
flags.close_within(Duration::from_millis(200)).await;
assert!(
started.elapsed() < Duration::from_secs(2),
"close returns at its deadline, not when the flush gives up: {:?}",
started.elapsed()
);
}
#[tokio::test]
async fn a_malformed_proxy_url_fails_startup() {
let provider = parse(&[
(PROVIDER_ENV, "launchdarkly"),
(SDK_KEY_ENV, "sdk-key"),
(RELAY_PROXY_ENV, "ld-proxy.example.org"),
])
.expect("parses");
let Err(error) = FeatureFlags::build(provider, Duration::from_millis(50)).await else {
panic!("a relative proxy URL must fail startup");
};
assert!(
error.to_string().contains("LaunchDarkly configuration"),
"{error}"
);
}
}