adapter: remove ENABLE_CLUSTER_CONTROLLER and the legacy REFRESH scheduler

The cluster controller has been default-on since v26.29 and owns the replica
set of every user managed cluster. The break-glass dyncfg kept two legacy
paths reachable: the REFRESH scheduler in `cluster_scheduling.rs`, whose two
entry points already returned before doing any work while the gate was on,
and the legacy branches in the ALTER sequencer.

Delete the gate and everything only it kept alive: the scheduler module, its
coordinator plumbing (two messages, the timer, the select-loop tick, the
`cluster_scheduling_decisions` state), the two scheduler metrics, the
`cluster_check_scheduling_policies_interval` system var, and the
`ReplicaCreateDropReason::ClusterScheduling` variant.

The persisted audit vocabulary stays: `SchedulingDecisionsWithReasonsV2` and
friends are written by the controller's on-refresh path too, and old events
must remain decodable.

Author: aljoscha

doc/user/data/metrics.yml

@@ -453,28 +453,6 @@ metrics:
   help: Total number of started transactions.
   source: src/catalog/src/durable/metrics.rs
   visibility: internal
-- name: mz_check_scheduling_policies_seconds_bucket
-  help: The time each policy in `check_scheduling_policies` takes.
-  labels:
-  - le
-  - policy
-  - thread
-  source: src/adapter/src/metrics.rs
-  visibility: internal
-- name: mz_check_scheduling_policies_seconds_count
-  help: The time each policy in `check_scheduling_policies` takes.
-  labels:
-  - policy
-  - thread
-  source: src/adapter/src/metrics.rs
-  visibility: internal
-- name: mz_check_scheduling_policies_seconds_sum
-  help: The time each policy in `check_scheduling_policies` takes.
-  labels:
-  - policy
-  - thread
-  source: src/adapter/src/metrics.rs
-  visibility: internal
 - name: mz_cluster_handle_command_duration_seconds_bucket
   help: Time spent in handling commands.
   labels:
@@ -971,25 +949,6 @@ metrics:
   help: The time it takes to advance the catalog shard upper for a txns-shard write (group commits and table register/forget).
   source: src/adapter/src/metrics.rs
   visibility: internal
-- name: mz_handle_scheduling_decisions_seconds_bucket
-  help: The time `handle_scheduling_decisions` takes.
-  labels:
-  - altered_a_cluster
-  - le
-  source: src/adapter/src/metrics.rs
-  visibility: internal
-- name: mz_handle_scheduling_decisions_seconds_count
-  help: The time `handle_scheduling_decisions` takes.
-  labels:
-  - altered_a_cluster
-  source: src/adapter/src/metrics.rs
-  visibility: internal
-- name: mz_handle_scheduling_decisions_seconds_sum
-  help: The time `handle_scheduling_decisions` takes.
-  labels:
-  - altered_a_cluster
-  source: src/adapter/src/metrics.rs
-  visibility: internal
 - name: mz_index_peek_cursor_setup_seconds_bucket
   help: Time setting up cursor and literal constraints.
   labels:

misc/python/materialize/mzcompose/__init__.py

@@ -97,14 +97,9 @@ def get_minimal_system_parameters(
         "enable_refresh_every_mvs": "true",
         "enable_replacement_materialized_views": "true",
         "enable_cluster_schedule_refresh": "true",
-        # The cluster controller and background ALTER CLUSTER dyncfgs default on
-        # in current versions. Pin them explicitly so runs against older versions
-        # (which predate the flags or defaulted them off) exercise the legacy
-        # paths while current versions exercise the controller owning the
-        # managed-cluster replica set.
-        "enable_cluster_controller": (
-            "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false"
-        ),
+        # Pinned explicitly so runs against older versions (which predate the
+        # flag or defaulted it off) behave like current ones, where it defaults
+        # on.
         "enable_background_alter_cluster": (
             "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false"
         ),
@@ -127,6 +122,15 @@ def get_minimal_system_parameters(
     if version < MzVersion.parse_mz("v0.163.0-dev"):
         config["enable_compute_active_dataflow_cancelation"] = "true"
 
+    # The cluster controller's break-glass gate. Removed in v26.37, where the
+    # controller runs unconditionally. Older binaries still read it, and
+    # defaulted it off before v26.29, so pin it on for them to keep mixed-version
+    # runs exercising the same path as current versions.
+    if version < MzVersion.parse_mz("v26.37.0-dev"):
+        config["enable_cluster_controller"] = (
+            "true" if version >= MzVersion.parse_mz("v26.29.0-dev") else "false"
+        )
+
     return config
 
 

misc/python/materialize/parallel_workload/action.py

@@ -3225,7 +3225,6 @@ def __init__(
             "oidc_group_role_sync_strict",
             "console_oidc_client_id",
             "console_oidc_scopes",
-            "enable_cluster_controller",
             "cluster_controller_tick_interval",
             "enable_background_alter_cluster",
             "default_cluster_reconfiguration_timeout",

src/adapter-types/src/dyncfgs.rs

@@ -353,23 +353,7 @@ pub const ENABLE_SCOPED_SYSTEM_PARAMETERS: Config<bool> = Config::new(
     "Whether per-cluster and per-replica scoped system parameters are evaluated and applied.",
 );
 
-/// Top-level gate for the cluster controller. When on, the controller owns the
-/// managed-cluster replica set and the legacy paths (the graceful 3-stage
-/// machine and `cluster_scheduling.rs`) are bypassed. The replica set cannot
-/// have two writers, so this is a clean switch, not a per-strategy toggle.
-///
-/// Defaults on. This is the break-glass switch to fall back to the legacy
-/// paths if the controller misbehaves.
-pub const ENABLE_CLUSTER_CONTROLLER: Config<bool> = Config::new(
-    "enable_cluster_controller",
-    true,
-    "Whether the cluster controller owns the managed-cluster replica set. When false, the legacy scheduling and graceful-reconfiguration paths run instead.",
-);
-
 /// Cadence of the cluster controller's reconcile tick.
-///
-/// Replaces `cluster_check_scheduling_policies_interval` once the controller is
-/// the sole owner; while the controller is dark both intervals exist.
 pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config<Duration> = Config::new(
     "cluster_controller_tick_interval",
     Duration::from_secs(5),
@@ -380,9 +364,6 @@ pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config<Duration> = Config::new(
 /// controller converging in the background, or blocks the session on a
 /// wait-shim until the reconfiguration completes or its deadline passes.
 ///
-/// Only consulted while [`ENABLE_CLUSTER_CONTROLLER`] is on, when the
-/// controller owns the reconfiguration.
-///
 /// Defaults on. This is the break-glass switch back to the blocking wait-shim
 /// if returning immediately causes trouble.
 pub const ENABLE_BACKGROUND_ALTER_CLUSTER: Config<bool> = Config::new(
@@ -404,9 +385,9 @@ pub const DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT: Config<Duration> = Config::ne
 /// runs a burst replica; graceful reconfiguration and `ON REFRESH` scheduling
 /// are unaffected.
 ///
-/// Only consulted while [`ENABLE_CLUSTER_CONTROLLER`] is on. A cluster can only
-/// carry an `AUTO SCALING STRATEGY` while its SQL acceptance feature flag is
-/// on, so this is the second of the two gates burst sits behind.
+/// A cluster can only carry an `AUTO SCALING STRATEGY` while its SQL acceptance
+/// feature flag is on, so this is the second of the two gates burst sits
+/// behind.
 pub const ENABLE_HYDRATION_BURST: Config<bool> = Config::new(
     "enable_hydration_burst",
     true,
@@ -426,7 +407,6 @@ pub const DEFAULT_HYDRATION_BURST_LINGER: Config<Duration> = Config::new(
 pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
     configs
         .add(&ALLOW_USER_SESSIONS)
-        .add(&ENABLE_CLUSTER_CONTROLLER)
         .add(&CLUSTER_CONTROLLER_TICK_INTERVAL)
         .add(&ENABLE_BACKGROUND_ALTER_CLUSTER)
         .add(&DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT)

src/adapter/src/catalog/open.rs

@@ -1227,9 +1227,9 @@ fn reconcile_builtin_cluster_replicas(
         }
 
         // Reading the cluster's factor is what makes this compose with the other
-        // writers of a replica set. The refresh scheduler parks a scheduled cluster
-        // by writing its factor to 0, so converging on the factor honors that
-        // instead of resurrecting a replica the scheduler just dropped.
+        // writers of a replica set. The controller's on-refresh strategy parks a
+        // scheduled cluster by writing its factor to 0, so converging on the
+        // factor honors that instead of resurrecting a replica it just dropped.
         let mut surplus = replicas_by_cluster.remove(&cluster.id).unwrap_or_default();
         for index in 0..managed.replication_factor {
             let replica_name = managed_cluster_replica_name(index);

src/adapter/src/catalog/transact.rs

@@ -86,7 +86,6 @@ use crate::catalog::{
 use crate::config::{ScopedParameters, ScopedParametersScope};
 use crate::coord::ConnMeta;
 use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
-use crate::coord::cluster_scheduling::SchedulingDecision;
 use crate::util::ResultExt;
 
 /// A manually injected audit event.
@@ -360,9 +359,6 @@ pub enum ReplicaCreateDropReason {
     /// - ALTERing various options on a managed cluster,
     /// - CREATE/DROP CLUSTER REPLICA on an unmanaged cluster.
     Manual,
-    /// The automated cluster scheduling initiated the replica create or drop, e.g., a
-    /// materialized view is needing a refresh on a SCHEDULE ON REFRESH cluster.
-    ClusterScheduling(Vec<SchedulingDecision>),
     /// The cluster controller's graceful-reconfiguration strategy created the replica while
     /// converging a cluster onto an in-flight `reconfiguration` target (a background
     /// `ALTER CLUSTER`).
@@ -373,11 +369,7 @@ pub enum ReplicaCreateDropReason {
     /// The cluster controller's on-refresh strategy created the replica for a refresh window on
     /// a `SCHEDULE = ON REFRESH` cluster. Audited as the `schedule` reason, carrying the tick's
     /// window decision (which MVs needed a refresh or compaction time, and the hydration-time
-    /// estimate) as the `scheduling_policies` detail, the same detail the legacy scheduler's
-    /// [`ReplicaCreateDropReason::ClusterScheduling`] records. Deliberately not that variant
-    /// itself: its legacy shape carries a per-policy `Vec` and an on/off flag for auditing
-    /// off-decisions, neither of which the controller has (controller drops are uniformly
-    /// `Retired`), and it is removed together with the legacy scheduler.
+    /// estimate) as the `scheduling_policies` detail.
     OnRefresh(RefreshWindowDecision),
     /// The cluster controller dropped the replica because the cluster's configuration no longer
     /// calls for it. The uniform reason on every controller-emitted drop (e.g. a
@@ -394,12 +386,6 @@ impl ReplicaCreateDropReason {
     ) {
         match self {
             ReplicaCreateDropReason::Manual => (CreateOrDropClusterReplicaReasonV1::Manual, None),
-            ReplicaCreateDropReason::ClusterScheduling(scheduling_decisions) => (
-                CreateOrDropClusterReplicaReasonV1::Schedule,
-                Some(SchedulingDecision::reasons_to_audit_log_reasons(
-                    &scheduling_decisions,
-                )),
-            ),
             ReplicaCreateDropReason::GracefulReconfiguration => {
                 (CreateOrDropClusterReplicaReasonV1::Reconfiguration, None)
             }
@@ -416,8 +402,8 @@ impl ReplicaCreateDropReason {
 }
 
 /// Convert the controller's on-refresh window decision into the audit log's
-/// `scheduling_policies` detail, the same shape the legacy scheduler records:
-/// ids as strings and the hydration-time estimate as an interval string.
+/// `scheduling_policies` detail: ids as strings and the hydration-time estimate
+/// as an interval string.
 fn refresh_window_decision_to_audit_log(
     decision: RefreshWindowDecision,
 ) -> SchedulingDecisionsWithReasonsV2 {
@@ -514,8 +500,8 @@ impl Catalog {
     /// status change, a fresh record, or the drop of an in-progress record.
     ///
     /// Every such movement is an audit-log transition, so a write performing
-    /// one must declare the matching intent. Status-preserving copies (legacy
-    /// paths carrying a record forward, re-targets that stay in progress with a
+    /// one must declare the matching intent. Status-preserving copies (a write
+    /// carrying a record forward, re-targets that stay in progress with a
     /// declared `Started`) and drops of already-settled records move nothing.
     fn reconfiguration_lifecycle_moved(
         old_config: &ClusterConfig,
@@ -3827,8 +3813,8 @@ mod tests {
             &unmanaged,
         ));
 
-        // Not movements: no record at all, a status-preserving copy (legacy
-        // paths carry the record forward), and dropping a settled record.
+        // Not movements: no record at all, a status-preserving copy (a write
+        // that carries the record forward), and dropping a settled record.
         assert!(!Catalog::reconfiguration_lifecycle_moved(
             &managed(None),
             &managed(None),
@@ -3985,9 +3971,8 @@ mod tests {
 
         use crate::catalog::ReplicaCreateDropReason;
 
-        // `OnRefresh` shares the `schedule` audit word with the legacy
-        // `ClusterScheduling` variant and converts the controller's window
-        // decision into the same `scheduling_policies` detail blob: ids as
+        // `OnRefresh` audits the `schedule` word and converts the controller's
+        // window decision into the `scheduling_policies` detail blob: ids as
         // strings, the hydration-time estimate as an interval string, and the
         // decision hardcoded `on` (the controller produces a create, and so
         // this detail, only for an open window).

src/adapter/src/coord.rs

@@ -195,7 +195,6 @@ use crate::coord::appends::{
     PendingWriteTxn,
 };
 use crate::coord::caught_up::CaughtUpCheckContext;
-use crate::coord::cluster_scheduling::SchedulingDecision;
 use crate::coord::id_bundle::CollectionIdBundle;
 use crate::coord::introspection::IntrospectionSubscribe;
 use crate::coord::peek::PendingPeek;
@@ -220,7 +219,6 @@ use crate::{AdapterNotice, ReadHolds, flags};
 pub(crate) mod appends;
 pub(crate) mod catalog_serving;
 pub(crate) mod cluster_controller;
-pub(crate) mod cluster_scheduling;
 pub(crate) mod consistency;
 pub(crate) mod id_bundle;
 pub(crate) mod in_memory_oracle;
@@ -444,13 +442,6 @@ pub enum Message {
     },
     DrainStatementLog,
     PrivateLinkVpcEndpointEvents(Vec<VpcEndpointEvent>),
-    CheckSchedulingPolicies,
-
-    /// Scheduling policy decisions about turning clusters On/Off.
-    /// `Vec<(policy name, Vec of decisions by the policy)>`
-    /// A cluster will be On if and only if there is at least one On decision for it.
-    /// Scheduling decisions for clusters that have `SCHEDULE = MANUAL` are ignored.
-    SchedulingDecisions(Vec<(&'static str, Vec<(ClusterId, SchedulingDecision)>)>),
 
     /// One pull/apply call from the cluster controller task, answered on the main
     /// coordinator message loop from the catalog and live controller signals.
@@ -558,8 +549,6 @@ impl Message {
             Message::DrainStatementLog => "drain_statement_log",
             Message::AlterConnectionValidationReady(..) => "alter_connection_validation_ready",
             Message::PrivateLinkVpcEndpointEvents(_) => "private_link_vpc_endpoint_events",
-            Message::CheckSchedulingPolicies => "check_scheduling_policies",
-            Message::SchedulingDecisions { .. } => "scheduling_decision",
             Message::ClusterControllerRequest(_) => "cluster_controller_request",
             Message::DeferredStatementReady => "deferred_statement_ready",
         }
@@ -2132,14 +2121,6 @@ pub struct Coordinator {
     /// a timestamp oracle backend is configured.
     timestamp_oracle_config: Option<TimestampOracleConfig>,
 
-    /// Periodically asks cluster scheduling policies to make their decisions.
-    check_cluster_scheduling_policies_interval: Interval,
-
-    /// This keeps the last On/Off decision for each cluster and each scheduling policy.
-    /// (Clusters that have been dropped or are otherwise out of scope for automatic scheduling are
-    /// periodically cleaned up from this Map.)
-    cluster_scheduling_decisions: BTreeMap<ClusterId, BTreeMap<&'static str, SchedulingDecision>>,
-
     /// When doing 0dt upgrades/in read-only mode, periodically ask all known
     /// clusters/collections whether they are caught up.
     caught_up_check_interval: Interval,
@@ -4039,13 +4020,6 @@ impl Coordinator {
                         linearize_reads_notified.set(linearize_reads_notify.notified());
                         messages.push(Message::LinearizeReads);
                     }
-                    // `tick()` on `Interval` is cancel-safe:
-                    // https://docs.rs/tokio/1.19.2/tokio/time/struct.Interval.html#cancel-safety
-                    // Receive a single command.
-                    _ = self.check_cluster_scheduling_policies_interval.tick() => {
-                        messages.push(Message::CheckSchedulingPolicies);
-                    },
-
                     // `tick()` on `Interval` is cancel-safe:
                     // https://docs.rs/tokio/1.19.2/tokio/time/struct.Interval.html#cancel-safety
                     // Receive a single command.
@@ -4926,12 +4900,6 @@ pub fn serve(
         let coord_now = now.clone();
         let advance_timelines_interval =
             tokio::time::interval(catalog.system_config().default_timestamp_interval());
-        let mut check_scheduling_policies_interval = tokio::time::interval(
-            catalog
-                .system_config()
-                .cluster_check_scheduling_policies_interval(),
-        );
-        check_scheduling_policies_interval.set_missed_tick_behavior(MissedTickBehavior::Delay);
 
         let clusters_caught_up_check_interval = if read_only_controllers {
             let dyncfgs = catalog.system_config().dyncfgs();
@@ -5120,8 +5088,6 @@ pub fn serve(
                     statement_logging: StatementLogging::new(coord_now.clone()),
                     webhook_concurrency_limit,
                     timestamp_oracle_config,
-                    check_cluster_scheduling_policies_interval: check_scheduling_policies_interval,
-                    cluster_scheduling_decisions: BTreeMap::new(),
                     caught_up_check_interval: clusters_caught_up_check_interval,
                     caught_up_check: clusters_caught_up_check,
                     installed_watch_sets: BTreeMap::new(),