adapter: delete the staged cluster reconfiguration machine

With the controller owning every managed cluster's replica set,
`NeedsFinalization::Yes` has no producer. Delete the machine it drove: the
`WaitForHydrated` and `Finalize` stages, the `-pending` overlap replicas, the
`pending_cluster_alters` connection state and its retire paths, and the
`AlterClusterWhilePendingReplicas` error.

The direct reshape path is deliberately kept, as the synchronous cut-over. It
is now routed to by an explicitly zero-timeout commit strategy
(`WITH (WAIT FOR '0s')`, or `WAIT UNTIL READY (TIMEOUT '0s', ON TIMEOUT
'COMMIT')`) rather than by the absence of a `WAIT` clause. Two reasons. It is
the escape hatch: every other reshape depends on the controller ticking and
applying, and this is the one that still works when the controller itself is
the problem, while also unsticking a wedged reconfiguration by retiring its
record. And the semantics are honest: a zero timeout with commit already means
"cut over now, hydrated or not", so doing it synchronously in the ALTER is the
same outcome minus a tick.

"The same outcome" has to be true, so the cut-over does not improvise. It
folds its target onto an in-flight one exactly as the reshape path does, and it
converges the replica set with the controller's own reconcile kernel, so a
replica that already has the target shape is kept rather than bounced. Forcing
a stuck-but-hydrating resize to commit therefore keeps the replica that was
already up, and lands the record on `finalized` (forced) rather than
`cancelled`, since the cut-over reached the record's own target.

Creating a replica from the controller's `ReplicaShape` also drops a lossy
round-trip through the planner's `ComputeReplicaConfig`, which cannot represent
`INTROSPECTION DEBUGGING` without an interval. A cluster in that state used to
get a replica whose logging disagreed with the config that called for it.

Author: aljoscha

src/adapter/src/catalog/open.rs

@@ -1430,8 +1430,14 @@ fn remove_invalid_config_param_role_defaults_migration(
     Ok(())
 }
 
-/// Cluster Replicas may be created ephemerally during an alter statement, these replicas
-/// are marked as pending and should be cleaned up on catalog open.
+/// Drops replicas left durably marked `pending`.
+///
+/// No runtime path creates one anymore. An upgrade can still come from a version
+/// whose staged reconfiguration machine crashed between the pending-create commit
+/// and the finalize, and those replicas are excluded from the cluster
+/// controller's ownership test, so this catalog-open sweep is their only
+/// remaining cleaner. It goes away together with the durable `pending` field,
+/// once no supported upgrade source can still write one.
 fn remove_pending_cluster_replicas_migration(
     tx: &mut Transaction,
     boot_ts: mz_repr::Timestamp,

src/adapter/src/coord.rs

@@ -104,8 +104,8 @@ use mz_catalog::config::{AwsPrincipalContext, BuiltinItemMigrationConfig, Cluste
 use mz_catalog::durable::OpenableDurableCatalogState;
 use mz_catalog::expr_cache::{GlobalExpressions, LocalExpressions};
 use mz_catalog::memory::objects::{
-    CatalogEntry, CatalogItem, ClusterReplicaProcessStatus, ClusterVariantManaged, Connection,
-    DataSourceDesc, ReconfigurationTarget, Table, TableDataSource,
+    CatalogEntry, CatalogItem, ClusterReplicaProcessStatus, Connection, DataSourceDesc,
+    ReconfigurationTarget, Table, TableDataSource,
 };
 use mz_cloud_resources::{CloudResourceController, VpcEndpointConfig, VpcEndpointEvent};
 use mz_compute_client::as_of_selection;
@@ -153,7 +153,7 @@ use mz_sql::names::{QualifiedItemName, ResolvedIds, SchemaSpecifier};
 use mz_sql::optimizer_metrics::OptimizerMetrics;
 use mz_sql::plan::{
     self, AlterSinkPlan, ConnectionDetails, CreateConnectionPlan, HirRelationExpr,
-    NetworkPolicyRule, OnTimeoutAction, Params, QueryWhen,
+    NetworkPolicyRule, Params, QueryWhen,
 };
 use mz_sql::session::user::User;
 use mz_sql::session::vars::{MAX_CREDIT_CONSUMPTION_RATE, SystemVars, Var};
@@ -875,8 +875,6 @@ pub struct ExplainTimestampFinish {
 #[derive(Debug)]
 pub enum ClusterStage {
     Alter(AlterCluster),
-    WaitForHydrated(AlterClusterWaitForHydrated),
-    Finalize(AlterClusterFinalize),
     /// The foreground wait-shim over a controller-driven background
     /// reconfiguration: poll the durable `reconfiguration` record until it
     /// clears, then report success or timeout depending on whether the realized
@@ -890,24 +888,6 @@ pub struct AlterCluster {
     plan: plan::AlterClusterPlan,
 }
 
-#[derive(Debug)]
-pub struct AlterClusterWaitForHydrated {
-    validity: PlanValidity,
-    plan: plan::AlterClusterPlan,
-    new_config: ClusterVariantManaged,
-    workload_class: Option<String>,
-    timeout_time: Instant,
-    on_timeout: OnTimeoutAction,
-}
-
-#[derive(Debug)]
-pub struct AlterClusterFinalize {
-    validity: PlanValidity,
-    plan: plan::AlterClusterPlan,
-    new_config: ClusterVariantManaged,
-    workload_class: Option<String>,
-}
-
 #[derive(Debug)]
 pub struct AlterClusterAwaitReconfiguration {
     validity: PlanValidity,
@@ -1312,10 +1292,6 @@ pub struct ConnMeta {
     #[serde(skip)]
     deferred_lock: Option<OwnedMutexGuard<()>>,
 
-    /// Cluster reconfigurations that will need to be
-    /// cleaned up when the current transaction is cleared
-    pending_cluster_alters: BTreeSet<ClusterId>,
-
     /// Channel on which to send notices to a session.
     #[serde(skip)]
     notice_tx: mpsc::UnboundedSender<AdapterNotice>,

src/adapter/src/coord/cluster_controller.rs

@@ -363,7 +363,11 @@ impl Coordinator {
 
     /// Build the controller's view of one managed cluster from the catalog.
     /// Returns `None` for a missing or unmanaged cluster.
-    fn observe_cluster_state(&self, cluster_id: ClusterId) -> Option<ClusterState> {
+    ///
+    /// 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 {
             return None;

src/adapter/src/coord/command_handler.rs

@@ -839,7 +839,6 @@ impl Coordinator {
                     secret_key,
                     notice_tx,
                     drop_sinks: BTreeSet::new(),
-                    pending_cluster_alters: BTreeSet::new(),
                     connected_at: self.now(),
                     user,
                     application_name,
@@ -1984,8 +1983,6 @@ impl Coordinator {
         // SQL cancellation has no success response to delay. Each subscribe
         // still waits for its own retraction before it observes retirement.
         drop(retire_notify);
-        self.cancel_cluster_reconfigurations_for_conn(&conn_id)
-            .await;
         self.cancel_pending_copy(&conn_id);
         if let Some((tx, _rx)) = self.connection_cancel_watches.get_mut(&conn_id) {
             let _ = tx.send(true);

src/adapter/src/coord/ddl.rs

@@ -51,7 +51,7 @@ use serde_json::json;
 use tracing::{Instrument, Level, event, info_span, warn};
 
 use crate::active_compute_sink::{ActiveComputeSink, ActiveComputeSinkRetireReason};
-use crate::catalog::{DropObjectInfo, Op, ReplicaCreateDropReason, TransactionResult};
+use crate::catalog::{DropObjectInfo, Op, TransactionResult};
 use crate::coord::Coordinator;
 use crate::coord::appends::{BuiltinTableAppendCompletion, BuiltinTableAppendNotify};
 use crate::coord::catalog_implications::parsed_state_updates::ParsedStateUpdate;
@@ -859,43 +859,6 @@ impl Coordinator {
         }))
     }
 
-    /// Drops all pending replicas for a set of clusters
-    /// that are undergoing reconfiguration.
-    pub async fn drop_reconfiguration_replicas(
-        &mut self,
-        cluster_ids: BTreeSet<ClusterId>,
-    ) -> Result<(), AdapterError> {
-        let pending_cluster_ops: Vec<Op> = cluster_ids
-            .iter()
-            .map(|c| {
-                self.catalog()
-                    .get_cluster(c.clone())
-                    .replicas()
-                    .filter_map(|r| match r.config.location {
-                        ReplicaLocation::Managed(ref l) if l.pending => {
-                            Some(DropObjectInfo::ClusterReplica((
-                                c.clone(),
-                                r.replica_id,
-                                ReplicaCreateDropReason::Manual,
-                            )))
-                        }
-                        _ => None,
-                    })
-                    .collect::<Vec<DropObjectInfo>>()
-            })
-            .filter_map(|pending_replica_drop_ops_by_cluster| {
-                match pending_replica_drop_ops_by_cluster.len() {
-                    0 => None,
-                    _ => Some(Op::DropObjects(pending_replica_drop_ops_by_cluster)),
-                }
-            })
-            .collect();
-        if !pending_cluster_ops.is_empty() {
-            self.catalog_transact(None, pending_cluster_ops).await?;
-        }
-        Ok(())
-    }
-
     /// Cancels all active compute sinks for the identified connection.
     #[mz_ore::instrument(level = "debug")]
     pub(crate) async fn cancel_compute_sinks_for_conn(
@@ -906,15 +869,6 @@ impl Coordinator {
             .await
     }
 
-    /// Cancels all active cluster reconfigurations sinks for the identified connection.
-    #[mz_ore::instrument(level = "debug")]
-    pub(crate) async fn cancel_cluster_reconfigurations_for_conn(
-        &mut self,
-        conn_id: &ConnectionId,
-    ) {
-        self.retire_cluster_reconfigurations_for_conn(conn_id).await
-    }
-
     /// Retires all active compute sinks for the identified connection with the
     /// specified reason.
     #[mz_ore::instrument(level = "debug")]
@@ -934,30 +888,6 @@ impl Coordinator {
         self.retire_compute_sinks(drop_sinks).await
     }
 
-    /// Cleans pending cluster reconfiguraiotns for the identified connection
-    #[mz_ore::instrument(level = "debug")]
-    pub(crate) async fn retire_cluster_reconfigurations_for_conn(
-        &mut self,
-        conn_id: &ConnectionId,
-    ) {
-        let reconfiguring_clusters = self
-            .active_conns
-            .get(conn_id)
-            .expect("must exist for active session")
-            .pending_cluster_alters
-            .clone();
-        // try to drop reconfig replicas
-        self.drop_reconfiguration_replicas(reconfiguring_clusters)
-            .await
-            .unwrap_or_terminate("cannot fail to drop reconfiguration replicas");
-
-        self.active_conns
-            .get_mut(conn_id)
-            .expect("must exist for active session")
-            .pending_cluster_alters
-            .clear();
-    }
-
     pub(crate) fn drop_storage_sinks(&mut self, sink_gids: Vec<GlobalId>) {
         let storage_metadata = self.catalog.state().storage_metadata();
         self.controller