dyncfg: require a ParameterScope on every Config

src/adapter/src/coord.rs

@@ -2349,20 +2349,15 @@ impl Coordinator {
         })
     }
 
-    /// Resolves the replica-local scoped overrides from the catalog working copy
-    /// into the compute controller's per-replica dyncfg layer, then re-pushes
-    /// the environment-wide compute configuration so replicas observe the new
-    /// values. Driven by the catalog implication for replica-scoped
-    /// configuration changes, and called once on bootstrap.
-    pub(crate) fn push_replica_dyncfg_overrides(&mut self) {
-        // Clone the (sparse) replica overrides so we don't hold a catalog borrow
-        // across the mutable controller calls below.
-        let replica_overrides = self
-            .catalog()
-            .state()
-            .scoped_system_parameters()
-            .replica
-            .clone();
+    /// Renders the replica-local scoped overrides in the catalog working copy as
+    /// per-replica [`ConfigUpdates`], grouped by cluster.
+    ///
+    /// Sparse: only replicas with an override are present. Parameters that are
+    /// not dyncfgs are skipped, as are values that fail to parse.
+    pub(crate) fn replica_dyncfg_overrides(
+        &self,
+    ) -> BTreeMap<ComputeInstanceId, BTreeMap<ReplicaId, ConfigUpdates>> {
+        let replica_overrides = &self.catalog().state().scoped_system_parameters().replica;
 
         let dyncfgs = self.catalog().system_config().dyncfgs();
         let mut instance_overrides: BTreeMap<
@@ -2397,22 +2392,32 @@ impl Coordinator {
             }
         }
 
+        instance_overrides
+    }
+
+    /// Resolves the replica-local scoped overrides from the catalog working copy
+    /// into the controllers' per-replica dyncfg layers, then re-pushes the
+    /// environment-wide configuration so replicas observe the new values.
+    /// Driven by the catalog implication for replica-scoped configuration
+    /// changes, and called once on bootstrap.
+    pub(crate) fn push_replica_dyncfg_overrides(&mut self) {
+        let instance_overrides = self.replica_dyncfg_overrides();
+
         // Both controllers carry a per-replica dyncfg layer, because the two
         // protocols realize configs in different worker `ConfigSet`s on
         // `clusterd`. The compute worker's `handle_update_configuration`
         // applies the pushed dyncfg updates to compute's own worker
-        // `ConfigSet` and to the shared persist client `ConfigSet`
+        // `ConfigSet`, to the shared persist client `ConfigSet`
         // (`persist_clients.cfg()`) that the co-located storage server reads
-        // from the same `Arc`, which covers persist-backed and process-global
-        // configs such as persist client tuning and `lgalloc`. Configs
-        // realized from the storage worker's own `ConfigSet` (read in its
-        // `UpdateConfiguration` handler) are reached only by the storage
-        // controller's layer.
-        self.controller
-            .compute
-            .update_replica_dyncfg_overrides(instance_overrides.clone());
+        // from the same `Arc`, and to `mz_metrics`, which covers
+        // persist-backed and process-global configs such as persist client
+        // tuning and `lgalloc`. Configs realized from the storage worker's own
+        // `ConfigSet` (read in its `UpdateConfiguration` handler) are reached
+        // only by the storage controller's layer. A third class is not pushed
+        // to a running replica at all but baked into its process configuration
+        // when the controller provisions it, which is why the overrides also go
+        // to the outer controller.
         self.controller
-            .storage
             .update_replica_dyncfg_overrides(instance_overrides);
         // Re-push the env-wide configs so existing replicas pick up their
         // (possibly changed) overrides. This also reverts a removed override:
@@ -2501,6 +2506,16 @@ impl Coordinator {
             .update_orchestrator_scheduling_config(scheduling_config);
         self.controller.update_configuration(dyncfg_updates);
 
+        // Install the replica-local scoped overrides before creating any
+        // replica below. Parts of a replica's configuration (its `TimelyConfig`,
+        // its expiration offset) are resolved once, when the controller
+        // provisions the replica, and must see its overrides at that point. The
+        // push after the creation loop cannot serve this purpose, because those
+        // values are frozen by then.
+        let replica_dyncfg_overrides = self.replica_dyncfg_overrides();
+        self.controller
+            .update_replica_dyncfg_overrides(replica_dyncfg_overrides);
+
         // Skip the credit consumption check at bootstrap under DisableClusterCreation behavior:
         // this codepath validates existing replicas at startup, not cluster creation, so it
         // must not block startup. New cluster creation is still gated by the DDL-time check.
@@ -2556,7 +2571,7 @@ impl Coordinator {
         }
 
         // Now that the compute instances and their replicas exist, push the
-        // replica-local scoped overrides into the compute controller so existing
+        // replica-local scoped overrides into the controllers so existing
         // replicas observe them at startup. The scoped (per-cluster and
         // per-replica) working copy was restored from the durable cache into
         // `CatalogState` while opening the catalog, so the last-known values are

src/adapter/src/coord/catalog_implications.rs

@@ -697,9 +697,11 @@ impl Coordinator {
         // Apply replica-scoped overrides after clusters are created (so their
         // compute instances exist) but before replicas are created below. The
         // override layer must be set before `create_replica`, so the new
-        // replica's first configuration replays with its override. The push
-        // reads the catalog working copy, which already reflects this
-        // transaction's scoped-config changes.
+        // replica's first configuration replays with its override, and so the
+        // configuration the controller freezes into the replica's process at
+        // provisioning time resolves against it. The push reads the catalog
+        // working copy, which already reflects this transaction's scoped-config
+        // changes.
         if replica_scoped_config_changed {
             self.push_replica_dyncfg_overrides();
         }

src/balancerd/src/dyncfgs.rs

@@ -13,7 +13,7 @@ use std::str::FromStr;
 use std::time::Duration;
 
 use anyhow::anyhow;
-use mz_dyncfg::{Config, ConfigSet, ConfigUpdates};
+use mz_dyncfg::{Config, ConfigSet, ConfigUpdates, ParameterScope};
 use mz_tracing::params::TracingParameters;
 use mz_tracing::{CloneableEnvFilter, SerializableDirective};
 use tracing_subscriber::filter::Directive;
@@ -27,34 +27,39 @@ pub const SIGTERM_CONNECTION_WAIT: Config<Duration> = Config::new(
     "balancerd_sigterm_connection_wait",
     Duration::from_secs(60 * 9),
     "Duration to wait after listeners closed via SIGTERM for outstanding connections to complete.",
+    ParameterScope::Environment,
 );
 
 /// Duration to wait after SIGTERM to begin shutdown of servers.
 pub const SIGTERM_LISTEN_WAIT: Config<Duration> = Config::new(
     "balancerd_sigterm_listen_wait",
     Duration::from_secs(60),
     "Duration to wait after SIGTERM to begin shutdown of servers.",
+    ParameterScope::Environment,
 );
 
 /// Whether to inject tcp proxy protocol headers to downstream http servers.
 pub const INJECT_PROXY_PROTOCOL_HEADER_HTTP: Config<bool> = Config::new(
     "balancerd_inject_proxy_protocol_header_http",
     false,
     "Whether to inject tcp proxy protocol headers to downstream http servers.",
+    ParameterScope::Environment,
 );
 
 /// Sets the filter to apply to stderr logging.
 pub const LOGGING_FILTER: Config<&str> = Config::new(
     "balancerd_log_filter",
     "info",
     "Sets the filter to apply to stderr logging.",
+    ParameterScope::Environment,
 );
 
 /// Sets the filter to apply to OpenTelemetry-backed distributed tracing.
 pub const OPENTELEMETRY_FILTER: Config<&str> = Config::new(
     "balancerd_opentelemetry_filter",
     "info",
     "Sets the filter to apply to OpenTelemetry-backed distributed tracing.",
+    ParameterScope::Environment,
 );
 
 /// Sets additional default directives to apply to stderr logging.
@@ -66,6 +71,7 @@ pub const LOGGING_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
     "Sets additional default directives to apply to stderr logging. \
     These apply to all variations of `log_filter`. Directives other than \
     `module=off` are likely incorrect. Comma separated list.",
+    ParameterScope::Environment,
 );
 
 /// Sets additional default directives to apply to OpenTelemetry-backed
@@ -79,6 +85,7 @@ pub const OPENTELEMETRY_FILTER_DEFAULTS: Config<fn() -> String> = Config::new(
     distributed tracing. \
     These apply to all variations of `opentelemetry_filter`. Directives other than \
     `module=off` are likely incorrect. Comma separated list.",
+    ParameterScope::Environment,
 );
 
 /// Sets additional default directives to apply to sentry logging. \
@@ -90,6 +97,7 @@ pub const SENTRY_FILTERS: Config<fn() -> String> = Config::new(
     "Sets additional default directives to apply to sentry logging. \
     These apply on top of a default `info` directive. Directives other than \
     `module=off` are likely incorrect. Comma separated list.",
+    ParameterScope::Environment,
 );
 
 /// Adds the full set of all balancer `Config`s.

src/compute-client/src/controller.rs

@@ -212,6 +212,12 @@ pub struct ComputeController {
     /// Updated through `ComputeController::update_configuration` calls and shared with all
     /// subcomponents of the compute controller.
     dyncfg: Arc<ConfigSet>,
+    /// The replica-local scoped overrides of [`Self::dyncfg`], by replica.
+    ///
+    /// Sparse, and kept here in addition to on the `Instance`s because replica
+    /// configuration that the controller resolves once, at replica creation,
+    /// must be read through the new replica's overrides.
+    replica_dyncfg_overrides: BTreeMap<ReplicaId, ConfigUpdates>,
 
     /// Receiver for responses produced by `Instance`s.
     response_rx: mpsc::UnboundedReceiver<ComputeControllerResponse>,
@@ -307,6 +313,7 @@ impl ComputeController {
             now,
             wallclock_lag,
             dyncfg: Arc::new(mz_dyncfgs::all_dyncfgs()),
+            replica_dyncfg_overrides: BTreeMap::new(),
             response_rx,
             response_tx,
             introspection_rx: Some(introspection_rx),
@@ -471,6 +478,7 @@ impl ComputeController {
             now: _,
             wallclock_lag: _,
             dyncfg: _,
+            replica_dyncfg_overrides: _,
             response_rx: _,
             response_tx: _,
             introspection_rx: _,
@@ -641,16 +649,22 @@ impl ComputeController {
 
     /// Replaces the per-replica dyncfg overrides for the given instances.
     ///
-    /// This only stores the overrides; callers should follow with a
-    /// configuration push (e.g. [`Self::update_configuration`]) so existing
-    /// replicas observe the new values. Instances absent from `overrides` have
-    /// their overrides cleared, so a replica that no longer has an override
-    /// reverts to the environment-wide configuration. Used by the scoped
-    /// feature flags (replica-local) layer.
+    /// This only stores the overrides, here and on the instances; callers
+    /// should follow with a configuration push (e.g.
+    /// [`Self::update_configuration`]) so existing replicas observe the new
+    /// values. Instances absent from `overrides` have their overrides cleared,
+    /// so a replica that no longer has an override reverts to the
+    /// environment-wide configuration. Used by the scoped feature flags
+    /// (replica-local) layer.
     pub fn update_replica_dyncfg_overrides(
         &mut self,
         mut overrides: BTreeMap<ComputeInstanceId, BTreeMap<ReplicaId, ConfigUpdates>>,
     ) {
+        self.replica_dyncfg_overrides = overrides
+            .values()
+            .flat_map(|replicas| replicas.iter())
+            .map(|(replica_id, updates)| (*replica_id, updates.clone()))
+            .collect();
         for (id, instance) in self.instances.iter_mut() {
             let instance_overrides = overrides.remove(id).unwrap_or_default();
             instance.call(move |i| i.update_replica_dyncfg_overrides(instance_overrides));
@@ -719,7 +733,17 @@ impl ComputeController {
             None => (false, Duration::from_secs(1)),
         };
 
-        let expiration_offset = COMPUTE_REPLICA_EXPIRATION_OFFSET.get(&self.dyncfg);
+        // Both configs below are `ParameterScope::Replica` and are resolved
+        // here, once, for the replica being created. Reading them through the
+        // new replica's scoped overrides is what makes those declarations
+        // effective: the values are frozen into `ReplicaConfig` and never
+        // re-read from the environment-wide set. The overrides for a replica
+        // created by DDL are committed in the same transaction that creates it,
+        // so they are already installed by the time we get here.
+        let overrides = self.replica_dyncfg_overrides.get(&replica_id);
+
+        let expiration_offset =
+            COMPUTE_REPLICA_EXPIRATION_OFFSET.get_with_overrides(&self.dyncfg, overrides);
 
         // Capture dictionary compression once, at replica creation, and hold it fixed for the
         // replica's lifetime (see `InstanceConfig::arrangement_dictionary_compression`). This is
@@ -728,7 +752,7 @@ impl ComputeController {
         // while the flag is enabled, so turning the flag off disables compression on new or
         // restarted replicas regardless of their configuration.
         let arrangement_dictionary_compression = ENABLE_ARRANGEMENT_DICTIONARY_COMPRESSION_ALPHA
-            .get(&self.dyncfg)
+            .get_with_overrides(&self.dyncfg, overrides)
             && config.arrangement_compression;
 
         let replica_config = ReplicaConfig {
@@ -772,6 +796,12 @@ impl ComputeController {
 
         instance.replicas.remove(&replica_id);
 
+        // The coordinator only re-pushes the override map when the scoped
+        // configuration itself changes, so a dropped replica's entry would
+        // otherwise be retained until the next such change.
+        self.replica_dyncfg_overrides.remove(&replica_id);
+
+        let instance = self.instance_mut(instance_id).expect("validated");
         instance.call(move |i| i.remove_replica(replica_id).expect("validated"));
 
         Ok(())

src/compute-client/src/controller/instance.rs

@@ -1294,6 +1294,11 @@ impl Instance {
     pub fn remove_replica(&mut self, id: ReplicaId) -> Result<(), ReplicaMissing> {
         let replica = self.replicas.remove(&id).ok_or(ReplicaMissing(id))?;
 
+        // The coordinator only re-pushes the override map when the scoped configuration itself
+        // changes, so a dropped replica's entry would otherwise be retained until the next such
+        // change.
+        self.replica_dyncfg_overrides.remove(&id);
+
         // Before dropping the replica state (and the contained input read holds), log read holds
         // that are the last line of defense against compaction of a dataflow's storage inputs. If
         // the corresponding global read hold has already been released, dropping the per-replica