mz_dyncfg/

lib.rs

// Copyright Materialize, Inc. and contributors. All rights reserved.
//
// Use of this software is governed by the Business Source License
// included in the LICENSE file.
//
// 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.

//! Dynamically updatable configuration.
//!
//! Basic usage:
//! - A type-safe static `Config` is defined near where it is used.
//! - Once in the lifetime of a process, all interesting `Config`s are
//!   registered to a `ConfigSet`. The values within a `ConfigSet` are shared,
//!   though multiple `ConfigSet`s may be created and each are completely
//!   independent (i.e. one in each unit test).
//! - A `ConfigSet` is plumbed around as necessary and may be used to get or
//!   set the value of `Config`.
//!
//! ```
//! # use mz_dyncfg::{Config, ConfigSet, ParameterScope};
//! const FOO: Config<bool> =
//!     Config::new("foo", false, "description of foo", ParameterScope::Environment);
//! fn bar(cfg: &ConfigSet) {
//!     assert_eq!(FOO.get(&cfg), false);
//! }
//! fn main() {
//!     let cfg = ConfigSet::default().add(&FOO);
//!     bar(&cfg);
//! }
//! ```
//!
//! # Design considerations for this library
//!
//! - The primary motivation is minimal boilerplate. Runtime dynamic
//!   configuration is one of the most powerful tools we have to quickly react
//!   to incidents, etc in production. Adding and using them should be easy
//!   enough that engineers feel empowered to use them generously.
//!
//!   The theoretical minimum boilerplate is 1) declare a config and 2) use a
//!   config to get/set the value. These could be combined into one step if (2)
//!   were based on global state, but that doesn't play well with testing. So
//!   instead we accomplish (2) by constructing a shared bag of config values in
//!   each `fn main`, amortizing the cost by plumbing it once to each component
//!   (not once per config).
//! - Config definitions are kept next to the usage. The common case is that a
//!   config is used in only one place and this makes it easy to see the
//!   associated documentation at the usage site. Configs that are used in
//!   multiple places may be defined in some common place as appropriate.
//! - Everything is type-safe.
//! - Secondarily: set up the ability to get and use the latest values of
//!   configs in tooling like `persistcli` and `stash debug`. As we've embraced
//!   dynamic configuration, we've ended up in a situation where it's stressful
//!   to run the read-write `persistcli admin` tooling with the defaults
//!   compiled into code, but `persistcli` doesn't have access to the vars stuff
//!   and doesn't want to instantiate a catalog impl.

use std::collections::BTreeMap;
use std::marker::PhantomData;
use std::sync::atomic::Ordering::SeqCst;
use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, AtomicUsize};
use std::sync::{Arc, RwLock};
use std::time::Duration;

use serde::{Deserialize, Serialize};
use tracing::error;

/// The scope at which a synced parameter's value may be overridden.
///
/// Every synced parameter declares its scope class as part of its definition.
/// The declaration is the single source of truth for which contexts the
/// LaunchDarkly sync loop evaluates and where the resolved value may be
/// overridden. See `doc/developer/design/20260609_scoped_feature_flags.md`.
///
/// [`Environment`] is the safe declaration, and the right one when in doubt. It
/// preserves the unscoped behavior of a single value everywhere. A finer scope
/// *enables* divergence, so declaring one asserts that divergence is both safe
/// and useful for this config. Getting that wrong introduces a way to break an
/// environment that did not previously exist, while erring toward
/// [`Environment`] only forgoes a capability. Prefer forgoing the capability.
///
/// Where a config is *realized* is therefore a necessary condition for a finer
/// scope, not a sufficient one. A config qualifies for [`Replica`] only if it is
/// realized per replica, and should be declared [`Replica`] only if it also
/// tunes that replica process's own resource usage (memory, CPU, I/O,
/// concurrency, timing) and cannot change what a dataflow produces, what reaches
/// durable state, or anything externally visible. `lgalloc`, the memory limiter,
/// the spill and pager knobs, and the timely zero-copy settings are the shape to
/// match.
///
/// In particular, declare [`Environment`] for a flag selecting a different
/// implementation or code path whose output-equivalence is an assumption rather
/// than a guarantee. Where the assumption holds, [`Environment`] costs nothing.
/// Where it does not, a per-replica rollout turns one bug into query results
/// that differ by which replica served them.
///
/// Which contexts a config can be realized in:
///
/// - A config realized inside a `clusterd` process is a [`Replica`] candidate.
///   That covers the compute and storage worker config sets and `mz_metrics`,
///   which the per-replica dyncfg push reaches. `environmentd` may read such a
///   config too, for its own process. That read legitimately sees the
///   environment-wide value.
/// - A config that `environmentd` resolves for *one specific replica*, whether
///   it ships the value there or acts on it itself, is also a candidate. The
///   read site has to resolve that replica's override, either with
///   [`Config::get_with_overrides`] or from a per-replica config set. Without
///   that, the override never takes effect and the declaration is a silent
///   no-op.
/// - A config realized in `environmentd` with no single replica in scope is
///   [`Environment`]. So is every `balancerd` config. `balancerd` syncs
///   LaunchDarkly itself, against a `balancer` context keyed by cloud provider,
///   region and build version, and has no environment, cluster or replica
///   beneath it to target.
/// - A config consumed at plan time, once per cluster, is [`Cluster`].
///
/// NOTE: a config whose value must agree across the replicas of one cluster
/// (because they render the same dataflow and their outputs are compared) is
/// [`Environment`] even when it is read on `clusterd`. Per-replica divergence in
/// *how* a dataflow is rendered is fine, in *what* it produces is not.
///
/// NOTE: persist configs are [`Environment`] as a class, even though the persist
/// client runs on `clusterd` and the per-replica push reaches its config set.
/// The same client code also runs in `environmentd`, and every copy of it acts
/// on shared durable state. A replica-scoped persist config could never reach
/// the `environmentd` client, so a rollout targeting replicas would leave a
/// shard's other writer on the old value indefinitely. [`Environment`] is the
/// only scope covering every persist client in an environment.
///
/// [`Cluster`]: ParameterScope::Cluster
/// [`Environment`]: ParameterScope::Environment
/// [`Replica`]: ParameterScope::Replica
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum ParameterScope {
    /// Environment-wide only; no cluster/replica overrides. The default, so all
    /// existing synced parameters are unchanged.
    ///
    /// NOTE: this names the coarsest targeting granularity, not the
    /// `environmentd` process. A config a process other than `environmentd`
    /// resolves for itself, with nothing finer beneath it, is `Environment`.
    Environment,
    /// Cluster-coherent: env-wide base plus per-cluster overrides. Evaluated
    /// with the `cluster` context (replica-free) and resolved at plan time via
    /// `OptimizerFeatureOverrides`. e.g. optimizer features.
    Cluster,
    /// Replica-local: env-wide base plus per-replica / per-size-family
    /// overrides. Evaluated with the `replica` context and resolved at the
    /// controller's per-replica dyncfg push. e.g. `lgalloc`, persist pager, LZ4.
    Replica,
}

impl Default for ParameterScope {
    fn default() -> Self {
        Self::DEFAULT
    }
}

impl ParameterScope {
    /// The scope applied to a parameter that does not declare one: environment-
    /// wide, i.e. no cluster/replica overrides. A `const` so it can be used in
    /// the `const` contexts (system-var constructors, the `feature_flags!`
    /// macro) where [`Default::default`] is unavailable.
    pub const DEFAULT: ParameterScope = ParameterScope::Environment;

    /// Returns the lowercase string name of this scope, as surfaced in
    /// documentation and introspection.
    pub const fn as_str(&self) -> &'static str {
        match self {
            ParameterScope::Environment => "environment",
            ParameterScope::Cluster => "cluster",
            ParameterScope::Replica => "replica",
        }
    }
}

/// A handle to a dynamically updatable configuration value.
///
/// This represents a strongly-typed named config of type `T`. It may be
/// registered to a set of such configs with [ConfigSet::add] and then later
/// used to retrieve the latest value at any time with [Self::get].
///
/// The supported types are [bool], [usize], [Duration], and [String], as well as [Option]
/// variants of these as necessary.
#[derive(Clone, Debug)]
pub struct Config<D: ConfigDefault> {
    name: &'static str,
    desc: &'static str,
    default: D,
    scope: ParameterScope,
}

impl<D: ConfigDefault> Config<D> {
    /// Constructs a handle for a config of type `T`.
    ///
    /// It is best practice, but not strictly required, for the name to be
    /// globally unique within a process.
    ///
    /// `scope` declares where the config's value may be overridden. It is a
    /// required parameter rather than a builder step so that every new config
    /// makes the choice deliberately. [`ParameterScope`] documents how to pick
    /// one from the config's read sites.
    ///
    /// TODO(cfg): Add some sort of categorization of config purpose here: e.g.
    /// limited-lifetime rollout flag, CYA, magic number that we never expect to
    /// tune, magic number that we DO expect to tune, etc. This could be used to
    /// power something like a `--future-default-flags` for CI, to replace part
    /// or all of the manually maintained list.
    ///
    /// TODO(cfg): See if we can make this more Rust-y and take these params as
    /// a struct (the obvious thing hits some issues with const combined with
    /// Drop).
    pub const fn new(
        name: &'static str,
        default: D,
        desc: &'static str,
        scope: ParameterScope,
    ) -> Self {
        Config {
            name,
            default,
            desc,
            scope,
        }
    }

    /// The name of this config.
    pub fn name(&self) -> &str {
        self.name
    }

    /// The description of this config.
    pub fn desc(&self) -> &str {
        self.desc
    }

    /// The [`ParameterScope`] of this config.
    pub fn scope(&self) -> ParameterScope {
        self.scope
    }

    /// The default value of this config.
    pub fn default(&self) -> &D {
        &self.default
    }

    /// Returns the latest value of this config within the given set.
    ///
    /// Panics if this config was not previously registered to the set.
    ///
    /// TODO(cfg): Decide if this should be a method on `ConfigSet` instead to
    /// match the precedent of `BTreeMap/HashMap::get` taking a key. It's like
    /// this initially because it was thought that the `Config` definition was
    /// the more important "noun" and also that rustfmt would maybe work better
    /// on this ordering.
    pub fn get(&self, set: &ConfigSet) -> D::ConfigType {
        D::ConfigType::from_val(self.shared(set).load())
    }

    /// Returns the value of this config within the given set, with `overrides`
    /// layered on top.
    ///
    /// This is how `environmentd` must read a [`ParameterScope::Replica`] config
    /// whose value it ships to one specific replica: the set holds the
    /// environment-wide value and `overrides` holds that replica's scoped
    /// overrides, which win. Reading such a config with [`Self::get`] instead
    /// makes its scope declaration a silent no-op.
    ///
    /// Panics if this config was not previously registered to the set. An
    /// override whose type does not match the config's is logged and ignored,
    /// rather than panicking a read site that is often on a critical path.
    pub fn get_with_overrides(
        &self,
        set: &ConfigSet,
        overrides: Option<&ConfigUpdates>,
    ) -> D::ConfigType {
        let val = self.shared(set).load();
        let val = match overrides.and_then(|o| o.updates.get(self.name)) {
            None => val,
            Some(o) if std::mem::discriminant(o) == std::mem::discriminant(&val) => o.clone(),
            Some(o) => {
                error!(
                    "override {:?} for config {} does not match its type {:?}",
                    o, self.name, val
                );
                val
            }
        };
        D::ConfigType::from_val(val)
    }

    /// Returns a handle to the value of this config in the given set.
    ///
    /// This allows users to amortize the cost of the name lookup.
    pub fn handle(&self, set: &ConfigSet) -> ConfigValHandle<D::ConfigType> {
        ConfigValHandle {
            val: self.shared(set).clone(),
            _type: PhantomData,
        }
    }

    /// Returns the shared value of this config in the given set.
    fn shared<'a>(&self, set: &'a ConfigSet) -> &'a ConfigValAtomic {
        &set.configs
            .get(self.name)
            .unwrap_or_else(|| panic!("config {} should be registered to set", self.name))
            .val
    }

    /// Parse a string value for this config.
    pub fn parse_val(&self, val: &str) -> Result<ConfigVal, String> {
        let val = D::ConfigType::parse(val)?;
        let val = Into::<ConfigVal>::into(val);
        Ok(val)
    }
}

/// A type usable as a [Config].
pub trait ConfigType: Into<ConfigVal> + Clone + Sized {
    /// Converts a type-erased enum value to this type.
    ///
    /// Panics if the enum's variant does not match this type.
    fn from_val(val: ConfigVal) -> Self;

    /// Parses this string slice into a [`ConfigType`].
    fn parse(s: &str) -> Result<Self, String>;
}

/// A trait for a type that can be used as a default for a [`Config`].
pub trait ConfigDefault: Clone {
    type ConfigType: ConfigType;

    /// Converts into the config type.
    fn into_config_type(self) -> Self::ConfigType;
}

impl<T: ConfigType> ConfigDefault for T {
    type ConfigType = T;

    fn into_config_type(self) -> T {
        self
    }
}

impl<T: ConfigType> ConfigDefault for fn() -> T {
    type ConfigType = T;

    fn into_config_type(self) -> T {
        (self)()
    }
}

/// An set of [Config]s with values that may or may not be independent of other
/// [ConfigSet]s.
///
/// When constructing a ConfigSet from scratch with [ConfigSet::default]
/// followed by [ConfigSet::add], the values added to the ConfigSet will be
/// independent of the values in all other ConfigSets.
///
/// When constructing a ConfigSet by cloning an existing ConfigSet, any values
/// cloned from the original ConfigSet will be shared with the original
/// ConfigSet. Updates to these values in one ConfigSet will be seen in the
/// other ConfigSet, and vice versa. Any value added to the new ConfigSet via
/// ConfigSet::add will be independent of values in the original ConfigSet,
/// unless the new ConfigSet is later cloned.
#[derive(Clone, Default)]
pub struct ConfigSet {
    configs: BTreeMap<String, ConfigEntry>,
}

impl ConfigSet {
    /// Adds the given config to this set.
    ///
    /// Names are required to be unique within a set, but each set is entirely
    /// independent. The same `Config` may be registered to multiple
    /// [`ConfigSet`]s and thus have independent values (e.g. imagine a unit
    /// test executing concurrently in the same process).
    ///
    /// Panics if a config with the same name has been previously registered
    /// to this set.
    pub fn add<D: ConfigDefault>(mut self, config: &Config<D>) -> Self {
        let default = config.default.clone().into_config_type();
        let default = Into::<ConfigVal>::into(default);
        let config = ConfigEntry {
            name: config.name,
            desc: config.desc,
            scope: config.scope,
            default: default.clone(),
            val: ConfigValAtomic::from(default),
        };
        if let Some(prev) = self.configs.insert(config.name.to_owned(), config) {
            panic!("{} registered twice", prev.name);
        }
        self
    }

    /// Returns the configs currently registered to this set.
    pub fn entries(&self) -> impl Iterator<Item = &ConfigEntry> {
        self.configs.values()
    }

    /// Returns the config with `name` registered to this set, if one exists.
    pub fn entry(&self, name: &str) -> Option<&ConfigEntry> {
        self.configs.get(name)
    }
}

/// An entry for a config in a [ConfigSet].
#[derive(Clone, Debug)]
pub struct ConfigEntry {
    name: &'static str,
    desc: &'static str,
    scope: ParameterScope,
    default: ConfigVal,
    val: ConfigValAtomic,
}

impl ConfigEntry {
    /// The name of this config.
    pub fn name(&self) -> &'static str {
        self.name
    }

    /// The description of this config.
    pub fn desc(&self) -> &'static str {
        self.desc
    }

    /// The [`ParameterScope`] of this config.
    pub fn scope(&self) -> ParameterScope {
        self.scope
    }

    /// The default value of this config.
    ///
    /// This value is never updated.
    pub fn default(&self) -> &ConfigVal {
        &self.default
    }

    /// Parses a string into a [`ConfigVal`] of this config's type.
    ///
    /// The type-erased analog of [`Config::parse_val`], dispatching on the
    /// variant of this entry's value.
    pub fn parse_val(&self, val: &str) -> Result<ConfigVal, String> {
        match self.default {
            ConfigVal::Bool(_) => <bool as ConfigType>::parse(val).map(Into::into),
            ConfigVal::U32(_) => <u32 as ConfigType>::parse(val).map(Into::into),
            ConfigVal::Usize(_) => <usize as ConfigType>::parse(val).map(Into::into),
            ConfigVal::OptUsize(_) => <Option<usize> as ConfigType>::parse(val).map(Into::into),
            ConfigVal::F64(_) => <f64 as ConfigType>::parse(val).map(Into::into),
            ConfigVal::String(_) => <String as ConfigType>::parse(val).map(Into::into),
            ConfigVal::OptString(_) => <Option<String> as ConfigType>::parse(val).map(Into::into),
            ConfigVal::Duration(_) => <Duration as ConfigType>::parse(val).map(Into::into),
            ConfigVal::Json(_) => <serde_json::Value as ConfigType>::parse(val).map(Into::into),
        }
    }

    /// The value of this config in the set.
    pub fn val(&self) -> ConfigVal {
        self.val.load()
    }
}

/// A handle to a configuration value in a [`ConfigSet`].
///
/// Allows users to amortize the lookup of a name within a set.
///
/// Handles can be cheaply cloned.
#[derive(Debug, Clone)]
pub struct ConfigValHandle<T> {
    val: ConfigValAtomic,
    _type: PhantomData<T>,
}

impl<T: ConfigType> ConfigValHandle<T> {
    /// Returns the latest value of this config within the set associated with
    /// the handle.
    pub fn get(&self) -> T {
        T::from_val(self.val.load())
    }

    /// Return a new handle that returns the constant value provided,
    /// generally for testing.
    pub fn disconnected<X>(value: X) -> Self
    where
        X: ConfigDefault<ConfigType = T>,
    {
        let config_val: ConfigVal = value.into_config_type().into();
        Self {
            val: config_val.into(),
            _type: Default::default(),
        }
    }
}

/// A type-erased configuration value for when set of different types are stored
/// in a collection.
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
pub enum ConfigVal {
    /// A `bool` value.
    Bool(bool),
    /// A `u32` value.
    U32(u32),
    /// A `usize` value.
    Usize(usize),
    /// An `Option<usize>` value.
    OptUsize(Option<usize>),
    /// An `f64` value.
    F64(f64),
    /// A `String` value.
    String(String),
    /// An `Option<String>` value
    OptString(Option<String>),
    /// A `Duration` value.
    Duration(Duration),
    /// A JSON value.
    #[serde(with = "serde_json_string")]
    Json(serde_json::Value),
}

/// To make `ConfigVal` compatible with non-self-describing serialization formats like bincode,
/// serialize JSON values as strings.
mod serde_json_string {
    use serde::de::{Deserialize, Deserializer, Error};
    use serde::ser::Serializer;

    pub fn serialize<S>(value: &serde_json::Value, serializer: S) -> Result<S::Ok, S::Error>
    where
        S: Serializer,
    {
        serializer.serialize_str(&value.to_string())
    }

    pub fn deserialize<'de, D>(deserializer: D) -> Result<serde_json::Value, D::Error>
    where
        D: Deserializer<'de>,
    {
        let s = String::deserialize(deserializer)?;
        serde_json::from_str(&s).map_err(D::Error::custom)
    }
}

/// An atomic version of [`ConfigVal`] to allow configuration values to be
/// shared between configuration writers and readers.
///
/// TODO(cfg): Consider moving these Arcs to be a single one around the map in
/// `ConfigSet` instead. That would mean less pointer-chasing in the common
/// case, but would remove the possibility of amortizing the name lookup via
/// [Config::handle].
#[derive(Clone, Debug)]
enum ConfigValAtomic {
    Bool(Arc<AtomicBool>),
    U32(Arc<AtomicU32>),
    Usize(Arc<AtomicUsize>),
    OptUsize(Arc<RwLock<Option<usize>>>),
    // Shared via to_bits/from_bits so we can use the atomic instead of Mutex.
    F64(Arc<AtomicU64>),
    String(Arc<RwLock<String>>),
    OptString(Arc<RwLock<Option<String>>>),
    Duration(Arc<RwLock<Duration>>),
    Json(Arc<RwLock<serde_json::Value>>),
}

impl From<ConfigVal> for ConfigValAtomic {
    fn from(val: ConfigVal) -> ConfigValAtomic {
        match val {
            ConfigVal::Bool(x) => ConfigValAtomic::Bool(Arc::new(AtomicBool::new(x))),
            ConfigVal::U32(x) => ConfigValAtomic::U32(Arc::new(AtomicU32::new(x))),
            ConfigVal::Usize(x) => ConfigValAtomic::Usize(Arc::new(AtomicUsize::new(x))),
            ConfigVal::OptUsize(x) => ConfigValAtomic::OptUsize(Arc::new(RwLock::new(x))),
            ConfigVal::F64(x) => ConfigValAtomic::F64(Arc::new(AtomicU64::new(x.to_bits()))),
            ConfigVal::String(x) => ConfigValAtomic::String(Arc::new(RwLock::new(x))),
            ConfigVal::OptString(x) => ConfigValAtomic::OptString(Arc::new(RwLock::new(x))),
            ConfigVal::Duration(x) => ConfigValAtomic::Duration(Arc::new(RwLock::new(x))),
            ConfigVal::Json(x) => ConfigValAtomic::Json(Arc::new(RwLock::new(x))),
        }
    }
}

impl ConfigValAtomic {
    fn load(&self) -> ConfigVal {
        match self {
            ConfigValAtomic::Bool(x) => ConfigVal::Bool(x.load(SeqCst)),
            ConfigValAtomic::U32(x) => ConfigVal::U32(x.load(SeqCst)),
            ConfigValAtomic::Usize(x) => ConfigVal::Usize(x.load(SeqCst)),
            ConfigValAtomic::OptUsize(x) => ConfigVal::OptUsize(*x.read().expect("lock poisoned")),
            ConfigValAtomic::F64(x) => ConfigVal::F64(f64::from_bits(x.load(SeqCst))),
            ConfigValAtomic::String(x) => {
                ConfigVal::String(x.read().expect("lock poisoned").clone())
            }
            ConfigValAtomic::OptString(x) => {
                ConfigVal::OptString(x.read().expect("lock poisoned").clone())
            }
            ConfigValAtomic::Duration(x) => ConfigVal::Duration(*x.read().expect("lock poisoned")),
            ConfigValAtomic::Json(x) => ConfigVal::Json(x.read().expect("lock poisoned").clone()),
        }
    }

    fn store(&self, val: ConfigVal) {
        match (self, val) {
            (ConfigValAtomic::Bool(x), ConfigVal::Bool(val)) => x.store(val, SeqCst),
            (ConfigValAtomic::U32(x), ConfigVal::U32(val)) => x.store(val, SeqCst),
            (ConfigValAtomic::Usize(x), ConfigVal::Usize(val)) => x.store(val, SeqCst),
            (ConfigValAtomic::OptUsize(x), ConfigVal::OptUsize(val)) => {
                *x.write().expect("lock poisoned") = val
            }
            (ConfigValAtomic::F64(x), ConfigVal::F64(val)) => x.store(val.to_bits(), SeqCst),
            (ConfigValAtomic::String(x), ConfigVal::String(val)) => {
                *x.write().expect("lock poisoned") = val
            }
            (ConfigValAtomic::OptString(x), ConfigVal::OptString(val)) => {
                *x.write().expect("lock poisoned") = val
            }
            (ConfigValAtomic::Duration(x), ConfigVal::Duration(val)) => {
                *x.write().expect("lock poisoned") = val
            }
            (ConfigValAtomic::Json(x), ConfigVal::Json(val)) => {
                *x.write().expect("lock poisoned") = val
            }
            (ConfigValAtomic::Bool(_), val)
            | (ConfigValAtomic::U32(_), val)
            | (ConfigValAtomic::Usize(_), val)
            | (ConfigValAtomic::OptUsize(_), val)
            | (ConfigValAtomic::F64(_), val)
            | (ConfigValAtomic::String(_), val)
            | (ConfigValAtomic::OptString(_), val)
            | (ConfigValAtomic::Duration(_), val)
            | (ConfigValAtomic::Json(_), val) => {
                panic!("attempted to store {val:?} value in {self:?} parameter")
            }
        }
    }
}

/// A batch of value updates to [Config]s in a [ConfigSet].
///
/// This may be sent across processes to apply the same value updates, but may not be durably
/// written down.
#[derive(Clone, Debug, Default, PartialEq, Serialize, Deserialize)]
pub struct ConfigUpdates {
    pub updates: BTreeMap<String, ConfigVal>,
}

impl ConfigUpdates {
    /// Adds an update for the given config and value.
    ///
    /// If a value of the same config has previously been added to these
    /// updates, replaces it.
    pub fn add<T, U>(&mut self, config: &Config<T>, val: U)
    where
        T: ConfigDefault,
        U: ConfigDefault<ConfigType = T::ConfigType>,
    {
        self.add_dynamic(config.name, val.into_config_type().into());
    }

    /// Adds an update for the given configuration name and value.
    ///
    /// It is the callers responsibility to ensure the value is of the
    /// appropriate type for the configuration.
    ///
    /// If a value of the same config has previously been added to these
    /// updates, replaces it.
    pub fn add_dynamic(&mut self, name: &str, val: ConfigVal) {
        self.updates.insert(name.to_owned(), val);
    }

    /// Adds the entries in `other` to `self`, with `other` taking precedence.
    pub fn extend(&mut self, mut other: Self) {
        self.updates.append(&mut other.updates)
    }

    /// Applies these config updates to the given [ConfigSet].
    ///
    /// This doesn't need to be the same set that the value updates were added
    /// from. In fact, the primary use of this is propagating config updates
    /// across processes.
    ///
    /// The value updates for any configs unknown by the given set are skipped.
    /// Ditto for config type mismatches. However, this is unexpected usage at
    /// present and so is logged to Sentry.
    pub fn apply(&self, set: &ConfigSet) {
        for (name, val) in self.updates.iter() {
            let Some(config) = set.configs.get(name) else {
                error!("config update {} {:?} not known set: {:?}", name, val, set);
                continue;
            };
            config.val.store(val.clone());
        }
    }
}

impl From<&ConfigSet> for ConfigUpdates {
    /// Captures the current value of every config in the set as a dense set of
    /// updates. Applying the result to another set seeded with the same configs
    /// reproduces this set's values, regardless of which configs the other set
    /// has previously had updated.
    fn from(set: &ConfigSet) -> Self {
        let mut updates = ConfigUpdates::default();
        for entry in set.entries() {
            updates.add_dynamic(entry.name(), entry.val());
        }
        updates
    }
}

mod impls {
    use std::num::{ParseFloatError, ParseIntError};
    use std::str::ParseBoolError;
    use std::time::Duration;

    use crate::{ConfigDefault, ConfigSet, ConfigType, ConfigVal};

    impl ConfigType for bool {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::Bool(x) => x,
                x => panic!("expected bool value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            match s {
                "on" => return Ok(true),
                "off" => return Ok(false),
                _ => {}
            }
            s.parse().map_err(|e: ParseBoolError| e.to_string())
        }
    }

    impl From<bool> for ConfigVal {
        fn from(val: bool) -> ConfigVal {
            ConfigVal::Bool(val)
        }
    }

    impl ConfigType for u32 {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::U32(x) => x,
                x => panic!("expected u32 value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            s.parse().map_err(|e: ParseIntError| e.to_string())
        }
    }

    impl From<u32> for ConfigVal {
        fn from(val: u32) -> ConfigVal {
            ConfigVal::U32(val)
        }
    }

    impl ConfigType for usize {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::Usize(x) => x,
                x => panic!("expected usize value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            s.parse().map_err(|e: ParseIntError| e.to_string())
        }
    }

    impl From<usize> for ConfigVal {
        fn from(val: usize) -> ConfigVal {
            ConfigVal::Usize(val)
        }
    }

    impl ConfigType for Option<usize> {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::OptUsize(x) => x,
                x => panic!("expected usize value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            if s.is_empty() {
                Ok(None)
            } else {
                let val = s.parse().map_err(|e: ParseIntError| e.to_string())?;
                Ok(Some(val))
            }
        }
    }

    impl From<Option<usize>> for ConfigVal {
        fn from(val: Option<usize>) -> ConfigVal {
            ConfigVal::OptUsize(val)
        }
    }

    impl ConfigType for f64 {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::F64(x) => x,
                x => panic!("expected f64 value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            s.parse().map_err(|e: ParseFloatError| e.to_string())
        }
    }

    impl From<f64> for ConfigVal {
        fn from(val: f64) -> ConfigVal {
            ConfigVal::F64(val)
        }
    }

    impl ConfigType for String {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::String(x) => x,
                x => panic!("expected String value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            Ok(s.to_string())
        }
    }

    impl From<String> for ConfigVal {
        fn from(val: String) -> ConfigVal {
            ConfigVal::String(val)
        }
    }

    impl ConfigDefault for &str {
        type ConfigType = String;

        fn into_config_type(self) -> String {
            self.into()
        }
    }

    impl ConfigType for Option<String> {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::OptString(x) => x,
                x => panic!("expected String value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            Ok(Some(s.to_string()))
        }
    }

    impl From<Option<String>> for ConfigVal {
        fn from(val: Option<String>) -> ConfigVal {
            ConfigVal::OptString(val)
        }
    }

    impl ConfigDefault for Option<&str> {
        type ConfigType = Option<String>;

        fn into_config_type(self) -> Option<String> {
            self.map(|s| s.to_string())
        }
    }

    impl ConfigType for Duration {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::Duration(x) => x,
                x => panic!("expected Duration value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            humantime::parse_duration(s).map_err(|e| e.to_string())
        }
    }

    impl From<Duration> for ConfigVal {
        fn from(val: Duration) -> ConfigVal {
            ConfigVal::Duration(val)
        }
    }

    impl ConfigType for serde_json::Value {
        fn from_val(val: ConfigVal) -> Self {
            match val {
                ConfigVal::Json(x) => x,
                x => panic!("expected JSON value got {:?}", x),
            }
        }

        fn parse(s: &str) -> Result<Self, String> {
            serde_json::from_str(s).map_err(|e| e.to_string())
        }
    }

    impl From<serde_json::Value> for ConfigVal {
        fn from(val: serde_json::Value) -> ConfigVal {
            ConfigVal::Json(val)
        }
    }

    impl std::fmt::Debug for ConfigSet {
        fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
            let ConfigSet { configs } = self;
            f.debug_map()
                .entries(configs.iter().map(|(name, val)| (name, val.val())))
                .finish()
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    use mz_ore::assert_err;

    const BOOL: Config<bool> = Config::new("bool", true, "", ParameterScope::Environment);
    const U32: Config<u32> = Config::new("u32", 4, "", ParameterScope::Environment);
    const USIZE: Config<usize> = Config::new("usize", 1, "", ParameterScope::Environment);
    const OPT_USIZE: Config<Option<usize>> =
        Config::new("opt_usize", Some(2), "", ParameterScope::Environment);
    const F64: Config<f64> = Config::new("f64", 5.0, "", ParameterScope::Environment);
    const STRING: Config<&str> = Config::new("string", "a", "", ParameterScope::Environment);
    const OPT_STRING: Config<Option<&str>> =
        Config::new("opt_string", Some("a"), "", ParameterScope::Environment);
    const DURATION: Config<Duration> = Config::new(
        "duration",
        Duration::from_nanos(3),
        "",
        ParameterScope::Environment,
    );
    const JSON: Config<fn() -> serde_json::Value> = Config::new(
        "json",
        || serde_json::json!({}),
        "",
        ParameterScope::Environment,
    );

    #[mz_ore::test]
    fn all_types() {
        let configs = ConfigSet::default()
            .add(&BOOL)
            .add(&USIZE)
            .add(&U32)
            .add(&OPT_USIZE)
            .add(&F64)
            .add(&STRING)
            .add(&OPT_STRING)
            .add(&DURATION)
            .add(&JSON);
        assert_eq!(BOOL.get(&configs), true);
        assert_eq!(U32.get(&configs), 4);
        assert_eq!(USIZE.get(&configs), 1);
        assert_eq!(OPT_USIZE.get(&configs), Some(2));
        assert_eq!(F64.get(&configs), 5.0);
        assert_eq!(STRING.get(&configs), "a");
        assert_eq!(OPT_STRING.get(&configs), Some("a".to_string()));
        assert_eq!(DURATION.get(&configs), Duration::from_nanos(3));
        assert_eq!(JSON.get(&configs), serde_json::json!({}));

        let mut updates = ConfigUpdates::default();
        updates.add(&BOOL, false);
        updates.add(&U32, 7);
        updates.add(&USIZE, 2);
        updates.add(&OPT_USIZE, None::<usize>);
        updates.add(&F64, 8.0);
        updates.add(&STRING, "b");
        updates.add(&OPT_STRING, None::<String>);
        updates.add(&DURATION, Duration::from_nanos(4));
        updates.add(&JSON, serde_json::json!({"a": 1}));
        updates.apply(&configs);

        assert_eq!(BOOL.get(&configs), false);
        assert_eq!(U32.get(&configs), 7);
        assert_eq!(USIZE.get(&configs), 2);
        assert_eq!(OPT_USIZE.get(&configs), None);
        assert_eq!(F64.get(&configs), 8.0);
        assert_eq!(STRING.get(&configs), "b");
        assert_eq!(OPT_STRING.get(&configs), None);
        assert_eq!(DURATION.get(&configs), Duration::from_nanos(4));
        assert_eq!(JSON.get(&configs), serde_json::json!({"a": 1}));
    }

    #[mz_ore::test]
    fn fn_default() {
        const BOOL_FN_DEFAULT: Config<fn() -> bool> =
            Config::new("bool", || !true, "", ParameterScope::Environment);
        const STRING_FN_DEFAULT: Config<fn() -> String> =
            Config::new("string", || "x".repeat(3), "", ParameterScope::Environment);

        const OPT_STRING_FN_DEFAULT: Config<fn() -> Option<String>> = Config::new(
            "opt_string",
            || Some("x".repeat(3)),
            "",
            ParameterScope::Environment,
        );

        let configs = ConfigSet::default()
            .add(&BOOL_FN_DEFAULT)
            .add(&STRING_FN_DEFAULT)
            .add(&OPT_STRING_FN_DEFAULT);
        assert_eq!(BOOL_FN_DEFAULT.get(&configs), false);
        assert_eq!(STRING_FN_DEFAULT.get(&configs), "xxx");
        assert_eq!(OPT_STRING_FN_DEFAULT.get(&configs), Some("xxx".to_string()));
    }

    #[mz_ore::test]
    fn config_set() {
        let c0 = ConfigSet::default().add(&USIZE);
        assert_eq!(USIZE.get(&c0), 1);
        let mut updates = ConfigUpdates::default();
        updates.add(&USIZE, 2);
        updates.apply(&c0);
        assert_eq!(USIZE.get(&c0), 2);

        // Each ConfigSet is independent, even if they contain the same set of
        // configs.
        let c1 = ConfigSet::default().add(&USIZE);
        assert_eq!(USIZE.get(&c1), 1);
        let mut updates = ConfigUpdates::default();
        updates.add(&USIZE, 3);
        updates.apply(&c1);
        assert_eq!(USIZE.get(&c1), 3);
        assert_eq!(USIZE.get(&c0), 2);

        // We can copy values from one to the other, though (envd -> clusterd).
        let mut updates = ConfigUpdates::default();
        for e in c0.entries() {
            updates.add_dynamic(e.name, e.val());
        }
        assert_eq!(USIZE.get(&c1), 3);
        updates.apply(&c1);
        assert_eq!(USIZE.get(&c1), 2);
    }

    #[mz_ore::test]
    fn get_with_overrides() {
        let configs = ConfigSet::default().add(&USIZE).add(&STRING);

        // No overrides at all, and an override map that doesn't mention the
        // config, both resolve to the set's value.
        assert_eq!(USIZE.get_with_overrides(&configs, None), 1);
        let mut overrides = ConfigUpdates::default();
        overrides.add(&STRING, "b");
        assert_eq!(USIZE.get_with_overrides(&configs, Some(&overrides)), 1);

        // An override wins over the set's value, without disturbing it.
        overrides.add(&USIZE, 2);
        assert_eq!(USIZE.get_with_overrides(&configs, Some(&overrides)), 2);
        assert_eq!(USIZE.get(&configs), 1);

        // An override of the wrong type is ignored rather than panicking.
        let mut mistyped = ConfigUpdates::default();
        mistyped.add_dynamic(USIZE.name(), ConfigVal::Bool(true));
        assert_eq!(USIZE.get_with_overrides(&configs, Some(&mistyped)), 1);
    }

    #[mz_ore::test]
    fn config_updates_extend() {
        // Regression test for database-issues#7793.
        //
        // Construct two ConfigUpdates with overlapping, but not identical, sets
        // of configs. Combine them and assert that the expected number of
        // updates is present.
        let mut u1 = {
            let c = ConfigSet::default().add(&USIZE).add(&STRING);
            let mut x = ConfigUpdates::default();
            for e in c.entries() {
                x.add_dynamic(e.name(), e.val());
            }
            x
        };
        let u2 = {
            let c = ConfigSet::default().add(&USIZE).add(&DURATION);
            let mut updates = ConfigUpdates::default();
            updates.add(&USIZE, 2);
            updates.apply(&c);
            let mut x = ConfigUpdates::default();
            for e in c.entries() {
                x.add_dynamic(e.name(), e.val());
            }
            x
        };
        assert_eq!(u1.updates.len(), 2);
        assert_eq!(u2.updates.len(), 2);
        u1.extend(u2);
        assert_eq!(u1.updates.len(), 3);

        // Assert that extend kept the correct (later) value for the overlapping
        // config.
        let c = ConfigSet::default().add(&USIZE);
        u1.apply(&c);
        assert_eq!(USIZE.get(&c), 2);
    }

    #[mz_ore::test]
    fn config_parse() {
        assert_eq!(BOOL.parse_val("true"), Ok(ConfigVal::Bool(true)));
        assert_eq!(BOOL.parse_val("on"), Ok(ConfigVal::Bool(true)));
        assert_eq!(BOOL.parse_val("false"), Ok(ConfigVal::Bool(false)));
        assert_eq!(BOOL.parse_val("off"), Ok(ConfigVal::Bool(false)));
        assert_err!(BOOL.parse_val("42"));
        assert_err!(BOOL.parse_val("66.6"));
        assert_err!(BOOL.parse_val("farragut"));
        assert_err!(BOOL.parse_val(""));
        assert_err!(BOOL.parse_val("5 s"));

        assert_err!(U32.parse_val("true"));
        assert_err!(U32.parse_val("false"));
        assert_eq!(U32.parse_val("42"), Ok(ConfigVal::U32(42)));
        assert_err!(U32.parse_val("66.6"));
        assert_err!(U32.parse_val("farragut"));
        assert_err!(U32.parse_val(""));
        assert_err!(U32.parse_val("5 s"));

        assert_err!(USIZE.parse_val("true"));
        assert_err!(USIZE.parse_val("false"));
        assert_eq!(USIZE.parse_val("42"), Ok(ConfigVal::Usize(42)));
        assert_err!(USIZE.parse_val("66.6"));
        assert_err!(USIZE.parse_val("farragut"));
        assert_err!(USIZE.parse_val(""));
        assert_err!(USIZE.parse_val("5 s"));

        assert_err!(OPT_USIZE.parse_val("true"));
        assert_err!(OPT_USIZE.parse_val("false"));
        assert_eq!(OPT_USIZE.parse_val("42"), Ok(ConfigVal::OptUsize(Some(42))));
        assert_err!(OPT_USIZE.parse_val("66.6"));
        assert_err!(OPT_USIZE.parse_val("farragut"));
        assert_eq!(OPT_USIZE.parse_val(""), Ok(ConfigVal::OptUsize(None)));
        assert_err!(OPT_USIZE.parse_val("5 s"));

        assert_err!(F64.parse_val("true"));
        assert_err!(F64.parse_val("false"));
        assert_eq!(F64.parse_val("42"), Ok(ConfigVal::F64(42.0)));
        assert_eq!(F64.parse_val("66.6"), Ok(ConfigVal::F64(66.6)));
        assert_err!(F64.parse_val("farragut"));
        assert_err!(F64.parse_val(""));
        assert_err!(F64.parse_val("5 s"));

        assert_eq!(
            STRING.parse_val("true"),
            Ok(ConfigVal::String("true".to_string()))
        );
        assert_eq!(
            STRING.parse_val("false"),
            Ok(ConfigVal::String("false".to_string()))
        );
        assert_eq!(
            STRING.parse_val("66.6"),
            Ok(ConfigVal::String("66.6".to_string()))
        );
        assert_eq!(
            STRING.parse_val("42"),
            Ok(ConfigVal::String("42".to_string()))
        );
        assert_eq!(
            STRING.parse_val("farragut"),
            Ok(ConfigVal::String("farragut".to_string()))
        );
        assert_eq!(STRING.parse_val(""), Ok(ConfigVal::String("".to_string())));
        assert_eq!(
            STRING.parse_val("5 s"),
            Ok(ConfigVal::String("5 s".to_string()))
        );

        assert_eq!(
            OPT_STRING.parse_val("true"),
            Ok(ConfigVal::OptString(Some("true".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val("false"),
            Ok(ConfigVal::OptString(Some("false".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val("66.6"),
            Ok(ConfigVal::OptString(Some("66.6".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val("42"),
            Ok(ConfigVal::OptString(Some("42".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val("farragut"),
            Ok(ConfigVal::OptString(Some("farragut".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val(""),
            Ok(ConfigVal::OptString(Some("".to_string())))
        );
        assert_eq!(
            OPT_STRING.parse_val("5 s"),
            Ok(ConfigVal::OptString(Some("5 s".to_string())))
        );

        assert_err!(DURATION.parse_val("true"));
        assert_err!(DURATION.parse_val("false"));
        assert_err!(DURATION.parse_val("42"));
        assert_err!(DURATION.parse_val("66.6"));
        assert_err!(DURATION.parse_val("farragut"));
        assert_err!(DURATION.parse_val(""));
        assert_eq!(
            DURATION.parse_val("5 s"),
            Ok(ConfigVal::Duration(Duration::from_secs(5)))
        );

        assert_eq!(
            JSON.parse_val("true"),
            Ok(ConfigVal::Json(serde_json::json!(true)))
        );
        assert_eq!(
            JSON.parse_val("false"),
            Ok(ConfigVal::Json(serde_json::json!(false)))
        );
        assert_eq!(
            JSON.parse_val("42"),
            Ok(ConfigVal::Json(serde_json::json!(42)))
        );
        assert_eq!(
            JSON.parse_val("66.6"),
            Ok(ConfigVal::Json(serde_json::json!(66.6)))
        );
        assert_err!(JSON.parse_val("farragut"));
        assert_err!(JSON.parse_val(""));
        assert_err!(JSON.parse_val("5 s"));
        assert_eq!(
            JSON.parse_val("{\"joe\": \"developer\"}"),
            Ok(ConfigVal::Json(serde_json::json!({"joe": "developer"})))
        );
    }
}