[PoC] cluster-controller: gate graceful cut-over on lag, not hydration alone

doc/developer/design/20260522_cluster_autoscaling.md

@@ -31,7 +31,7 @@ Beneath both capabilities is a shared architectural problem: today, cluster sche
 - **Graceful reconfiguration mechanics** live in `src/adapter/src/coord/sequencer/inner/cluster.rs` as a three-stage state machine driven by the executor. Stage 1 creates pending replicas at the new size with a durable `pending: bool` flag. Stage 2 polls for hydration of the pending replicas. Stage 3 drops the old replicas, renames the pending ones (removing the `-pending` suffix), flips `pending` to `false`, and updates the cluster's durable `size` field.
 - **The user's intent during a graceful reconfiguration is not durable.** The `Op::UpdateClusterConfig` that sets the new cluster `size` is deliberately held back until the finalization stage. Mid-reconfig, the durable catalog shows the old `size` and a pending replica at the new size; the *intent* "the user asked for size X" lives only in transient session state (the connection's pending-alter tracker and the executor stage's strategy/timeout).
 - **Existing scheduling policy lives in `cluster_scheduling.rs`.** It runs on a coordinator timer interval, computes decisions for managed clusters with `SCHEDULE = ('on-refresh', ...)`, and sends `Message::SchedulingDecisions` to the coordinator's internal message channel, which then sequences ALTER operations. Decisions are recorded in `mz_audit_events` via `SchedulingDecisionsWithReasonsV2`. Conflicts with an in-flight graceful reconfiguration are absorbed by the scheduler swallowing the planner's `AlterClusterWhilePendingReplicas` reject — an implicit coupling the controller model replaces.
-- **Hydration signal is available in-process from the controller(s).** Per-replica, per-collection hydration state is tracked in-memory by the controller(s) and updated reactively as frontier information flows in from replicas. An in-process API already exists for asking "are all of these collections hydrated on any of these replicas?" and is used today by the graceful reconfiguration wait stage.
+- **Hydration signal is available in-process from the controller(s).** Per-replica, per-collection hydration state is tracked in-memory by the controller(s) and updated reactively as frontier information flows in from replicas. An in-process API already exists for asking "are all of these collections ready on any of these replicas?", meaning hydrated and optionally within a lag allowance of the furthest output frontier any replica reports for the collection, and is used today by the graceful reconfiguration wait stage.
 - **Audit log already records scheduling decisions with reasons** (`SchedulingDecisionsWithReasonsV2`). This is the natural place to record additional autoscaling events.
 - **Cluster configuration is fully durable in the catalog**, including `ClusterVariantManaged { size, replication_factor, availability_zones, schedule, logging, optimizer_feature_overrides }`.
 
@@ -70,7 +70,7 @@ Initial strategies. The implicit baseline is always present; the rest engage per
 
 - **Implicit baseline.** Desires the replicas implied by the realized config, `replication_factor` replicas at `cluster.size`, with the configured AZ and other cluster shape. It is what lets the policy strategies be purely additive: it holds the steady set, so they only ever add to it.
 
-- **Graceful reconfiguration.** Engaged when `ALTER` writes a durable `reconfiguration` record with `status = InProgress`. This desires `replication_factor` replicas, the target's, since an `ALTER` can change it, at the record's `target` **config shape** (target size, logging, and availability-zone list). When `update_state` observes the target replicas present and hydrated, it updates the cluster configuration (`cluster.size := target`, ...) and marks the record `Finalized`. `update_state` also reads the `deadline` and `on_timeout`. Success takes precedence: a tick that sees the target replicas hydrated cuts over even if the deadline has passed. Otherwise, once `now >= deadline` with the target not fully hydrated, `update_state` applies `on_timeout`: the default `ROLLBACK` marks the record `TimedOut` without touching the realized config and stops contributing the target replicas, so the cluster reverts to the pre-reconfiguration set and the strategy disengages; `COMMIT` instead cuts over to the still-unhydrated target and marks the record `Finalized`. Cut-over keys on **hydration**, today's graceful-reconfiguration signal, not a stronger caught-up check: hydration already guarantees correct answers, so a caught-up check would only avoid a brief post-cut-over latency bump, a possible later refinement. One consequence worth noting: an OOM- or crash-looping target replica never hydrates, so it can never cut over. The deadline fires and the default `on_timeout` reverts. No special OOM-loop detection needed.
+- **Graceful reconfiguration.** Engaged when `ALTER` writes a durable `reconfiguration` record with `status = InProgress`. This desires `replication_factor` replicas, the target's, since an `ALTER` can change it, at the record's `target` **config shape** (target size, logging, and availability-zone list). When `update_state` observes the target replicas present and **ready**, it updates the cluster configuration (`cluster.size := target`, ...) and marks the record `Finalized`. `update_state` also reads the `deadline` and `on_timeout`. Success takes precedence: a tick that sees the target replicas ready cuts over even if the deadline has passed. Otherwise, once `now >= deadline` with the target not ready, `update_state` applies `on_timeout`: the default `ROLLBACK` marks the record `TimedOut` without touching the realized config and stops contributing the target replicas, so the cluster reverts to the pre-reconfiguration set and the strategy disengages; `COMMIT` instead cuts over to the still-not-ready target and marks the record `Finalized`. Cut-over keys on **readiness**: the target replicas are hydrated *and*, per collection, within `cluster_reconfiguration_allowed_lag` of the furthest output frontier any replica of the cluster reports for it, which while the outgoing replicas are still present is theirs. Output frontiers, not write frontiers: a materialized view's write frontier is the persist shard's upper, shared by every replica writing it, and for a `REFRESH` materialized view it jumps to the next refresh time; the output frontier is each replica's own progress. The storage-side check remains hydration only. This revises the original decision to key on hydration alone. That reading held that hydration already guarantees correct answers, so a stronger check would only avoid a brief post-cut-over latency bump. Correctness is indeed unaffected either way, but the latency bump is not brief: a dataflow's as-of is pinned when its replica is added and never moves, so a collection that spends hours on its initial snapshot reports hydrated the instant that snapshot lands, with everything since the as-of still to replay. Cutting over there drops the caught-up replicas and freezes the cluster's frontiers until the new ones drain the backlog. Observed in production 2026-08-28: a 4h08m reconfiguration cut over with roughly 15 minutes of index lag remaining. `enable_cluster_reconfiguration_lag_gate` is the break-glass back to hydration alone. A consequence of the stricter gate: a target replica that can replay its backlog no faster than the outgoing replicas advance the live frontier never becomes ready, so under the default `ROLLBACK` the reconfiguration now times out where it previously cut over with a latency bump. That is most plausible for same-size reshapes (availability zone, logging, arrangement compression), where the target has no throughput advantage; the deadline, `ON TIMEOUT = COMMIT`, and the break-glass flag are the levers. A second consequence, unchanged from the hydration-only gate: an OOM- or crash-looping target replica never hydrates, so it can never cut over. The deadline fires and the default `on_timeout` reverts. No special OOM-loop detection needed.
 
 - **Hydration burst.** When the cluster's `AUTO SCALING STRATEGY` sets `ON HYDRATION (HYDRATION SIZE = ...)`, the cluster is On (`replication_factor > 0`), and there exists an object on the cluster that no realized-config replica has hydrated (zero objects warrant no burst, vacuously — a brand-new cluster does not burst at creation), `update_state` writes a `burst` record (its size and linger duration). When that record is present this desires one extra replica at `HYDRATION SIZE`. When `update_state` notices that at least one steady-state replica is hydrated, it records that timestamp. Once time has passed that timestamp plus linger duration it removes the burst record. Additionally, when there is a burst record and we recorded successful hydration of the steady-state replicas, but the steady-state replicas become un-hydrated again, we reset burst state so that the linger duration can restart after the next successful hydration. Finally, `update_state` clears the `burst` record — regardless of linger — whenever a burst is no longer warranted by current config: the `AUTO SCALING STRATEGY` was removed or its `HYDRATION SIZE` changed, or the cluster was turned off (`replication_factor = 0`). Because `desired_replicas` keys the burst replica purely on the record's presence, this cleanup is what stops a stale record from pinning a burst replica on a cluster that is off or no longer configured for burst; on a `HYDRATION SIZE` change a fresh record is written at the new size on a later tick if a burst is still warranted.
 
@@ -131,8 +131,8 @@ Take a MANUAL cluster at `100cc`, `replication_factor = 2`, serving replicas `r1
 
 1. **`ALTER` returns.** It writes `reconfiguration = { target: 200cc, deadline, status: InProgress }` and leaves `cluster.size = 100cc`.
 2. **Reconcile.** The implicit baseline desires *2 replicas at 100cc* (the realized config). Graceful reconfiguration desires *2 replicas at 200cc* (the record's target). The desired set is their union, `{2×100cc, 2×200cc}`. Actual is `{r1, r2 @ 100cc}`, matched to the 100cc slots by config. The two 200cc slots are unfilled, so the controller creates them as fresh replicas `r3`, `r4`. Actual is now `{r1, r2 @ 100cc, r3, r4 @ 200cc}`. The old and new sets overlap, and all four serve.
-3. **Hydration.** While `r3`, `r4` hydrate, the desired set is unchanged, so the controller does nothing and the `100cc` replicas keep serving.
-4. **Cut-over. The tick's `update_state` phase.** On the first tick where `r3`, `r4` are present and hydrated, `update_state` commits `cluster.size := 200cc` and marks the `reconfiguration` record `Finalized`, and the controller awaits that write before continuing.
+3. **Hydration.** While `r3`, `r4` hydrate and catch up, the desired set is unchanged, so the controller does nothing and the `100cc` replicas keep serving.
+4. **Cut-over. The tick's `update_state` phase.** On the first tick where `r3`, `r4` are present and ready (hydrated and caught up), `update_state` commits `cluster.size := 200cc` and marks the `reconfiguration` record `Finalized`, and the controller awaits that write before continuing.
 5. **Old set falls out. The same tick's `desired_replicas` phase.** With the cut-over applied, the implicit baseline now desires *2 replicas at 200cc*, matched to `r3`, `r4`, and graceful reconfiguration, with a terminal record, desires nothing. The desired set is `{2×200cc}`. `r1`, `r2` (at `100cc`) are desired by no strategy, so the controller drops them in the same tick. Actual settles at `{r3, r4 @ 200cc}`.
 
 Advancing `cluster.size` at cut-over is the single durable write that retires the old set: it flips the implicit baseline from holding the `100cc` replicas to holding the `200cc` ones, and the old replicas fall out of the union on their own. Because the desired set is matched to actual by config and count, the freshly-named `r3`, `r4` satisfy the new steady state directly. Nothing is renamed, and the cluster never drops below `replication_factor` serving replicas.
@@ -142,7 +142,7 @@ Advancing `cluster.size` at cut-over is the single durable write that retires th
 The cluster's durable configuration represents the cluster's **realized, currently-serving state**, what is actually running at steady state. We add additional records in cluster state for use by the strategies:
 
 - `auto_scaling_strategy: Option<AutoScalingStrategy>` — the strategy block (v1: `ON HYDRATION` with its `HYDRATION SIZE` and optional `LINGER DURATION`). This is user-configured *policy*, distinct from the two transition records below, which are `ALTER`/controller-managed *runtime state*.
-- `reconfiguration: Option<ReconfigurationState>`: the latest graceful reconfiguration record. It holds the `target` (the size / replication-factor / availability-zones / logging the cluster is moving to or most recently moved toward), a `deadline`, the `on_timeout` action (`COMMIT` or `ROLLBACK`) to apply if the deadline passes before the target hydrates, and a `status` (`InProgress`, `Finalized`, `TimedOut`, `Cancelled`, or `ResourceExhausted`). The strategy engages only while `status = InProgress`. Terminal records are retained for observability until overwritten by a later reconfiguration.
+- `reconfiguration: Option<ReconfigurationState>`: the latest graceful reconfiguration record. It holds the `target` (the size / replication-factor / availability-zones / logging the cluster is moving to or most recently moved toward), a `deadline`, the `on_timeout` action (`COMMIT` or `ROLLBACK`) to apply if the deadline passes before the target becomes ready, and a `status` (`InProgress`, `Finalized`, `TimedOut`, `Cancelled`, or `ResourceExhausted`). The strategy engages only while `status = InProgress`. Terminal records are retained for observability until overwritten by a later reconfiguration.
 - `burst: Option<BurstState>` — the analogous record for an active hydration burst: the `burst_size` of the in-flight burst replica, a `linger_duration`, and the timestamp at which we observed the steady-state replicas as hydrated. Burst is controller-initiated (not tied to an `ALTER`), the strategy writes the record when we determine burst is needed. It is cleared when burst tears down on success.
 
 An `ALTER CLUSTER SET (...)` that changes a replica's **config shape** writes the `reconfiguration` record with `status = InProgress` in a transaction and returns; the realized config is left untouched until the controller cuts over. Shape changes are `SIZE`, logging (`INTROSPECTION ...`), and `AVAILABILITY ZONES`. When no reconfiguration is active, changes that need no overlap (replication-factor-only, etc.) skip the record and update the realized config directly. But once an in-progress `reconfiguration` record is present, every further `ALTER` instead **folds into it**, overwriting its `target`, deadline, and status. So the realized config is advanced only by the controller at cut-over, and no direct config write ever races an in-flight transition. Re-targeting to a new non-realized shape writes `status = InProgress`. ALTER-back to the realized shape writes `status = Cancelled`, which immediately disengages the strategy and lets the target replicas fall out of the desired set. The controller's job is to converge the actual replica set onto the active target and, at cut-over, advance the realized config to match.

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

@@ -165,16 +165,16 @@ immediately.
 During a graceful resize, Materialize:
 1. Provisions new replicas at the target size, alongside the current replicas.
 2. Waits for the new replicas to
-   [hydrate](/concepts/hydration/).
+   [hydrate](/concepts/hydration/) and catch up to the current replicas.
 3. Retires the old replicas.
 
 Throughout, the cluster keeps serving queries, first from the old replicas,
 then from both sets as the new replicas come up, so the resize incurs no
 downtime.
 
-If the new replicas do not hydrate within the reconfiguration timeout (24 hours
-by default), Materialize rolls back the resize and the cluster keeps its current
-size. To customize the timeout behavior, use the `WAIT UNTIL READY` or `WAIT FOR` options.
+If the new replicas do not hydrate and catch up within the reconfiguration
+timeout (24 hours by default), Materialize rolls back the resize and the cluster
+keeps its current size. To customize the timeout behavior, use the `WAIT UNTIL READY` or `WAIT FOR` options.
 The resize still proceeds in the background.
 
 {{< private-preview >}}
@@ -183,17 +183,18 @@ Customizing the resize timeout with `WAIT UNTIL READY` or `WAIT FOR`
 
 - `WAIT UNTIL READY (TIMEOUT = ..., ON TIMEOUT = ...)` sets the timeout for the
   resize. On timeout, `ON TIMEOUT` selects whether to `COMMIT` (retire the old
-  replicas and proceed with the not-yet-hydrated new ones, which can cause
-  downtime) or `ROLLBACK` (keep the current size). Default: `ROLLBACK`.
+  replicas and proceed with the new ones even if they have not yet hydrated or
+  caught up, which can cause downtime or stale results) or `ROLLBACK` (keep the
+  current size). Default: `ROLLBACK`.
 
   ```mzsql
   ALTER CLUSTER c1
   SET (SIZE = '100cc') WITH (WAIT UNTIL READY (TIMEOUT = '10m'));
   ```
 
 - `WAIT FOR '<duration>'` sets the timeout and commits when it expires,
-  regardless of hydration status, which can cause downtime. Prefer
-  `WAIT UNTIL READY`.
+  regardless of whether the new replicas have hydrated or caught up, which can
+  cause downtime. Prefer `WAIT UNTIL READY`.
 
 See [Monitoring a resize](#monitoring-a-resize) to track progress and
 [cancel](#monitoring-a-resize) an in-flight resize.
@@ -230,8 +231,8 @@ configuration.
 You can use the `WAIT UNTIL READY` option to perform a zero-downtime resizing,
 which incurs **no downtime**. Instead of restarting the cluster, this approach
 spins up an additional cluster replica under the covers with the desired new
-size, waits for the replica to be hydrated, and then replaces the original
-replica.
+size, waits for the replica to be hydrated and caught up, and then replaces the
+original replica.
 
 ```sql
 ALTER CLUSTER c1

misc/python/materialize/mzcompose/__init__.py

@@ -857,6 +857,8 @@ def get_default_system_parameters(
     "read_then_write_max_dependencies",
     "enable_hydration_burst",
     "default_hydration_burst_linger",
+    "enable_cluster_reconfiguration_lag_gate",
+    "cluster_reconfiguration_allowed_lag",
 ]
 
 

misc/python/materialize/parallel_workload/action.py

@@ -3428,6 +3428,11 @@ def __init__(
             "read_then_write_max_dependencies",
             "enable_hydration_burst",
             "default_hydration_burst_linger",
+            # The graceful cut-over lag gate. Flipping the gate off or the
+            # allowance to an arbitrary value mid-reconfiguration changes when a
+            # cut-over fires, which the workload does not model.
+            "enable_cluster_reconfiguration_lag_gate",
+            "cluster_reconfiguration_allowed_lag",
         ]
 
     def errors_to_ignore(self, exe: Executor) -> list[str]: