cluster-controller: swap the replica set at a forced cut-over

A graceful reconfiguration overlaps the realized and target replica sets so the
old one keeps serving while the new one hydrates. A forced cut-over has given up
on hydration, so that overlap buys nothing and only doubles what the reshape has
to fit. The baseline now yields its replicas once the record is past its deadline
under ON TIMEOUT COMMIT, which turns the reshape into a single transaction that
retires the realized replicas and creates the target's. Only the net has to fit
the budget, so a resize lands on a budget with no room for both sets at once, and
shrinking a cluster that already sits near its limit works at all. If the swap
still does not fit it is rejected whole, leaving the record in progress for the
controller to shed.

Stop shedding a hydration burst on resource exhaustion. Nothing durable records
that a burst was unaffordable, so clearing the record let the unchanged policy
arm the same burst on the next tick: every cycle wrote a start and a finish,
allocated a replica id, and woke the next reconciliation immediately. An
unaffordable burst now keeps retrying its create, bounded by the tick interval,
and settles once the steady set hydrates. That drops the durable
BurstFinishCause::ResourceExhausted variant, so the stack no longer changes the
catalog protos at all.

Bound the sleeping dataflow in the cc_nonshape testdrive section. mz_sleep blocks
the worker thread and dropping the view does not interrupt a sleep already in
flight, so an hour-long sleep wedged the target replica and the section's
convergence assertions could never pass.

Author: aljoscha

src/adapter-types/src/cluster_state.rs

@@ -175,9 +175,6 @@ 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/src/catalog/transact.rs

@@ -630,7 +630,6 @@ impl Catalog {
                 let cause = match cause {
                     BurstFinishCause::LingerElapsed => BurstFinishCauseV1::LingerElapsed,
                     BurstFinishCause::NoLongerWarranted => BurstFinishCauseV1::NoLongerWarranted,
-                    BurstFinishCause::ResourceExhausted => BurstFinishCauseV1::ResourceExhausted,
                 };
                 (
                     HydrationBurstLifecycleV1::Finished,
@@ -3992,21 +3991,6 @@ 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/audit-log/src/lib.rs

@@ -522,9 +522,6 @@ pub enum BurstFinishCauseV1 {
     /// 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,
 }
 
 /// A cluster-level transition in a hydration burst's lifecycle.

src/catalog-protos/objects_hashes.json

@@ -1,7 +1,7 @@
 [
   {
     "name": "objects.rs",
-    "md5": "07f8174bfbc0f6678573152bef73da25"
+    "md5": "033aea36ed61e9181489faffbf9de51d"
   },
   {
     "name": "objects_v74.rs",
@@ -77,6 +77,6 @@
   },
   {
     "name": "objects_v92.rs",
-    "md5": "07f8174bfbc0f6678573152bef73da25"
+    "md5": "033aea36ed61e9181489faffbf9de51d"
   }
 ]

src/catalog-protos/src/audit_log.rs

@@ -431,7 +431,6 @@ impl RustType<crate::objects::audit_log_event_v1::BurstFinishCauseV1> for BurstF
         let cause = match self {
             BurstFinishCauseV1::LingerElapsed => Cause::LingerElapsed(Empty {}),
             BurstFinishCauseV1::NoLongerWarranted => Cause::NoLongerWarranted(Empty {}),
-            BurstFinishCauseV1::ResourceExhausted => Cause::ResourceExhausted(Empty {}),
         };
         crate::objects::audit_log_event_v1::BurstFinishCauseV1 { cause }
     }
@@ -443,7 +442,6 @@ impl RustType<crate::objects::audit_log_event_v1::BurstFinishCauseV1> for BurstF
         Ok(match proto.cause {
             Cause::LingerElapsed(_) => BurstFinishCauseV1::LingerElapsed,
             Cause::NoLongerWarranted(_) => BurstFinishCauseV1::NoLongerWarranted,
-            Cause::ResourceExhausted(_) => BurstFinishCauseV1::ResourceExhausted,
         })
     }
 }

src/catalog-protos/src/objects.rs

@@ -1478,7 +1478,6 @@ pub mod audit_log_event_v1 {
         pub enum Cause {
             LingerElapsed(Empty),
             NoLongerWarranted(Empty),
-            ResourceExhausted(Empty),
         }
     }
 

src/catalog-protos/src/objects_v92.rs

@@ -1478,7 +1478,6 @@ pub mod audit_log_event_v1 {
         pub enum Cause {
             LingerElapsed(Empty),
             NoLongerWarranted(Empty),
-            ResourceExhausted(Empty),
         }
     }
 

src/cluster-controller/src/lib.rs

@@ -44,9 +44,9 @@ use mz_dyncfg::ConfigSet;
 use mz_ore::soft_panic_or_log;
 
 use crate::ctx::{
-    ApplyOutcome, BurstAudit, BurstFinishCause, BurstWrite, ClusterControllerCtx, ClusterState,
-    CreateReason, Decision, ObservedReplica, ReconfigurationAudit, ReconfigurationRecord,
-    ReconfigurationStatus, ReconfigurationWrite, RefreshWindowInputs, ReplicaShape, StateWrite,
+    ApplyOutcome, ClusterControllerCtx, ClusterState, CreateReason, Decision, ObservedReplica,
+    ReconfigurationAudit, ReconfigurationRecord, ReconfigurationStatus, ReconfigurationWrite,
+    RefreshWindowInputs, ReplicaShape, StateWrite,
 };
 use crate::strategy::{
     BaselineStrategy, ConfigSignals, DesiredReplica, GracefulReconfigurationStrategy,
@@ -199,42 +199,39 @@ impl ClusterController {
         }
     }
 
-    /// The decision that sheds this cluster's most expendable transient strategy
-    /// after a resource-exhausted apply, or `None` if nothing sheddable is
-    /// active.
+    /// The decision that sheds this cluster's in-progress graceful
+    /// reconfiguration after a resource-exhausted apply, or `None` if there is
+    /// none to shed.
     ///
-    /// The strategy to shed is chosen by presence, ranked by expendability, not
-    /// by which create failed: validation is aggregate, and the strategy worth
-    /// giving up may be one whose replicas already materialized rather than one
-    /// in the failed batch. The graceful reconfiguration is the most expendable:
-    /// a discretionary user change that fails cleanly and can be retried.
-    /// Aborting it leaves the cluster running at its realized shape. A hydration
-    /// burst is next. The baseline is never shed, it is the committed floor.
+    /// A reconfiguration is a discretionary user change: it fails cleanly, it
+    /// can be retried, and abandoning it leaves the cluster running at its
+    /// realized shape. That makes it the one strategy worth giving up under
+    /// budget pressure. It is shed on presence, not on which create failed:
+    /// validation is aggregate, so the batch that failed may not be the one
+    /// whose replicas are the problem.
     ///
-    /// We shed one strategy per exhausted apply. If that was not enough, the
-    /// next tick recomputes and sheds the next one.
+    /// The baseline is the committed floor and is never shed. Neither is a
+    /// hydration burst, even though it is transient: nothing in the durable
+    /// burst state records that a burst was unaffordable, so clearing it would
+    /// let the unchanged policy arm the same burst on the next tick. Each
+    /// cycle would write a start and a finish, allocate a replica id, and wake
+    /// the next reconciliation immediately. An unaffordable burst keeps
+    /// retrying its create instead, which costs one rejected transaction per
+    /// tick and settles once the steady set hydrates.
     fn shed_decision(state: &ClusterState) -> Option<Decision> {
-        let write = match &state.reconfiguration {
-            Some(record) if record.is_in_progress() => StateWrite {
-                reconfiguration: Some(ReconfigurationWrite {
-                    record: Some(ReconfigurationRecord {
-                        status: ReconfigurationStatus::ResourceExhausted,
-                        ..record.clone()
-                    }),
-                    audit: Some(ReconfigurationAudit::ResourceExhausted),
+        let record = state
+            .reconfiguration
+            .as_ref()
+            .filter(|record| record.is_in_progress())?;
+        let write = StateWrite {
+            reconfiguration: Some(ReconfigurationWrite {
+                record: Some(ReconfigurationRecord {
+                    status: ReconfigurationStatus::ResourceExhausted,
+                    ..record.clone()
                 }),
-                ..Default::default()
-            },
-            _ if state.burst.is_some() => StateWrite {
-                burst: Some(BurstWrite {
-                    record: None,
-                    audit: Some(BurstAudit::Finished {
-                        cause: BurstFinishCause::ResourceExhausted,
-                    }),
-                }),
-                ..Default::default()
-            },
-            _ => return None,
+                audit: Some(ReconfigurationAudit::ResourceExhausted),
+            }),
+            ..Default::default()
         };
         Some(Decision::UpdateClusterState {
             cluster_id: state.cluster_id,

src/cluster-controller/src/strategy.rs

@@ -178,20 +178,51 @@ pub struct LiveSignals {
 /// contributor. (The on-refresh strategy also normalizes a scheduled cluster's
 /// `replication_factor` to `0` via `update_state`, so the two views agree after
 /// the first tick regardless.)
+///
+/// The one case where the baseline steps aside is a forced cut-over, see
+/// [`forced_cutover_pending`].
 #[derive(Clone, Copy, Debug, Default)]
 pub struct BaselineStrategy;
 
+/// Whether a forced cut-over is imminent: an in-progress reconfiguration is
+/// past its deadline under `ON TIMEOUT COMMIT`, so the next cut-over commits
+/// the target whether or not it hydrated.
+///
+/// In that window the baseline yields its realized-shape replicas. Overlapping
+/// the two sets only buys availability while the target hydrates, and a forced
+/// cut-over has given up on hydration. Yielding turns the reshape into one
+/// transaction that retires the realized replicas and creates the target's, so
+/// it has to fit the larger of the two shapes rather than their sum. That is
+/// what lets a resize succeed on a budget that has no room for overlap, and it
+/// is the only way to shrink a cluster that is already near its limit.
+///
+/// If that single transaction still does not fit, it is rejected whole and the
+/// record is left in progress for [`ClusterController::shed_decision`] to shed,
+/// so an unaffordable target stays observable rather than half-applied.
+///
+/// [`ClusterController::shed_decision`]: crate::ClusterController::shed_decision
+fn forced_cutover_pending(state: &ClusterState, now: Timestamp) -> bool {
+    state.reconfiguration.as_ref().is_some_and(|record| {
+        record.is_in_progress()
+            && now >= record.deadline
+            && matches!(record.on_timeout, OnTimeout::Commit)
+    })
+}
+
 impl Strategy for BaselineStrategy {
     fn desired_replicas(
         &self,
         state: &ClusterState,
         _signals: &LiveSignals,
         _config: &ConfigSignals,
-        _now: Timestamp,
+        now: Timestamp,
     ) -> Vec<DesiredReplica> {
         if !matches!(state.schedule, ClusterSchedule::Manual) {
             return Vec::new();
         }
+        if forced_cutover_pending(state, now) {
+            return Vec::new();
+        }
         let shape = state.realized_shape();
         (0..state.replication_factor)
             .map(|_| DesiredReplica {
@@ -212,9 +243,11 @@ impl Strategy for BaselineStrategy {
 /// record is marked finalized, and the old replicas fall out of the union and
 /// are dropped. Success takes precedence over the deadline. On a timeout,
 /// `Commit` cuts over once the complete target set exists without waiting for
-/// hydration. `Rollback` (the default) marks the record timed out without
-/// touching the realized config and stops desiring the target replicas,
-/// reverting to the pre-reconfiguration set.
+/// hydration, and the baseline stops contributing in that window so the two
+/// sets swap in one transaction rather than overlapping (see
+/// [`forced_cutover_pending`]). `Rollback` (the default) marks the record timed
+/// out without touching the realized config and stops desiring the target
+/// replicas, reverting to the pre-reconfiguration set.
 ///
 /// Both functions are pure over the observed [`ClusterState`] and the fetched
 /// [`LiveSignals`]. Hydration is requested via [`Strategy::signal_request`]
@@ -298,6 +331,9 @@ impl Strategy for GracefulReconfigurationStrategy {
         // concrete create transaction can enforce resource limits. Otherwise a
         // zero-timeout commit could finalize first, fail to create the new
         // baseline, and leave no in-progress strategy for the controller to shed.
+        // The baseline yields while we wait (see `forced_cutover_pending`), so
+        // that create arrives in the same transaction that retires the realized
+        // replicas and does not have to fit alongside them.
         let hydrated = self.target_hydrated(state, signals, record);
         let deadline_reached = now >= record.deadline;
         let commit_on_timeout = deadline_reached && matches!(record.on_timeout, OnTimeout::Commit);

src/cluster-controller/src/tests.rs

@@ -2045,7 +2045,7 @@ async fn graceful_commit_at_timeout_cuts_over_through_seam() {
 #[mz_ore::test(tokio::test)]
 async fn resource_exhaustion_sheds_a_zero_timeout_reconfiguration() {
     // The elapsed commit deadline does not finalize before the target replica
-    // exists. Phase 2 attempts the create, which reports resource exhaustion,
+    // exists. Phase 2 attempts the swap, which reports resource exhaustion,
     // then the controller sheds the still-in-progress reconfiguration.
     let c = cluster(1);
     let (state, _signals) = reconfiguring_state(
@@ -2063,11 +2063,13 @@ async fn resource_exhaustion_sheds_a_zero_timeout_reconfiguration() {
 
     controller.reconcile(&mut ctx).await;
 
-    // Two applies: the exhausted create batch, then the shed.
+    // Two applies: the exhausted swap batch, then the shed.
     assert_eq!(ctx.applied.len(), 2);
     assert!(
-        matches!(ctx.applied[0][0], Decision::CreateReplica { .. }),
-        "the exhausted batch was the target create"
+        ctx.applied[0]
+            .iter()
+            .any(|d| matches!(d, Decision::CreateReplica { .. })),
+        "the exhausted batch carries the target create"
     );
     let [shed] = &ctx.applied[1][..] else {
         panic!("the shed is a single decision, got {:?}", ctx.applied[1]);
@@ -2114,8 +2116,68 @@ async fn resource_exhaustion_sheds_a_zero_timeout_reconfiguration() {
 }
 
 #[mz_ore::test(tokio::test)]
-async fn resource_exhaustion_sheds_reconfiguration_then_burst() {
-    use crate::ctx::{BurstFinishCause, BurstRecord, OnHydrationPolicy};
+async fn forced_cutover_swaps_the_replica_set_in_one_transaction() {
+    // A forced cut-over never overlaps the two replica sets: the baseline
+    // yields, so the tick that provisions the target also retires the realized
+    // replicas, in one transaction. That is what lets a reshape land on a
+    // budget with no room for both sets at once.
+    let c = cluster(1);
+    let (state, _signals) = reconfiguring_state(
+        c,
+        "100cc",
+        1,
+        vec![observed(replica(1), "r0", "100cc")],
+        record_on_timeout("200cc", 1, 0, OnTimeout::Commit),
+        BTreeSet::new(),
+    );
+    let mut ctx = FakeCtx::new(vec![state]);
+    let controller = controller();
+
+    controller.reconcile(&mut ctx).await;
+
+    assert_eq!(ctx.applied.len(), 1, "the swap is a single apply");
+    let swap = &ctx.applied[0];
+    assert_eq!(
+        swap.iter()
+            .filter(|d| matches!(d, Decision::CreateReplica { .. }))
+            .count(),
+        1,
+        "the target replica is created"
+    );
+    assert_eq!(
+        swap.iter()
+            .filter(|d| matches!(d, Decision::DropReplica { .. }))
+            .count(),
+        1,
+        "the realized replica is retired in the same batch"
+    );
+    let sizes: Vec<_> = ctx.states[&c]
+        .replicas
+        .iter()
+        .filter_map(|r| r.owned_shape().map(|shape| shape.size.as_str()))
+        .collect();
+    assert_eq!(sizes, vec!["200cc"], "no overlap was ever materialized");
+    assert_eq!(
+        ctx.states[&c].size, "100cc",
+        "the cut-over itself waits for the next tick's first phase"
+    );
+
+    // With the target now present, the deadline commits it.
+    controller.reconcile(&mut ctx).await;
+    assert_eq!(ctx.states[&c].size, "200cc", "the forced cut-over lands");
+    assert_eq!(
+        reconfiguration_status(&ctx.states[&c]),
+        Some(ReconfigurationStatus::Finalized)
+    );
+}
+
+#[mz_ore::test(tokio::test)]
+async fn resource_exhaustion_leaves_an_unaffordable_burst_armed() {
+    // A hydration burst is not shed on exhaustion. Nothing durable would record
+    // that it was unaffordable, so the unchanged policy would arm it again on
+    // the next tick and every cycle would write a start and a finish. The burst
+    // stays armed and its create is simply retried.
+    use crate::ctx::{BurstRecord, OnHydrationPolicy};
 
     let c = cluster(1);
     let (mut state, _signals) = reconfiguring_state(
@@ -2147,30 +2209,24 @@ async fn resource_exhaustion_sheds_reconfiguration_then_burst() {
     assert_eq!(
         reconfiguration_status(&ctx.states[&c]),
         Some(ReconfigurationStatus::ResourceExhausted),
-        "the graceful reconfiguration is shed first"
+        "the graceful reconfiguration is shed"
     );
     assert!(
         ctx.states[&c].burst.is_some(),
-        "the first exhausted apply leaves the lower-priority burst active"
+        "the burst survives the shed"
     );
 
     ctx.exhaust_next = 1;
+    let before = ctx.applied.len();
     controller.reconcile(&mut ctx).await;
     assert!(
-        ctx.states[&c].burst.is_none(),
-        "a later exhausted apply clears the burst"
+        ctx.states[&c].burst.is_some(),
+        "a later exhausted apply still leaves the burst alone"
     );
-    let [shed] = &ctx.applied.last().expect("burst shed was applied")[..] else {
-        panic!("the burst shed is one decision")
-    };
-    let Decision::UpdateClusterState { write, .. } = shed else {
-        panic!("expected the burst shed to update cluster state")
-    };
     assert_eq!(
-        written_burst_audit(write),
-        Some(BurstAudit::Finished {
-            cause: BurstFinishCause::ResourceExhausted,
-        })
+        ctx.applied.len(),
+        before + 1,
+        "the exhausted create is the only apply: no shed follows it"
     );
 }