diff --git a/doc/developer/design/20260522_cluster_autoscaling.md b/doc/developer/design/20260522_cluster_autoscaling.md index ad7596b6a8dd6..a2d064a57541a 100644 --- a/doc/developer/design/20260522_cluster_autoscaling.md +++ b/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` — 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`: 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`: 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` — 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. diff --git a/doc/user/content/sql/alter-cluster.md b/doc/user/content/sql/alter-cluster.md index 61a54fc473471..fb97f80244cc1 100644 --- a/doc/user/content/sql/alter-cluster.md +++ b/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,8 +183,9 @@ 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 @@ -192,8 +193,8 @@ Customizing the resize timeout with `WAIT UNTIL READY` or `WAIT FOR` ``` - `WAIT FOR ''` 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 diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 1be7ba3c7ec07..6519c46a6562d 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/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", ] diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 194d7b2f31516..e4f8363917970 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/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]: diff --git a/src/adapter-types/src/dyncfgs.rs b/src/adapter-types/src/dyncfgs.rs index 1f50dafbd0aad..d8f1869a54f7a 100644 --- a/src/adapter-types/src/dyncfgs.rs +++ b/src/adapter-types/src/dyncfgs.rs @@ -466,6 +466,35 @@ pub const CLUSTER_CONTROLLER_TICK_INTERVAL: Config = Config::new( ParameterScope::Environment, ); +/// Whether a replica must be caught up, not merely hydrated, before a graceful +/// reconfiguration cuts over to it. +/// +/// Break-glass: with this off the cut-over gate is hydration alone, which is the +/// behavior from before the lag term existed. +pub const ENABLE_CLUSTER_RECONFIGURATION_LAG_GATE: Config = Config::new( + "enable_cluster_reconfiguration_lag_gate", + true, + "Whether a graceful reconfiguration requires its target replicas to be within \ + cluster_reconfiguration_allowed_lag of the replicas they replace, on top of being hydrated.", + ParameterScope::Environment, +); + +/// How far behind a graceful reconfiguration's target replicas may be and still +/// be cut over to. +/// +/// Measured per collection against the furthest output frontier any replica of +/// the cluster reports for it, which while the outgoing replicas are still +/// present is theirs. The duration is applied as that many milliseconds of the +/// collection's timestamp domain, which is exact on the epoch-milliseconds +/// timeline and a raw tick count on any other. +pub const CLUSTER_RECONFIGURATION_ALLOWED_LAG: Config = Config::new( + "cluster_reconfiguration_allowed_lag", + Duration::from_secs(60), + "Maximum allowed lag when determining whether a graceful reconfiguration's target replicas \ + have caught up with the replicas they replace.", + ParameterScope::Environment, +); + /// Whether a config-shape `ALTER CLUSTER` returns immediately, with the /// controller converging in the background, or blocks the session on a /// wait-shim until the reconfiguration completes or its deadline passes. @@ -527,6 +556,8 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { configs .add(&ALLOW_USER_SESSIONS) .add(&CLUSTER_CONTROLLER_TICK_INTERVAL) + .add(&ENABLE_CLUSTER_RECONFIGURATION_LAG_GATE) + .add(&CLUSTER_RECONFIGURATION_ALLOWED_LAG) .add(&ENABLE_BACKGROUND_ALTER_CLUSTER) .add(&DEFAULT_CLUSTER_RECONFIGURATION_TIMEOUT) .add(&ENABLE_HYDRATION_BURST) diff --git a/src/adapter/src/coord/caught_up.rs b/src/adapter/src/coord/caught_up.rs index 37228f8682ea0..a02565b00b80e 100644 --- a/src/adapter/src/coord/caught_up.rs +++ b/src/adapter/src/coord/caught_up.rs @@ -46,8 +46,7 @@ use mz_controller_types::{ClusterId, ReplicaId}; use mz_orchestrator::OfflineReason; use mz_ore::channel::trigger::Trigger; use mz_ore::now::EpochMillis; -use mz_repr::{GlobalId, Timestamp}; -use timely::PartialOrder; +use mz_repr::{GlobalId, Timestamp, frontier_within_lag}; use timely::progress::{Antichain, Timestamp as _}; use crate::coord::{ClusterReplicaStatuses, Coordinator}; @@ -649,14 +648,10 @@ impl Coordinator { // NOTE: there is deliberately no `cutoff` escape hatch here. A frontier frozen // at the minimum is exactly what this gate must catch, so a collection stuck // here blocks promotion until `with_0dt_deployment_max_wait` elapses. - let write_frontier_plus_allowed_lag = Antichain::from_iter( - write_frontier - .iter() - .map(|t| t.step_forward_by(&allowed_lag)), - ); - let within_lag = PartialOrder::less_equal( + let within_lag = frontier_within_lag( + &write_frontier, &Antichain::from_elem(now), - &write_frontier_plus_allowed_lag, + allowed_lag, ); tracing::info!( @@ -713,17 +708,7 @@ impl Coordinator { continue; } - // We can't do easy comparisons and subtractions, so we bump up the - // write frontier by the allowed lag, and then compare that against - // the write frontier. - let write_frontier_plus_allowed_lag = write_frontier - .iter() - .map(|t| t.step_forward_by(&allowed_lag)); - let bumped_write_plus_allowed_lag = - Antichain::from_iter(write_frontier_plus_allowed_lag); - - let within_lag = - PartialOrder::less_equal(live_write_frontier, &bumped_write_plus_allowed_lag); + let within_lag = frontier_within_lag(&write_frontier, live_write_frontier, allowed_lag); // This call is on the expensive side, because we have to do a call // across a task/channel boundary, and our work competes with other diff --git a/src/adapter/src/coord/cluster_controller.rs b/src/adapter/src/coord/cluster_controller.rs index cd17a499ae059..e8846f410fd52 100644 --- a/src/adapter/src/coord/cluster_controller.rs +++ b/src/adapter/src/coord/cluster_controller.rs @@ -29,7 +29,10 @@ use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use std::time::Duration; -use mz_adapter_types::dyncfgs::CLUSTER_CONTROLLER_TICK_INTERVAL; +use mz_adapter_types::dyncfgs::{ + CLUSTER_CONTROLLER_TICK_INTERVAL, CLUSTER_RECONFIGURATION_ALLOWED_LAG, + ENABLE_CLUSTER_RECONFIGURATION_LAG_GATE, +}; use mz_catalog::memory::objects::{ClusterConfig, ClusterVariant}; use mz_cluster_controller::ClusterController; use mz_cluster_controller::ctx::{ @@ -38,6 +41,7 @@ use mz_cluster_controller::ctx::{ ReconfigurationTarget, RefreshMvInfo, RefreshWindowClusterInputs, RefreshWindowInputsBatch, ReplicaShape, StateWrite, }; +use mz_cluster_controller::strategy::duration_to_ts; use mz_compute_types::config::ComputeReplicaConfig; use mz_controller::clusters::ClusterStatus; use mz_controller_types::{ClusterId, ReplicaId}; @@ -56,7 +60,8 @@ use crate::error::AdapterError; /// `ManagedClusterIds` and `ClusterStates` are the per-tick batched reads. The /// `ClusterStates` reply also carries `now`. Refresh-window catalog inputs are /// pulled one cluster at a time, followed by one shared oracle read. -/// `HydratedReplicas` is a per-cluster live signal a strategy pulls on demand. +/// `HydratedReplicas` and `ReadyReplicas` are per-cluster live signals a strategy +/// pulls on demand. #[derive(Debug)] pub enum ClusterControllerRequest { /// The ids of all managed clusters the controller owns this tick. @@ -74,6 +79,18 @@ pub enum ClusterControllerRequest { replicas: Vec, tx: oneshot::Sender>, }, + /// Of `replicas` on `cluster`, which are online, have all current + /// collections hydrated, *and* are within the configured cut-over lag of + /// the replicas they would replace. + /// + /// The stronger form of [`Self::HydratedReplicas`], for callers deciding + /// whether to cut over to a replica rather than merely observing that its + /// dataflows have started producing output. + ReadyReplicas { + cluster_id: ClusterId, + replicas: Vec, + tx: oneshot::Sender>, + }, /// Whether the cluster has any hydratable (dataflow-backed) objects bound to /// it. HasHydratableObjects { @@ -99,9 +116,9 @@ pub enum ClusterControllerRequest { TickInterval { tx: oneshot::Sender }, } -struct ReplicaHydrationCheck { +struct ReplicaReadinessCheck { replica_id: ReplicaId, - compute_hydrated: oneshot::Receiver, + compute_ready: oneshot::Receiver, } /// The controller-task side of the boundary: a [`ClusterControllerCtx`] that @@ -174,6 +191,21 @@ impl ClusterControllerCtx for CoordCtx { .unwrap_or_default() } + async fn ready_replicas( + &mut self, + cluster_id: ClusterId, + replicas: &[ReplicaId], + ) -> BTreeSet { + let replicas = replicas.to_vec(); + self.request(|tx| ClusterControllerRequest::ReadyReplicas { + cluster_id, + replicas, + tx, + }) + .await + .unwrap_or_default() + } + async fn has_hydratable_objects(&mut self, cluster_id: ClusterId) -> bool { self.request(|tx| ClusterControllerRequest::HasHydratableObjects { cluster_id, tx }) .await @@ -323,19 +355,23 @@ impl Coordinator { replicas, tx, } => { - let checks = self.start_hydration_checks(cluster_id, replicas); - // Start the controller calls on the coordinator loop, then wait - // for compute's replies off-loop. The compute check can wait on - // the compute instance task. - spawn(|| "cluster_controller_hydration_probe", async move { - let mut hydrated = BTreeSet::new(); - for check in checks { - if check.compute_hydrated.await.unwrap_or(false) { - hydrated.insert(check.replica_id); - } - } - let _ = tx.send(hydrated); - }); + // Hydration only: this signal drives the burst strategy's + // linger, whose durable `steady_hydrated_at` stamp means exactly + // "hydration was observed". The cut-over gate is `ReadyReplicas`. + let checks = self.start_readiness_checks(cluster_id, replicas, None); + Self::finish_readiness_checks(checks, tx); + } + ClusterControllerRequest::ReadyReplicas { + cluster_id, + replicas, + tx, + } => { + let checks = self.start_readiness_checks( + cluster_id, + replicas, + self.reconfiguration_allowed_lag(), + ); + Self::finish_readiness_checks(checks, tx); } ClusterControllerRequest::HasHydratableObjects { cluster_id, tx } => { let _ = tx.send(self.cluster_has_hydratable_objects(cluster_id)); @@ -424,16 +460,55 @@ impl Coordinator { .any(|id| self.catalog().get_entry(id).item().is_hydratable()) } - /// Starts per-replica hydration checks for `cluster_id`. + /// The cut-over lag allowance, or `None` when the lag gate is disabled. + /// + /// Read per probe rather than latched per tick, so a runtime change takes + /// effect without a restart, matching `cluster_controller_tick_interval`. + pub(crate) fn reconfiguration_allowed_lag(&self) -> Option { + let dyncfgs = self.catalog().system_config().dyncfgs(); + ENABLE_CLUSTER_RECONFIGURATION_LAG_GATE + .get(dyncfgs) + // A lag beyond the timestamp domain means "any lag is fine", which + // is what `duration_to_ts` saturating at `MAX` expresses. + .then(|| duration_to_ts(CLUSTER_RECONFIGURATION_ALLOWED_LAG.get(dyncfgs))) + } + + /// Await the started checks off the coordinator loop and reply with the + /// replicas that passed. The compute check can wait on the compute instance + /// task, so it must not block the loop. + fn finish_readiness_checks( + checks: Vec, + tx: oneshot::Sender>, + ) { + spawn(|| "cluster_controller_readiness_probe", async move { + let mut ready = BTreeSet::new(); + for check in checks { + if check.compute_ready.await.unwrap_or(false) { + ready.insert(check.replica_id); + } + } + let _ = tx.send(ready); + }); + } + + /// Starts per-replica readiness checks for `cluster_id`: hydration, plus + /// the lag gate when `allowed_lag` is `Some`. /// /// Returns only checks for replicas whose processes are all online, that /// are already storage-hydrated, and that are known to the compute /// controller. The compute receiver completes off the coordinator loop. - fn start_hydration_checks( + /// + /// The storage-side check is hydration only. Storage hydration has its own + /// definition (see `StorageController::collections_hydrated_on_replicas`), + /// and no lag term is applied to it here; a pending replica hosting an + /// ingestion can pass this gate with the source's snapshot complete but its + /// replay still in progress. + fn start_readiness_checks( &self, cluster_id: ClusterId, replicas: Vec, - ) -> Vec { + allowed_lag: Option, + ) -> Vec { use mz_catalog::memory::objects::CatalogItem; // Materialized views pinned to a replica (via `IN CLUSTER ... REPLICA`) @@ -475,14 +550,15 @@ impl Coordinator { .filter(|(target, _)| *target != replica_id) .map(|(_, id)| *id) .collect(); - let compute_fut = match self.controller.compute.collections_hydrated_for_replicas( + let compute_fut = match self.controller.compute.collections_ready_for_replicas( cluster_id, vec![replica_id], exclude.clone(), + allowed_lag, ) { Ok(fut) => fut, // The replica is not known to the compute controller. Treat it - // as not hydrated. + // as not ready. Err(_) => continue, }; let storage_hydrated = match self.controller.storage.collections_hydrated_on_replicas( @@ -494,9 +570,9 @@ impl Coordinator { Err(_) => continue, }; if storage_hydrated { - checks.push(ReplicaHydrationCheck { + checks.push(ReplicaReadinessCheck { replica_id, - compute_hydrated: compute_fut, + compute_ready: compute_fut, }); } } diff --git a/src/adapter/src/coord/sequencer/inner/cluster.rs b/src/adapter/src/coord/sequencer/inner/cluster.rs index dadf70ff5dcd9..70328e6146b97 100644 --- a/src/adapter/src/coord/sequencer/inner/cluster.rs +++ b/src/adapter/src/coord/sequencer/inner/cluster.rs @@ -1152,11 +1152,28 @@ impl Coordinator { } } } - let compute_hydrated_fut = self + // The same gate the cluster controller's reconcile tick applies: a + // pending replica is ready only once it is hydrated *and* caught up + // with the replicas it is replacing. See + // `Coordinator::start_readiness_checks`. + // + // Unlike the tick, this shim asks about all pending replicas as one set + // ("every collection ready on some pending replica") and excludes + // nothing, so with `replication_factor > 1` one caught-up pending replica + // satisfies it, and a collection pinned to an outgoing replica can never + // satisfy it. Both predate the lag gate. + let allowed_lag = self.reconfiguration_allowed_lag(); + + let compute_ready_fut = self .controller .compute - .collections_hydrated_for_replicas(cluster.id, pending_replicas.clone(), [].into()) - .map_err(|e| AdapterError::internal("Failed to check hydration", e))?; + .collections_ready_for_replicas( + cluster.id, + pending_replicas.clone(), + [].into(), + allowed_lag, + ) + .map_err(|e| AdapterError::internal("Failed to check readiness", e))?; let storage_hydrated = self .controller @@ -1182,11 +1199,11 @@ impl Coordinator { Ok(StageResult::Handle(mz_ore::task::spawn( || "Alter Cluster: wait for hydrated", async move { - let compute_hydrated = compute_hydrated_fut + let compute_ready = compute_ready_fut .await - .map_err(|e| AdapterError::internal("Failed to check hydration", e))?; + .map_err(|e| AdapterError::internal("Failed to check readiness", e))?; - if compute_hydrated && storage_hydrated && replicas_online { + if compute_ready && storage_hydrated && replicas_online { // We're done Ok(Box::new(ClusterStage::Finalize(AlterClusterFinalize { validity, diff --git a/src/cluster-controller/src/ctx.rs b/src/cluster-controller/src/ctx.rs index c4bee468fc438..0da7b6e1b504a 100644 --- a/src/cluster-controller/src/ctx.rs +++ b/src/cluster-controller/src/ctx.rs @@ -428,6 +428,33 @@ pub trait ClusterControllerCtx: Send { replicas: &[ReplicaId], ) -> BTreeSet; + /// Of `replicas` on `cluster`, which are online, have *all* current + /// (non-transient) collections on the cluster hydrated, *and* for each + /// collection are no further than the configured allowance behind the + /// furthest output frontier any replica of the cluster reports for it. The + /// returned set is a subset of `replicas`, and of + /// [`Self::hydrated_replicas`]. + /// + /// This is the signal for deciding whether to cut over to a replica. + /// Hydration alone is not: a dataflow's as-of is pinned when its replica is + /// added and never moves, so a long-hydrating collection reports hydrated + /// the moment its initial snapshot lands, with everything since the as-of + /// still to replay. Cutting over on hydration alone drops the caught-up + /// replicas and leaves the cluster's frontiers frozen until the new ones + /// catch up. + /// + /// [`Self::hydrated_replicas`] remains the right signal for observing that + /// dataflows have started producing output, which is what the burst + /// strategy's durable `steady_hydrated_at` stamp records. + /// + /// Callers should request only replicas their strategy currently needs. This + /// keeps live-signal dependencies local to the strategies that consume them. + async fn ready_replicas( + &mut self, + cluster_id: ClusterId, + replicas: &[ReplicaId], + ) -> BTreeSet; + /// Whether `cluster_id` has at least one hydratable (dataflow-backed) object /// bound to it: an index, materialized view, ingestion source, or sink. /// diff --git a/src/cluster-controller/src/lib.rs b/src/cluster-controller/src/lib.rs index 9fac1055f9a58..c5ce0519dde94 100644 --- a/src/cluster-controller/src/lib.rs +++ b/src/cluster-controller/src/lib.rs @@ -355,7 +355,7 @@ impl ClusterController { if request.hydratable_objects { live.has_hydratable_objects = ctx.has_hydratable_objects(state.cluster_id).await; } - if request.hydration { + if request.hydration || request.readiness { let replica_ids: Vec<_> = state .replicas .iter() @@ -363,8 +363,20 @@ impl ClusterController { .map(|r| r.replica_id) .collect(); if !replica_ids.is_empty() { - live.hydrated_replicas = - ctx.hydrated_replicas(state.cluster_id, &replica_ids).await; + // Two probes only when a cluster has both an in-flight + // reconfiguration and an armed burst policy. Each is an + // in-memory pass over the instance's collections, and the + // two answer different questions, so neither subsumes the + // other cheaply enough to be worth deriving one from the + // other here. + if request.hydration { + live.hydrated_replicas = + ctx.hydrated_replicas(state.cluster_id, &replica_ids).await; + } + if request.readiness { + live.ready_replicas = + ctx.ready_replicas(state.cluster_id, &replica_ids).await; + } } } if request.refresh_window { diff --git a/src/cluster-controller/src/strategy.rs b/src/cluster-controller/src/strategy.rs index c8584c713bef7..1945589bcaa7f 100644 --- a/src/cluster-controller/src/strategy.rs +++ b/src/cluster-controller/src/strategy.rs @@ -103,6 +103,9 @@ pub trait Strategy: Send + Sync { pub struct SignalRequest { /// Probe which of the cluster's replicas report all collections hydrated. pub hydration: bool, + /// Probe which of the cluster's replicas are ready to be cut over to: + /// hydrated, and within the configured lag of the replicas they replace. + pub readiness: bool, /// Check whether the cluster has at least one hydratable object bound to /// it. See `ClusterControllerCtx::has_hydratable_objects` for what counts. pub hydratable_objects: bool, @@ -118,11 +121,13 @@ impl SignalRequest { // compile error here until its union is spelled out. let SignalRequest { hydration, + readiness, hydratable_objects, refresh_window, } = other; SignalRequest { hydration: self.hydration || hydration, + readiness: self.readiness || readiness, hydratable_objects: self.hydratable_objects || hydratable_objects, refresh_window: self.refresh_window || refresh_window, } @@ -152,6 +157,10 @@ pub struct LiveSignals { /// The replicas observed this tick to be online and to have *all* current /// collections on the cluster hydrated. pub hydrated_replicas: BTreeSet, + /// The replicas observed this tick to be ready to cut over to: hydrated, and + /// within the configured lag of the replicas they would replace. A subset of + /// `hydrated_replicas`. Empty when not requested. + pub ready_replicas: BTreeSet, /// Whether the cluster has at least one hydratable object. `false` when not /// requested. pub has_hydratable_objects: bool, @@ -207,52 +216,59 @@ impl Strategy for BaselineStrategy { /// Engaged whenever the durable `reconfiguration` record is in progress. It /// desires `target.replication_factor` replicas at the target shape in addition /// to the baseline's realized-shape replicas, so both sets serve while the new -/// one hydrates. Once rf-many target replicas are present and hydrated, +/// one hydrates and catches up. Once rf-many target replicas are present and ready, /// `update_state` cuts over: the realized config advances to the target, the /// 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 to the un-hydrated target anyway while `Rollback` (the +/// `Commit` cuts over to the not-yet-ready target anyway while `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`] +/// [`LiveSignals`]. Readiness is requested via [`Strategy::signal_request`] /// exactly while an in-progress reconfiguration is present. #[derive(Clone, Copy, Debug, Default)] pub struct GracefulReconfigurationStrategy; impl GracefulReconfigurationStrategy { /// Whether the cut-over precondition holds: at least - /// `target.replication_factor` replicas of the target shape report - /// hydrated. + /// `target.replication_factor` replicas of the target shape report ready. /// - /// Requiring rf-many hydrated replicas (not just one) preserves the + /// Ready, not merely hydrated: a hydrated replica has produced output past + /// the as-of it was installed with, which for a slow-hydrating collection + /// can be hours behind the replicas being replaced. See + /// [`ClusterControllerCtx::ready_replicas`]. + /// + /// Requiring rf-many ready replicas (not just one) preserves the /// high-availability guarantee of `replication_factor > 1` across the /// cut-over. Extra target-shape replicas beyond the rf do not block: the /// post-cut-over reconcile retires them anyway, so waiting for them to - /// hydrate would only delay the cut-over. - fn target_hydrated( + /// become ready would only delay the cut-over. + /// + /// [`ClusterControllerCtx::ready_replicas`]: + /// crate::ctx::ClusterControllerCtx::ready_replicas + fn target_ready( &self, state: &ClusterState, signals: &LiveSignals, record: &ReconfigurationRecord, ) -> bool { let target_shape = record.target.shape(); - let hydrated_target_replicas = state + let ready_target_replicas = state .replicas .iter() .filter(|r| r.owned_shape().is_some_and(|s| s.matches(&target_shape))) - .filter(|r| signals.hydrated_replicas.contains(&r.replica_id)) + .filter(|r| signals.ready_replicas.contains(&r.replica_id)) .count(); let target_rf = usize::try_from(record.target.replication_factor).unwrap_or(usize::MAX); - hydrated_target_replicas >= target_rf + ready_target_replicas >= target_rf } } impl Strategy for GracefulReconfigurationStrategy { fn signal_request(&self, state: &ClusterState, _config: &ConfigSignals) -> SignalRequest { SignalRequest { - hydration: state + readiness: state .reconfiguration .as_ref() .is_some_and(|record| record.is_in_progress()), @@ -276,10 +292,10 @@ impl Strategy for GracefulReconfigurationStrategy { // Cut over by advancing the realized config to the target and marking // the record finalized on either of two conditions: - // 1. rf-many target replicas are present and hydrated (success, which + // 1. rf-many target replicas are present and ready (success, which // takes precedence over the deadline regardless of `on_timeout`), or - // 2. the deadline has been reached un-hydrated and `on_timeout` is - // `Commit` (cut over to the not-yet-hydrated target anyway). + // 2. the deadline has been reached un-ready and `on_timeout` is + // `Commit` (cut over to the not-yet-ready target anyway). // // NOTE: the deadline is reached at `now >= deadline`, not `now > deadline`. // A `WAIT FOR '0s'` writes `deadline = now` to request an immediate @@ -288,10 +304,10 @@ impl Strategy for GracefulReconfigurationStrategy { // target replicas and only a later tick would cut over. `>=` fires the // deadline the instant it is reached, so the zero-timeout cut-over happens // on the first tick, before any overlap replica is desired. - let hydrated = self.target_hydrated(state, signals, record); + let ready = self.target_ready(state, signals, record); let deadline_reached = now >= record.deadline; let commit_on_timeout = deadline_reached && matches!(record.on_timeout, OnTimeout::Commit); - if hydrated || commit_on_timeout { + if ready || commit_on_timeout { return StateWrite { new_size: Some(record.target.size.clone()), new_replication_factor: Some(record.target.replication_factor), @@ -304,16 +320,16 @@ impl Strategy for GracefulReconfigurationStrategy { ..record.clone() }), // A cut-over that only happens because the deadline passed - // under `Commit` is forced: the target has not hydrated. + // under `Commit` is forced: the target is not ready. // Declared here because only this decision point knows. // The durable status reads `Finalized` either way. - audit: Some(ReconfigurationAudit::Finalized { forced: !hydrated }), + audit: Some(ReconfigurationAudit::Finalized { forced: !ready }), }), ..Default::default() }; } - // Past the deadline un-hydrated under `Rollback`: abandon the + // Past the deadline not ready under `Rollback`: abandon the // reconfiguration while leaving the realized config untouched. The // terminal status is the durable transition the audit event records. With // the record no longer in progress the strategy stops contributing the @@ -349,7 +365,7 @@ impl Strategy for GracefulReconfigurationStrategy { return Vec::new(); } - // Past the deadline with the target not hydrated under `Rollback`: stop + // Past the deadline with the target not ready under `Rollback`: stop // contributing the target replicas. `update_state` marks the record // timed out in this same tick's first phase, so this usually never fires // against a re-read state. It matters when the deadline crosses between @@ -363,7 +379,7 @@ impl Strategy for GracefulReconfigurationStrategy { // `now >= deadline` matches `update_state`'s boundary, so a zero-timeout // rollback stops desiring the target on the same tick it marks the // record timed out. - let timed_out = now >= record.deadline && !self.target_hydrated(state, signals, record); + let timed_out = now >= record.deadline && !self.target_ready(state, signals, record); if timed_out && matches!(record.on_timeout, OnTimeout::Rollback) { return Vec::new(); } @@ -552,7 +568,7 @@ impl Strategy for OnRefreshStrategy { /// on overflow rather than panicking the controller on a bad input. /// /// [`Duration`]: std::time::Duration -fn duration_to_ts(duration: std::time::Duration) -> Timestamp { +pub fn duration_to_ts(duration: std::time::Duration) -> Timestamp { Timestamp::try_from(duration).unwrap_or(Timestamp::MAX) } diff --git a/src/cluster-controller/src/tests.rs b/src/cluster-controller/src/tests.rs index d311bf69d8445..3813023178ea5 100644 --- a/src/cluster-controller/src/tests.rs +++ b/src/cluster-controller/src/tests.rs @@ -149,11 +149,18 @@ struct FakeCtx { /// `schedule` field of the witness. concurrent_schedule_alter: BTreeMap, /// Replicas the fake reports as hydrated when the controller probes. A - /// graceful test sets this to drive cut-over. + /// burst test sets this to drive the linger. hydrated: BTreeSet, + /// Replicas the fake reports as ready (hydrated *and* caught up) when the + /// controller probes. A graceful test sets this to drive cut-over. Held + /// separately from `hydrated` rather than derived from it, so a test can + /// pose the state this gate exists for: hydrated but still behind. + ready: BTreeSet, /// How many times the controller probed hydration, for asserting that an /// object-less cluster is never probed. hydration_probes: usize, + /// How many times the controller probed readiness. + readiness_probes: usize, /// What the fake answers when the controller pulls the object-existence /// signal; an absent entry reads `false` (no objects). Held beside the /// states (like `hydrated`) rather than read from them, and @@ -180,7 +187,9 @@ impl FakeCtx { concurrent_policy_alter: BTreeMap::new(), concurrent_schedule_alter: BTreeMap::new(), hydrated: BTreeSet::new(), + ready: BTreeSet::new(), hydration_probes: 0, + readiness_probes: 0, has_hydratable_objects: BTreeMap::new(), refresh_window: None, refresh_window_probes: Vec::new(), @@ -261,6 +270,19 @@ impl ClusterControllerCtx for FakeCtx { .collect() } + async fn ready_replicas( + &mut self, + _cluster_id: ClusterId, + replicas: &[ReplicaId], + ) -> BTreeSet { + self.readiness_probes += 1; + replicas + .iter() + .copied() + .filter(|r| self.ready.contains(r)) + .collect() + } + async fn has_hydratable_objects(&mut self, cluster_id: ClusterId) -> bool { self.has_hydratable_objects .get(&cluster_id) @@ -1267,14 +1289,19 @@ fn record_on_timeout( } /// Convenience: a `ClusterState` with an in-flight reconfiguration, plus the -/// [`LiveSignals`] carrying an explicit hydrated-replica set. +/// [`LiveSignals`] carrying an explicit ready-replica set. +/// +/// `ready` populates `LiveSignals::ready_replicas`, the signal the graceful +/// cut-over gate reads. `hydrated_replicas` is deliberately left empty: a +/// replica that is hydrated but not ready must not cut over, and a helper that +/// set both would hide a gate reading the wrong one. fn reconfiguring_state( cluster_id: ClusterId, size: &str, rf: u32, replicas: Vec, rec: ReconfigurationRecord, - hydrated: BTreeSet, + ready: BTreeSet, ) -> (ClusterState, LiveSignals) { let state = ClusterState { cluster_id, @@ -1290,7 +1317,7 @@ fn reconfiguring_state( replicas, }; let signals = LiveSignals { - hydrated_replicas: hydrated, + ready_replicas: ready, ..Default::default() }; (state, signals) @@ -1347,8 +1374,84 @@ fn graceful_desires_target_while_in_flight() { } #[mz_ore::test] -fn graceful_cuts_over_when_target_hydrated() { - // Both target replicas present and hydrated -> cut over, even before deadline. +fn graceful_holds_when_target_is_hydrated_but_lagging() { + // The regression this gate exists for. Both target replicas are present and + // report hydrated, but neither is caught up with the replicas it replaces, + // so `ready_replicas` is empty. Cutting over here would drop the two + // caught-up 100cc replicas and freeze the cluster's frontiers until the + // 200cc pair drained its backlog. + let c = cluster(1); + let (state, mut signals) = reconfiguring_state( + c, + "100cc", + 2, + vec![ + observed(replica(1), "r0", "100cc"), + observed(replica(2), "r1", "100cc"), + observed(replica(3), "r2", "200cc"), + observed(replica(4), "r3", "200cc"), + ], + record("200cc", 2, 5000), + BTreeSet::new(), + ); + // Hydrated but not ready: exactly the state a hydration-only gate cut over in. + signals.hydrated_replicas = BTreeSet::from([replica(3), replica(4)]); + let now = Timestamp::from(1000u64); + + let g = GracefulReconfigurationStrategy; + assert!( + g.update_state(&state, &signals, &config(), now).is_empty(), + "a hydrated but lagging target must not cut over", + ); + // The target replicas stay desired while we wait for them to catch up. + let desired = g.desired_replicas(&state, &signals, &config(), now); + assert_eq!(desired.len(), 2); + assert!(desired.iter().all(|d| d.shape.size == "200cc")); + + // Once they catch up, the same state cuts over. + signals.ready_replicas = BTreeSet::from([replica(3), replica(4)]); + let write = g.update_state(&state, &signals, &config(), now); + assert_eq!(write.new_size.as_deref(), Some("200cc")); + assert_eq!( + written_reconfiguration_audit(&write), + Some(ReconfigurationAudit::Finalized { forced: false }), + "a cut-over on a caught-up target is not forced", + ); +} + +#[mz_ore::test] +fn graceful_commit_on_timeout_cuts_over_lagging_target() { + // The lag gate strengthens the success condition, not the timeout action: + // past the deadline under `Commit`, a hydrated-but-lagging target is still + // cut over to, and the finalize is recorded as forced. + let c = cluster(1); + let (state, mut signals) = reconfiguring_state( + c, + "100cc", + 1, + vec![ + observed(replica(1), "r0", "100cc"), + observed(replica(2), "r1", "200cc"), + ], + record_on_timeout("200cc", 1, 5000, OnTimeout::Commit), + BTreeSet::new(), + ); + signals.hydrated_replicas = BTreeSet::from([replica(2)]); + let past_deadline = Timestamp::from(9999u64); + + let g = GracefulReconfigurationStrategy; + let write = g.update_state(&state, &signals, &config(), past_deadline); + assert_eq!(write.new_size.as_deref(), Some("200cc")); + assert_eq!( + written_reconfiguration_audit(&write), + Some(ReconfigurationAudit::Finalized { forced: true }), + "a deadline cut-over on a lagging target is forced", + ); +} + +#[mz_ore::test] +fn graceful_cuts_over_when_target_ready() { + // Both target replicas present and ready -> cut over, even before deadline. let c = cluster(1); let (state, signals) = reconfiguring_state( c, @@ -1412,7 +1515,7 @@ fn graceful_partial_hydration_does_not_cut_over() { #[mz_ore::test] fn graceful_rf_zero_target_cuts_over_on_first_tick() { // A target with replication_factor 0 has no replicas to hydrate, so - // `target_hydrated` is vacuously true and `update_state` finalizes on the + // `target_ready` is vacuously true and `update_state` finalizes on the // first tick, well before the deadline. The audit declares an unforced // (hydrated) finalize. let c = cluster(1); @@ -1769,9 +1872,9 @@ fn graceful_az_only_reconfiguration_is_a_shape_change() { .matches(state.replicas[0].shape.as_ref().unwrap()) ); - // Mark the realized replica hydrated: it is NOT a target replica (wrong AZ), + // Mark the realized replica ready: it is NOT a target replica (wrong AZ), // so this must not trigger a cut-over. - signals.hydrated_replicas.insert(replica(1)); + signals.ready_replicas.insert(replica(1)); assert!(g.update_state(&state, &signals, &config(), now).is_empty()); } @@ -1808,7 +1911,7 @@ async fn graceful_full_flow_overlap_then_cutover() { assert_eq!(ctx.states[&c].size, "100cc", "realized config unchanged"); assert_eq!(ctx.states[&c].replicas.len(), 4); - // The target replicas are the two 200cc ones. Mark them hydrated. + // The target replicas are the two 200cc ones. Mark them ready. let target_ids: BTreeSet<_> = ctx.states[&c] .replicas .iter() @@ -1816,7 +1919,7 @@ async fn graceful_full_flow_overlap_then_cutover() { .map(|r| r.replica_id) .collect(); assert_eq!(target_ids.len(), 2); - ctx.hydrated = target_ids.clone(); + ctx.ready = target_ids.clone(); // Tick 2: cut over (phase 1) then drop the old 100cc replicas (phase 2). let before = ctx.applied.len(); @@ -1846,6 +1949,18 @@ async fn graceful_full_flow_overlap_then_cutover() { .any(|d| matches!(d, Decision::DropReplica { .. })); assert!(dropped, "a drop happened"); + // The graceful strategy pulled the readiness signal, and only that one: a + // reconfiguration with no burst policy must never consult bare hydration, + // which is the signal that cut over early in production. + assert_eq!( + ctx.readiness_probes, 2, + "one readiness probe per tick while the record is in progress" + ); + assert_eq!( + ctx.hydration_probes, 0, + "the graceful path does not consult bare hydration" + ); + // Tick 3: converged, no further decisions. let before = ctx.applied.len(); controller.reconcile(&mut ctx).await; @@ -1872,9 +1987,9 @@ async fn graceful_alter_back_finalizes_without_churn() { BTreeSet::new(), ); let mut ctx = FakeCtx::new(vec![state]); - // The controller probes hydration through the ctx. The existing replica is - // already hydrated. - ctx.hydrated = BTreeSet::from([replica(1)]); + // The controller probes readiness through the ctx. The existing replica is + // already ready. + ctx.ready = BTreeSet::from([replica(1)]); let controller = controller(); controller.reconcile(&mut ctx).await; diff --git a/src/compute-client/src/controller.rs b/src/compute-client/src/controller.rs index bfd41301888a8..73bde47f07256 100644 --- a/src/compute-client/src/controller.rs +++ b/src/compute-client/src/controller.rs @@ -423,17 +423,23 @@ impl ComputeController { Ok(res) } - /// Returns `true` if all non-transient, non-excluded collections are hydrated on any of the - /// provided replicas. + /// Returns `true` if all non-transient, non-excluded collections are ready on any of the + /// provided replicas: hydrated, and, when `allowed_lag` is `Some`, no further than that + /// behind the furthest output frontier any replica of the instance reports for the + /// collection. /// - /// For this check, zero-replica clusters are always considered hydrated. + /// See `Instance::collections_ready_on_replicas` for why hydration alone is not a + /// readiness signal for a cut-over. + /// + /// For this check, zero-replica clusters are always considered ready. /// Their collections would never normally be considered hydrated but it's /// clearly intentional that they have no replicas. - pub fn collections_hydrated_for_replicas( + pub fn collections_ready_for_replicas( &self, instance_id: ComputeInstanceId, replicas: Vec, exclude_collections: BTreeSet, + allowed_lag: Option, ) -> Result, anyhow::Error> { let instance = self.instance(instance_id)?; @@ -447,7 +453,7 @@ impl ComputeController { let (tx, rx) = oneshot::channel(); instance.call(move |i| { let result = i - .collections_hydrated_on_replicas(Some(replicas), &exclude_collections) + .collections_ready_on_replicas(Some(replicas), &exclude_collections, allowed_lag) .expect("validated"); let _ = tx.send(result); }); diff --git a/src/compute-client/src/controller/instance.rs b/src/compute-client/src/controller/instance.rs index 2f444d89f4755..766c81f744e32 100644 --- a/src/compute-client/src/controller/instance.rs +++ b/src/compute-client/src/controller/instance.rs @@ -15,6 +15,7 @@ use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use chrono::{DateTime, DurationRound, TimeDelta, Utc}; +use differential_dataflow::lattice::Lattice; use mz_build_info::BuildInfo; use mz_cluster_client::WallclockLagFn; use mz_compute_types::dataflows::{BuildDesc, DataflowDescription}; @@ -36,7 +37,7 @@ use mz_ore::{soft_assert_or_log, soft_panic_or_log}; use mz_persist_types::PersistLocation; use mz_repr::adt::timestamp::CheckedTimestamp; use mz_repr::refresh_schedule::RefreshSchedule; -use mz_repr::{Datum, Diff, GlobalId, RelationDesc, Row, Timestamp}; +use mz_repr::{Datum, Diff, GlobalId, RelationDesc, Row, Timestamp, frontier_within_lag}; use mz_storage_client::controller::{IntrospectionType, WallclockLag, WallclockLagHistogramPeriod}; use mz_storage_types::read_holds::{self, ReadHold}; use mz_storage_types::read_policy::ReadPolicy; @@ -722,12 +723,7 @@ impl Instance { return Ok(true); } for replica_state in hosting_replicas { - let collection_state = replica_state - .collections - .get(&collection_id) - .expect("hosting replica must have per-replica collection state"); - - if collection_state.hydrated() { + if replica_state.expect_collection(collection_id).hydrated() { return Ok(true); } } @@ -735,16 +731,46 @@ impl Instance { Ok(false) } - /// Returns `true` if each non-transient, non-excluded collection is hydrated on at - /// least one replica. + /// Returns `true` if each non-transient, non-excluded collection is *ready* on at + /// least one of the target replicas. + /// + /// A collection is ready on a replica when both hold: + /// + /// 1. The replica reports it hydrated, i.e. the dataflow has produced output + /// past the as-of it was installed with. + /// 2. When `allowed_lag` is `Some`, the replica's output frontier for the + /// collection is at most that far behind the furthest output frontier any + /// replica of this instance reports for it. + /// + /// Condition 1 alone is not a readiness signal for a cut-over. The as-of is + /// pinned when the replica is added and never moves, so a dataflow that spends + /// a long time on its initial snapshot reports hydrated the moment that + /// snapshot lands, with everything since the as-of still to replay. Condition 2 + /// is what says the replica has caught up with the replicas it is replacing: + /// while the outgoing replicas are still present, the furthest output frontier + /// is theirs. + /// + /// Both sides of condition 2 are *output* frontiers, not write frontiers. For a + /// materialized view the replica-reported write frontier is the persist shard's + /// upper, which every replica writing the shard shares, and for a `REFRESH` + /// materialized view it jumps to the next refresh time. The output frontier is + /// the meet of the write frontier and the dataflow's compute probe, so it is + /// the replica's own progress in both cases. For an index the two coincide. + /// See [`ReplicaCollectionState::hydrated`], which is defined over the output + /// frontier for the same reason. + /// + /// Passing `None` for `allowed_lag` checks condition 1 only, which is the + /// break-glass behavior for callers whose config disables the lag gate, and + /// what callers that genuinely only want to observe hydration pass. /// /// This also returns `true` in case this cluster does not have any /// replicas. #[mz_ore::instrument(level = "debug")] - pub fn collections_hydrated_on_replicas( + pub fn collections_ready_on_replicas( &self, target_replica_ids: Option>, exclude_collections: &BTreeSet, + allowed_lag: Option, ) -> Result { if self.replicas.is_empty() { return Ok(true); @@ -765,49 +791,59 @@ impl Instance { } let mut unhydrated = BTreeSet::new(); + let mut lagging = BTreeSet::new(); for (id, _collection) in self.collections_iter() { if id.is_transient() || exclude_collections.contains(&id) { continue; } - let mut collection_hydrated = false; - // `replicas_hosting` cannot fail here because `collections_iter` - // only yields collections that exist. - for replica_state in self.replicas_hosting(id).expect("collection must exist") { - if !target_replicas.contains(&replica_state.id) { - continue; + // Every replica hosting the collection, target or not: the + // classifier derives the lag reference from all of them and the + // verdict from the targets. `replicas_hosting` cannot fail here + // because `collections_iter` only yields collections that exist. + let replicas = self + .replicas_hosting(id) + .expect("collection must exist") + .map(|replica| { + let state = replica.expect_collection(id); + ReplicaCollectionView { + target: target_replicas.contains(&replica.id), + hydrated: state.hydrated(), + output_frontier: &state.output_frontier, + } + }); + + match classify_collection_readiness(replicas, allowed_lag) { + CollectionReadiness::Ready => {} + // We collect all not-ready collections instead of breaking out + // early, so that the log below names every collection the caller + // is waiting on, and why. + CollectionReadiness::Lagging => { + lagging.insert(id); } - let collection_state = replica_state - .collections - .get(&id) - .expect("hosting replica must have per-replica collection state"); - - if collection_state.hydrated() { - collection_hydrated = true; - break; + CollectionReadiness::Unhydrated => { + unhydrated.insert(id); } } - - if !collection_hydrated { - // We collect all non-hydrated collections instead of breaking - // out early, so that the log below names every collection the - // caller is waiting on. - unhydrated.insert(id); - } } - if !unhydrated.is_empty() { + if !unhydrated.is_empty() || !lagging.is_empty() { // Callers poll this on the cluster controller's reconcile tick, // which tests turn down to milliseconds, so this is deliberately - // one line per call rather than one per collection. + // one line per call rather than one per collection. The two reasons + // are reported separately: "hydrated but still behind" is the state + // a cut-over must not fire in, and it is indistinguishable from + // "ready" in the logs otherwise. tracing::info!( replicas = ?target_replicas, - collections = ?unhydrated, - "collections are not hydrated on any target replica", + unhydrated = ?unhydrated, + lagging = ?lagging, + ?allowed_lag, + "collections are not ready on any target replica", ); } - Ok(unhydrated.is_empty()) + Ok(unhydrated.is_empty() && lagging.is_empty()) } /// Clean up collection state that is not needed anymore. @@ -3172,6 +3208,18 @@ impl ReplicaState { self.collections.remove(&id) } + /// Returns the per-replica state of a collection this replica hosts. + /// + /// # Panics + /// + /// Panics if the replica does not host the collection. Callers obtain the + /// replica from [`Instance::replicas_hosting`], which guarantees it does. + fn expect_collection(&self, id: GlobalId) -> &ReplicaCollectionState { + self.collections + .get(&id) + .expect("hosting replica must have per-replica collection state") + } + /// Returns whether all replica frontiers of the given collection are empty. fn collection_frontiers_empty(&self, id: GlobalId) -> bool { self.collections.get(&id).map_or(true, |c| { @@ -3429,6 +3477,88 @@ impl Drop for ReplicaCollectionIntrospection { } } +/// The readiness of one collection across the replicas a caller is asking about. +/// +/// See [`classify_collection_readiness`]. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum CollectionReadiness { + /// At least one target replica has the collection hydrated and, if a lag + /// gate was given, within the allowance of the reference frontier. + Ready, + /// At least one target replica has the collection hydrated, but none of + /// those is within the allowance. This is the state a cut-over must not + /// fire in. + Lagging, + /// No target replica has the collection hydrated. + Unhydrated, +} + +/// One replica's view of one collection, as [`classify_collection_readiness`] +/// needs it. +#[derive(Debug, Clone, Copy)] +struct ReplicaCollectionView<'a> { + /// Whether the caller is asking about this replica. Non-target replicas + /// contribute to the lag reference but not to the verdict. + target: bool, + /// Whether the replica reports the collection hydrated. + hydrated: bool, + /// The replica's output frontier for the collection. + output_frontier: &'a Antichain, +} + +/// Classifies one collection's readiness over `replicas`, every replica hosting +/// the collection, target or not. +/// +/// The lag reference is the join of `output_frontier` over all of `replicas`: +/// the furthest any of them has progressed, which while the outgoing replicas of +/// a reconfiguration are still present is theirs. A target replica is ready when +/// it is hydrated and, if `allowed_lag` is `Some`, its output frontier is within +/// that allowance of the reference. `None` checks hydration only. An empty +/// reference means some replica reports the collection complete, so there is +/// nothing to be behind and the allowance is not applied. +/// +/// This is the whole decision, kept free of `Instance` so it can be tested over +/// plain frontiers. [`Instance::collections_ready_on_replicas`] supplies the +/// per-replica views. +fn classify_collection_readiness<'a>( + replicas: impl IntoIterator>, + allowed_lag: Option, +) -> CollectionReadiness { + let replicas: Vec<_> = replicas.into_iter().collect(); + + let lag = allowed_lag.and_then(|allowed_lag| { + let mut reference = Antichain::from_elem(Timestamp::MIN); + for replica in &replicas { + reference.join_assign(replica.output_frontier); + } + (!reference.is_empty()).then_some((reference, allowed_lag)) + }); + + let mut any_hydrated = false; + for replica in replicas.iter().filter(|r| r.target) { + if !replica.hydrated { + continue; + } + any_hydrated = true; + + let within_lag = match &lag { + Some((reference, allowed_lag)) => { + frontier_within_lag(replica.output_frontier, reference, *allowed_lag) + } + None => true, + }; + if within_lag { + return CollectionReadiness::Ready; + } + } + + if any_hydrated { + CollectionReadiness::Lagging + } else { + CollectionReadiness::Unhydrated + } +} + #[cfg(test)] mod tests { use std::collections::BTreeMap; @@ -3436,10 +3566,189 @@ mod tests { use mz_compute_types::dyncfgs::{ENABLE_COLUMN_PAGED_BATCHER, ENABLE_MZ_JOIN_CORE}; use mz_dyncfg::{ConfigSet, ConfigUpdates, ConfigVal}; use mz_persist_types::PersistLocation; + use mz_repr::Timestamp; + use timely::progress::Antichain; use crate::protocol::command::{ComputeCommand, InstanceConfig}; - use super::{Instance, ReplicaId}; + use super::{ + CollectionReadiness, Instance, ReplicaCollectionView, ReplicaId, + classify_collection_readiness, + }; + + fn ac(ts: u64) -> Antichain { + Antichain::from_elem(Timestamp::new(ts)) + } + + /// An outgoing replica: not a target, hydrated, at `output_frontier`. + fn outgoing(output_frontier: &Antichain) -> ReplicaCollectionView<'_> { + ReplicaCollectionView { + target: false, + hydrated: true, + output_frontier, + } + } + + /// A pending (target) replica. + fn pending( + hydrated: bool, + output_frontier: &Antichain, + ) -> ReplicaCollectionView<'_> { + ReplicaCollectionView { + target: true, + hydrated, + output_frontier, + } + } + + const LAG: Option = Some(Timestamp::new(60)); + + /// The regression the lag gate exists for: a pending replica whose dataflow + /// has produced its first output past the as-of, and so reports hydrated, + /// while its output frontier is still far behind the outgoing replica. + #[mz_ore::test] + fn hydrated_but_lagging_is_not_ready() { + let live = ac(10_000); + // Hydrated, four hours behind. + let behind = ac(1_000); + assert_eq!( + classify_collection_readiness([outgoing(&live), pending(true, &behind)], LAG), + CollectionReadiness::Lagging, + ); + // Hydrated, within the allowance. + let close = ac(9_950); + assert_eq!( + classify_collection_readiness([outgoing(&live), pending(true, &close)], LAG), + CollectionReadiness::Ready, + ); + // Not hydrated at all, however close. + assert_eq!( + classify_collection_readiness([outgoing(&live), pending(false, &close)], LAG), + CollectionReadiness::Unhydrated, + ); + } + + #[mz_ore::test] + fn reference_is_the_furthest_replica() { + // The reference is a join over every hosting replica. A freshly added + // target sitting at its as-of cannot drag it down, and a second outgoing + // replica further ahead raises it. + let ahead = ac(10_000); + let trailing = ac(9_000); + let fresh = ac(1_000); + let close_to_trailing = ac(8_990); + // Against `trailing` alone the pending replica would be within 60. + assert_eq!( + classify_collection_readiness( + [outgoing(&trailing), pending(true, &close_to_trailing)], + LAG + ), + CollectionReadiness::Ready, + ); + // With `ahead` also hosting, the reference moves up and it is not. + assert_eq!( + classify_collection_readiness( + [ + outgoing(&trailing), + outgoing(&ahead), + pending(true, &close_to_trailing), + pending(false, &fresh), + ], + LAG + ), + CollectionReadiness::Lagging, + ); + } + + #[mz_ore::test] + fn one_ready_target_suffices() { + let live = ac(10_000); + let behind = ac(1_000); + let close = ac(9_990); + assert_eq!( + classify_collection_readiness( + [ + outgoing(&live), + pending(true, &behind), + pending(false, &close), + pending(true, &close), + ], + LAG + ), + CollectionReadiness::Ready, + ); + // Lagging beats unhydrated when reporting why nothing is ready. + assert_eq!( + classify_collection_readiness( + [ + outgoing(&live), + pending(false, &close), + pending(true, &behind) + ], + LAG + ), + CollectionReadiness::Lagging, + ); + } + + #[mz_ore::test] + fn no_lag_gate_is_hydration_only() { + // `None` must be a true no-op on top of hydration: the same frontiers that + // read `Lagging` under a gate read `Ready` without one. This is the + // break-glass path and the burst strategy's path. + let live = ac(10_000); + let behind = ac(1_000); + assert_eq!( + classify_collection_readiness([outgoing(&live), pending(true, &behind)], None), + CollectionReadiness::Ready, + ); + assert_eq!( + classify_collection_readiness([outgoing(&live), pending(false, &behind)], None), + CollectionReadiness::Unhydrated, + ); + } + + #[mz_ore::test] + fn complete_collection_has_nothing_to_lag_behind() { + // One replica reporting the collection complete (empty output frontier) + // makes the join empty. The allowance does not apply; a hydrated target + // at any frontier is ready. + let complete = Antichain::new(); + let behind = ac(1_000); + assert_eq!( + classify_collection_readiness( + [outgoing(&complete), pending(true, &behind)], + Some(Timestamp::new(0)) + ), + CollectionReadiness::Ready, + ); + } + + #[mz_ore::test] + fn sole_replica_is_its_own_reference() { + // After the outgoing replicas are gone, or on a brand-new cluster, the + // target is the furthest replica, so it is within any allowance of + // itself, including zero. The gate cannot wedge in steady state. + let alone = ac(5_000); + assert_eq!( + classify_collection_readiness([pending(true, &alone)], Some(Timestamp::new(0))), + CollectionReadiness::Ready, + ); + } + + #[mz_ore::test] + fn no_targets_is_unhydrated() { + // Nothing to judge: none of the asked-about replicas hosts the + // collection (e.g. it is pinned to an outgoing replica). Reported as + // unhydrated so a caller waiting on it keeps waiting rather than cutting + // over on a vacuous truth. This is why callers exclude replica-pinned + // collections they know the targets can never host. + let live = ac(10_000); + assert_eq!( + classify_collection_readiness([outgoing(&live)], LAG), + CollectionReadiness::Unhydrated, + ); + } fn create_instance_command() -> ComputeCommand { ComputeCommand::CreateInstance(Box::new(InstanceConfig { diff --git a/src/repr/src/lib.rs b/src/repr/src/lib.rs index 4994d487a4c51..b7c9a830526ed 100644 --- a/src/repr/src/lib.rs +++ b/src/repr/src/lib.rs @@ -80,7 +80,7 @@ pub use crate::scalar::{ PropArray, PropDatum, PropDict, PropList, arb_datum, arb_datum_for_column, arb_datum_for_scalar, arb_range_type, }; -pub use crate::timestamp::{Timestamp, TimestampManipulation}; +pub use crate::timestamp::{Timestamp, TimestampManipulation, frontier_within_lag}; pub use crate::update::{ Rows, RowsBuilder, SharedSlice, UpdateCollection, UpdateCollectionBuilder, }; diff --git a/src/repr/src/timestamp.rs b/src/repr/src/timestamp.rs index 58194ec2473be..6c90f72712ece 100644 --- a/src/repr/src/timestamp.rs +++ b/src/repr/src/timestamp.rs @@ -620,3 +620,89 @@ impl TryFrom for Timestamp { impl columnation::Columnation for Timestamp { type InnerRegion = columnation::CopyRegion; } + +/// Returns whether `frontier` is at most `allowed_lag` behind `reference`. +/// +/// The predicate is `reference <= frontier + allowed_lag`, evaluated on +/// antichains. Timestamps that would overflow when advanced by `allowed_lag` +/// saturate at [`Timestamp::MAX`] rather than panicking, so a frontier close to +/// the end of the timestamp domain reads as far ahead rather than aborting the +/// caller. +/// +/// Note the two degenerate cases, which callers usually want to handle +/// themselves rather than inherit: +/// +/// * An empty `reference` is the maximum antichain, so the result is `true` +/// only when `frontier` is empty too. A caller that reads an empty reference +/// as "this collection is complete, nothing can be behind it" must say so at +/// its own call site. +/// * An empty `frontier` is likewise the maximum, so the result is `true` for +/// every `reference`. +pub fn frontier_within_lag( + frontier: &timely::progress::Antichain, + reference: &timely::progress::Antichain, + allowed_lag: Timestamp, +) -> bool { + // We cannot subtract frontiers, so bump `frontier` forward by the allowance + // and compare that against `reference`. + let bumped = timely::progress::Antichain::from_iter(frontier.iter().map(|t| { + t.try_step_forward_by(&allowed_lag) + .unwrap_or(Timestamp::MAX) + })); + timely::order::PartialOrder::less_equal(reference, &bumped) +} + +#[cfg(test)] +mod frontier_within_lag_tests { + use timely::progress::Antichain; + + use super::{Timestamp, frontier_within_lag}; + + fn ac(ts: u64) -> Antichain { + Antichain::from_elem(Timestamp::new(ts)) + } + + #[mz_ore::test] + fn within_and_beyond_the_allowance() { + // Exactly at the allowance is within it; one past it is not. + assert!(frontier_within_lag(&ac(100), &ac(110), Timestamp::new(10))); + assert!(!frontier_within_lag(&ac(100), &ac(111), Timestamp::new(10))); + // A frontier ahead of the reference is trivially within any allowance. + assert!(frontier_within_lag(&ac(200), &ac(110), Timestamp::new(0))); + } + + #[mz_ore::test] + fn monotone_in_the_allowance() { + // Widening the allowance never turns a `true` into a `false`. + for lag in 0..40u64 { + let narrow = frontier_within_lag(&ac(100), &ac(120), Timestamp::new(lag)); + let wide = frontier_within_lag(&ac(100), &ac(120), Timestamp::new(lag + 1)); + assert!(wide || !narrow, "narrow={narrow} wide={wide} lag={lag}"); + } + } + + #[mz_ore::test] + fn empty_frontiers_are_the_maximum() { + let empty = Antichain::new(); + // An empty frontier is ahead of everything. + assert!(frontier_within_lag( + &empty, + &ac(u64::MAX), + Timestamp::new(0) + )); + // An empty reference is only matched by an empty frontier. + assert!(!frontier_within_lag(&ac(100), &empty, Timestamp::new(10))); + assert!(frontier_within_lag(&empty, &empty, Timestamp::new(10))); + } + + #[mz_ore::test] + fn saturates_instead_of_panicking_on_overflow() { + // `step_forward_by` would panic here; the saturating form reads as + // "arbitrarily far ahead" instead. + assert!(frontier_within_lag( + &ac(u64::MAX - 1), + &ac(u64::MAX), + Timestamp::new(u64::MAX), + )); + } +} diff --git a/test/launchdarkly-flag-consistency/mzcompose.py b/test/launchdarkly-flag-consistency/mzcompose.py index 860d76ab2ee0b..fbb21b846f92d 100644 --- a/test/launchdarkly-flag-consistency/mzcompose.py +++ b/test/launchdarkly-flag-consistency/mzcompose.py @@ -198,6 +198,7 @@ cluster_controller_tick_interval cluster_enable_topology_spread cluster_multi_process_replica_az_affinity_weight + cluster_reconfiguration_allowed_lag cluster_soften_az_affinity cluster_soften_az_affinity_weight cluster_soften_replication_anti_affinity @@ -239,6 +240,7 @@ enable_any_all_null_array_semantics enable_auto_scaling_strategy enable_background_alter_cluster + enable_cluster_reconfiguration_lag_gate enable_statement_arrival_logging enable_binary_date_bin enable_coalesce_case_transform