adapter: let the controller own every cluster reshape

Route zero-timeout ALTER CLUSTER requests through the same durable
reconfiguration record as every other shape change. Cluster catalog writes wake
the controller immediately, and a forced cut-over waits only for the target set
to materialize, not for it to hydrate.

This removes the sequencer's replica reconciliation path and lets the complete
strategy set decide the replica set. In particular, an active hydration-burst
replica survives a forced cut-over without a cold drop-and-recreate cycle.

Make create-time scoped configuration an invariant of the coordinator's catalog
transaction machinery. It derives contexts from concrete cluster and replica
create ops and folds the parameter update into the transaction, regardless of
whether SQL or the controller produced those ops.

Author: aljoscha

doc/developer/design/20260522_cluster_autoscaling.md

@@ -221,7 +221,7 @@ You can monitor a resize through the following:
 
 ##### Cancel a resize
 To **cancel** an in-flight resize, reissue `ALTER CLUSTER` with the cluster's
-current size. Materialize drops the pending replicas and keeps the current
+current size. Materialize drops the target replicas and keeps the current
 configuration.
 
 #### Downtime considerations for v26.34 or before

doc/user/content/sql/alter-cluster.md

@@ -175,6 +175,9 @@ pub enum BurstFinishCause {
     /// policy was removed or its hydration size changed, the cluster was
     /// turned off, or burst was disabled environment-wide.
     NoLongerWarranted,
+    /// The controller could not provision the desired replica set within the
+    /// resource budget and shed the burst.
+    ResourceExhausted,
 }
 
 /// The action a graceful reconfiguration applies once its `deadline` passes with

src/adapter-types/src/cluster_state.rs

@@ -242,11 +242,6 @@ pub enum Op {
         /// this write, alongside the record carried in `config`.
         burst_audit: Option<BurstAudit>,
     },
-    UpdateClusterReplicaConfig {
-        cluster_id: ClusterId,
-        replica_id: ReplicaId,
-        config: ReplicaConfig,
-    },
     UpdateItem {
         id: CatalogItemId,
         name: QualifiedItemName,
@@ -635,6 +630,7 @@ impl Catalog {
                 let cause = match cause {
                     BurstFinishCause::LingerElapsed => BurstFinishCauseV1::LingerElapsed,
                     BurstFinishCause::NoLongerWarranted => BurstFinishCauseV1::NoLongerWarranted,
+                    BurstFinishCause::ResourceExhausted => BurstFinishCauseV1::ResourceExhausted,
                 };
                 (
                     HydrationBurstLifecycleV1::Finished,
@@ -2928,24 +2924,6 @@ impl Catalog {
                     )?;
                 }
             }
-            Op::UpdateClusterReplicaConfig {
-                replica_id,
-                cluster_id,
-                config,
-            } => {
-                let replica = state.get_cluster_replica(cluster_id, replica_id).to_owned();
-                info!("update replica {}", replica.name);
-                tx.update_cluster_replica(
-                    replica_id,
-                    mz_catalog::durable::ClusterReplica {
-                        cluster_id,
-                        replica_id,
-                        name: replica.name.clone(),
-                        config: config.clone().into(),
-                        owner_id: replica.owner_id,
-                    },
-                )?;
-            }
             Op::UpdateItem { id, name, to_item } => {
                 // A non-temporary item must not depend on a temporary one.
                 // Temporary objects are session-scoped and disappear with
@@ -4014,6 +3992,21 @@ mod tests {
         );
         assert_eq!(finished.burst_size, "large");
 
+        let resource_exhausted = Catalog::burst_audit_details(
+            &managed(Some(record())),
+            &managed(None),
+            cluster_id,
+            "c",
+            BurstAudit::Finished {
+                cause: BurstFinishCause::ResourceExhausted,
+            },
+        )
+        .expect("resource shedding carries the cleared record's details");
+        assert_eq!(
+            resource_exhausted.finish_cause,
+            Some(BurstFinishCauseV1::ResourceExhausted)
+        );
+
         // An intent without a record on the corresponding side contradicts the
         // write and must fail the transaction.
         let incoherent = Catalog::burst_audit_details(

src/adapter/src/catalog/transact.rs

@@ -189,11 +189,11 @@ pub enum Command {
     },
 
     /// Install (or replace) the shared system-parameter frontend on the
-    /// coordinator, so the create-cluster / create-replica paths can resolve a
-    /// new object's scoped overrides synchronously, before the controller
-    /// installs it or its first dataflow is planned, rather than waiting for
-    /// the next sync tick. Sent by the sync loop whenever it (re)initializes the
-    /// frontend. See the scoped feature flags design.
+    /// coordinator, so catalog transactions can resolve a new object's scoped
+    /// overrides before its replica is provisioned or its first dataflow is
+    /// planned, rather than waiting for the next sync tick. Sent by the sync loop
+    /// whenever it (re)initializes the frontend. See the scoped feature flags
+    /// design.
     InstallScopedSystemParameterFrontend {
         frontend: Arc<SystemParameterFrontend>,
     },

src/adapter/src/command.rs

@@ -189,8 +189,9 @@ use crate::catalog::{BuiltinTableUpdate, Catalog, OpenCatalogResult};
 use crate::client::{Client, Handle};
 use crate::command::{Command, ExecuteResponse};
 use crate::config::{
-    ClusterEvalContext, ReplicaEvalContext, ScopedParameters, ScopedParametersScope,
-    SynchronizedParameters, SystemParameterFrontend, SystemParameterSyncConfig,
+    ClusterEvalContext, ClusterScopeContext, ReplicaEvalContext, ReplicaScopeContext,
+    ScopedParameters, ScopedParametersScope, SynchronizedParameters, SystemParameterFrontend,
+    SystemParameterSyncConfig,
 };
 use crate::coord::appends::{
     BuiltinTableAppendCompletion, BuiltinTableAppendNotify, DeferredOp, GroupCommitPermit,
@@ -2270,30 +2271,85 @@ impl Coordinator {
         }
     }
 
-    /// Evaluates the scoped overrides for freshly-created objects from explicit
-    /// eval contexts and returns an [`Op::UpdateScopedSystemParameters`] to fold
-    /// into the same transaction that creates them.
+    /// Evaluates scoped overrides for objects created by `ops` and returns an
+    /// [`Op::UpdateScopedSystemParameters`] to fold into the same transaction.
     ///
-    /// The objects are not yet in the catalog, so the contexts are built from
-    /// plan data and pre-allocated ids. Folding the op into the create
-    /// transaction makes its committed diff drive the replica-scoped controller
-    /// push, as a catalog implication, before `create_replica`. A new replica's
-    /// first configuration then carries its overrides rather than the env-wide
-    /// values. Render-frozen flags (e.g. the column-paged batcher, chosen at
-    /// arrangement-build time) make a later push too late, which is why this
-    /// happens in the create transaction rather than the next sync tick.
+    /// The objects are not yet in the catalog, so this derives their contexts
+    /// from the concrete create ops and pre-allocated ids. Centralizing the fold
+    /// here makes create-time configuration an invariant of coordinator-applied
+    /// catalog ops, independent of which component produced them. The committed
+    /// diff drives the replica-scoped controller push before `create_replica`.
+    /// Render-frozen flags make a later push too late.
     ///
-    /// Returns `None` when the shared frontend is not yet installed (e.g. before
-    /// LaunchDarkly connects), or when no override applies. The new objects then
-    /// resolve to the environment-wide value, and the periodic sync loop remains
-    /// the authoritative full-state reconciler.
+    /// Returns `None` when no scoped object is created or the shared frontend is
+    /// not yet installed. An installed frontend produces an op even when no
+    /// override applies, so a final DDL-transaction evaluation can clear a value
+    /// staged by an earlier statement. The periodic sync loop remains the
+    /// authoritative full-state reconciler.
     ///
     /// [`Op::UpdateScopedSystemParameters`]: crate::catalog::Op::UpdateScopedSystemParameters
-    fn scoped_overrides_create_op(
-        &self,
-        clusters: &[ClusterEvalContext],
-        replicas: &[ReplicaEvalContext],
-    ) -> Option<crate::catalog::Op> {
+    fn scoped_overrides_create_op(&self, ops: &[crate::catalog::Op]) -> Option<crate::catalog::Op> {
+        let mut created_clusters = BTreeMap::new();
+        let mut clusters = Vec::new();
+        for op in ops {
+            let crate::catalog::Op::CreateCluster { id, name, .. } = op else {
+                continue;
+            };
+            let cluster = ClusterScopeContext {
+                id: id.to_string(),
+                name: name.clone(),
+                is_builtin: id.is_system(),
+            };
+            created_clusters.insert(*id, cluster.clone());
+            clusters.push(ClusterEvalContext {
+                cluster_id: *id,
+                cluster,
+            });
+        }
+
+        let mut replicas = Vec::new();
+        for op in ops {
+            let crate::catalog::Op::CreateClusterReplica {
+                cluster_id,
+                replica_id,
+                name,
+                config,
+                ..
+            } = op
+            else {
+                continue;
+            };
+            let ReplicaLocation::Managed(location) = &config.location else {
+                continue;
+            };
+            let cluster = created_clusters.get(cluster_id).cloned().or_else(|| {
+                self.catalog()
+                    .try_get_cluster(*cluster_id)
+                    .map(|cluster| ClusterScopeContext {
+                        id: cluster_id.to_string(),
+                        name: cluster.name.clone(),
+                        is_builtin: cluster_id.is_system(),
+                    })
+            })?;
+            replicas.push(ReplicaEvalContext {
+                cluster_id: *cluster_id,
+                replica_id: *replica_id,
+                replica: ReplicaScopeContext {
+                    id: replica_id.to_string(),
+                    name: name.clone(),
+                    is_builtin: cluster_id.is_system(),
+                    size: location.size.clone(),
+                    size_family: location.allocation.family().to_string(),
+                    cluster_id: cluster_id.to_string(),
+                    cluster_name: cluster.name.clone(),
+                },
+                cluster,
+            });
+        }
+
+        if clusters.is_empty() && replicas.is_empty() {
+            return None;
+        }
         let frontend = self.scoped_frontend.clone()?;
         let catalog = self.catalog();
         let system_config = catalog.system_config();
@@ -2315,19 +2371,15 @@ impl Coordinator {
         let mut evaluated = ScopedParameters::default();
         if !cluster_param_names.is_empty() && !clusters.is_empty() {
             evaluated.cluster =
-                frontend.pull_cluster_overrides(&params, &cluster_param_names, clusters);
+                frontend.pull_cluster_overrides(&params, &cluster_param_names, &clusters);
         }
         if !replica_param_names.is_empty() && !replicas.is_empty() {
             evaluated.replica =
-                frontend.pull_replica_overrides(&params, &replica_param_names, replicas);
+                frontend.pull_replica_overrides(&params, &replica_param_names, &replicas);
         }
-        if evaluated.is_empty() {
-            return None;
-        }
-
-        // Prune only within the objects being created. They have no prior rows,
-        // so nothing is removed, and this op never touches another object whose
-        // override a concurrent reconcile may be writing.
+        // Prune only within the objects this transaction creates. A later
+        // statement in a DDL transaction can replace an earlier folded value,
+        // but this never touches an unrelated object's override.
         let prune_scope = ScopedParametersScope {
             clusters: clusters.iter().map(|cluster| cluster.cluster_id).collect(),
             replicas: replicas.iter().map(|replica| replica.replica_id).collect(),

src/adapter/src/coord.rs

@@ -371,10 +371,6 @@ impl Coordinator {
 
     /// Build the controller's view of one managed cluster from the catalog.
     /// Returns `None` for a missing or unmanaged cluster.
-    ///
-    /// Also used by the ALTER sequencer's synchronous cut-over, which runs the
-    /// controller's reconcile kernel against this same view so both paths
-    /// converge on the same replica set.
     pub(crate) fn observe_cluster_state(&self, cluster_id: ClusterId) -> Option<ClusterState> {
         let cluster = self.catalog().try_get_cluster(cluster_id)?;
         let ClusterVariant::Managed(managed) = &cluster.config.variant else {
@@ -871,9 +867,8 @@ impl Coordinator {
         })
     }
 
-    /// Build an [`Op::CreateClusterReplica`] for a desired replica `shape` on
-    /// `cluster_id` with the pre-allocated `replica_id`, attributed to `reason`.
-    /// Returns `Ok(None)` if the cluster is gone or unmanaged.
+    /// Builds a replica create op, or `Ok(None)` if the cluster is gone or
+    /// unmanaged.
     fn build_create_replica_op(
         &self,
         cluster_id: ClusterId,
@@ -920,14 +915,16 @@ impl Coordinator {
             },
         };
 
-        Ok(Some(Op::CreateClusterReplica {
+        let op = Op::CreateClusterReplica {
             cluster_id,
             replica_id,
             name,
             config,
             owner_id,
             reason,
-        }))
+        };
+
+        Ok(Some(op))
     }
 }
 

src/adapter/src/coord/cluster_controller.rs

@@ -254,7 +254,7 @@ impl Coordinator {
     pub(crate) async fn catalog_transact_with_ddl_transaction<F>(
         &mut self,
         ctx: &mut ExecuteContext,
-        ops: Vec<catalog::Op>,
+        mut ops: Vec<catalog::Op>,
         side_effect: F,
     ) -> Result<(), AdapterError>
     where
@@ -321,6 +321,21 @@ impl Coordinator {
         let prep_start = Instant::now();
         let mut combined_ops = txn_ops_clone;
         combined_ops.extend(ops.iter().cloned());
+        let creates_scoped_object = ops.iter().any(|op| {
+            matches!(
+                op,
+                catalog::Op::CreateCluster { .. } | catalog::Op::CreateClusterReplica { .. }
+            )
+        });
+        if creates_scoped_object {
+            // Include accumulated creates when deriving contexts. A replica can
+            // be created in a later DDL statement than its still-uncommitted
+            // cluster, which is absent from the coordinator's live catalog.
+            if let Some(scoped_op) = self.scoped_overrides_create_op(&combined_ops) {
+                ops.push(scoped_op.clone());
+                combined_ops.push(scoped_op);
+            }
+        }
         let conn_id = ctx.session().conn_id().clone();
         let validate_res = self.validate_resource_limits(&combined_ops, &conn_id);
         phase_seconds
@@ -384,12 +399,16 @@ impl Coordinator {
     pub(crate) async fn catalog_transact_inner(
         &mut self,
         conn_id: Option<&ConnectionId>,
-        ops: Vec<catalog::Op>,
+        mut ops: Vec<catalog::Op>,
     ) -> Result<(BuiltinTableAppendNotify, Vec<ParsedStateUpdate>), AdapterError> {
         if self.controller.read_only() {
             return Err(AdapterError::ReadOnly);
         }
 
+        if let Some(scoped_op) = self.scoped_overrides_create_op(&ops) {
+            ops.push(scoped_op);
+        }
+
         event!(Level::TRACE, ops = format!("{:?}", ops));
 
         let phase_seconds = self.metrics.catalog_transact_phase_seconds.clone();
@@ -1388,7 +1407,6 @@ impl Coordinator {
                 | Op::UpdateOwner { .. }
                 | Op::RevokeRole { .. }
                 | Op::UpdateClusterConfig { .. }
-                | Op::UpdateClusterReplicaConfig { .. }
                 | Op::UpdateSourceReferences { .. }
                 | Op::UpdateSystemConfiguration { .. }
                 | Op::ResetSystemConfiguration { .. }