From bec041de0a3d16676e22030865ca977474a0c7a2 Mon Sep 17 00:00:00 2001 From: Aljoscha Krettek Date: Mon, 10 Aug 2026 12:52:46 +0000 Subject: [PATCH] cluster-controller: batch ON REFRESH oracle reads The controller requested refresh-window inputs one cluster at a time. Each request issued and awaited a timestamp oracle read, so a slow oracle made every scheduled cluster add another round trip before unrelated clusters could reconcile. Gather each scheduled cluster's catalog and storage inputs in a separate coordinator turn, then attach one shared oracle timestamp to the completed phase batch. Skip clusters whose inputs are unavailable instead of treating them as outside their refresh window. Add seam tests for one batch per phase and unavailable input handling. Add an mzcompose regression that binds a periodic-refresh MV to each scheduled cluster and compares probe-cluster convergence with zero and eight scheduled clusters under injected oracle latency. Document why serial oracle awaits cannot coalesce and when one shared read is valid. Fixes SQL-569. --- doc/developer/guide-adapter.md | 17 +- src/adapter/src/coord/cluster_controller.rs | 113 +++++++----- src/cluster-controller/src/ctx.rs | 63 ++++--- src/cluster-controller/src/lib.rs | 53 +++++- src/cluster-controller/src/tests.rs | 182 +++++++++++++++++-- test/cluster/mzcompose.py | 185 +++++++++++++++++++- 6 files changed, 513 insertions(+), 100 deletions(-) diff --git a/doc/developer/guide-adapter.md b/doc/developer/guide-adapter.md index 5f33d134329df..d00fd58d2a9f5 100644 --- a/doc/developer/guide-adapter.md +++ b/doc/developer/guide-adapter.md @@ -91,12 +91,17 @@ This means: #### Why the batching oracle is correct -`BatchingTimestampOracle` collects multiple `read_ts` requests that arrive -concurrently and serves them all with a single call to the backing oracle. -Because the backing oracle is called *during* all of their real-time intervals -(after all requests arrived, before any have returned), the returned timestamp -is within bounds for every request. Batching can only push timestamps later, -never earlier. See the comment on the `BatchingTimestampOracle` struct. +`BatchingTimestampOracle` drains the queued `read_ts` requests and serves each +collected batch with one call to the backing oracle. That call occurs during +every collected request's real-time interval, after each request arrived and +before any returns. Its timestamp is therefore within bounds for every request. +Batching can only push timestamps later, never earlier. + +Coalescing is opportunistic. A request that overlaps a backing call but arrives +after the queue is drained waits for a later call. A serial await loop +guarantees that its calls cannot coalesce. When exactly one round trip is +required, use one explicit shared call only if it occurs within every +operation's real-time bounds and satisfies every caller's contract. #### Why caching an oracle result is not correct diff --git a/src/adapter/src/coord/cluster_controller.rs b/src/adapter/src/coord/cluster_controller.rs index a226054624e60..cd17a499ae059 100644 --- a/src/adapter/src/coord/cluster_controller.rs +++ b/src/adapter/src/coord/cluster_controller.rs @@ -14,9 +14,10 @@ //! the controller as a **separate task** and implements the ctx by marshaling //! each pull/apply to the Coordinator over the internal command channel, because //! the catalog and the live compute/storage signals are reachable only from the -//! coordinator loop. The two whole-tick reads are batched; the per-cluster live -//! signals are pulled on demand, so a tick's round-trips scale with the number of -//! managed clusters that need a live signal, not with a constant. +//! coordinator loop. Whole-tick reads are batched. Refresh-window catalog inputs +//! are pulled one cluster at a time and completed with one shared oracle read. +//! The remaining per-cluster live signals are pulled on demand, so steady +//! clusters do not pay for signals they do not use. //! //! The controller owns the replica set of every managed cluster, user and //! system alike. A builtin cluster's config-implied replicas are additionally @@ -24,7 +25,7 @@ //! derives the same target from the same config, so the two converge rather //! than compete. -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::sync::Arc; use std::time::Duration; @@ -34,7 +35,8 @@ use mz_cluster_controller::ClusterController; use mz_cluster_controller::ctx::{ ApplyOutcome, AvailabilityZones, ClusterControllerCtx, ClusterState, CreateReason, Decision, ExpectedClusterState, ObservedReplica, OnTimeout, ReconfigurationRecord, ReconfigurationStatus, - ReconfigurationTarget, RefreshMvInfo, RefreshWindowInputs, ReplicaShape, StateWrite, + ReconfigurationTarget, RefreshMvInfo, RefreshWindowClusterInputs, RefreshWindowInputsBatch, + ReplicaShape, StateWrite, }; use mz_compute_types::config::ComputeReplicaConfig; use mz_controller::clusters::ClusterStatus; @@ -52,8 +54,9 @@ use crate::error::AdapterError; /// [`ClusterControllerCtx`] call. Each variant carries a oneshot for the reply. /// /// `ManagedClusterIds` and `ClusterStates` are the per-tick batched reads. The -/// `ClusterStates` reply also carries `now`. `HydratedReplicas` is a -/// per-cluster live signal a strategy pulls on demand. +/// `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. #[derive(Debug)] pub enum ClusterControllerRequest { /// The ids of all managed clusters the controller owns this tick. @@ -77,13 +80,14 @@ pub enum ClusterControllerRequest { cluster_id: ClusterId, tx: oneshot::Sender, }, - /// The refresh-window live signals for one scheduled cluster (read ts, - /// compaction estimate, bound REFRESH MVs). `None` for a cluster that is not - /// scheduled `ON REFRESH`. - RefreshWindowInputs { + /// The catalog and storage refresh-window inputs for one scheduled cluster. + /// `None` if the cluster no longer qualifies at pull time. + RefreshWindowClusterInputs { cluster_id: ClusterId, - tx: oneshot::Sender>, + tx: oneshot::Sender>, }, + /// One timestamp-oracle read for a completed refresh-window input batch. + RefreshWindowReadTs { tx: oneshot::Sender }, /// Apply a tick's batch of decisions under their compare-and-append guards. Apply { decisions: Vec, @@ -180,11 +184,38 @@ impl ClusterControllerCtx for CoordCtx { async fn refresh_window_inputs( &mut self, - cluster_id: ClusterId, - ) -> Option { - self.request(|tx| ClusterControllerRequest::RefreshWindowInputs { cluster_id, tx }) - .await - .flatten() + cluster_ids: &[ClusterId], + ) -> Option { + let mut cluster_inputs = BTreeMap::new(); + for (index, &cluster_id) in cluster_ids.iter().enumerate() { + if index > 0 { + // The coordinator prioritizes its internal command channel. Give + // it a chance to service already-queued user commands instead of + // keeping that channel continuously ready for the whole batch. + tokio::task::yield_now().await; + } + let inputs = self + .request(|tx| ClusterControllerRequest::RefreshWindowClusterInputs { + cluster_id, + tx, + }) + .await + .flatten(); + if let Some(inputs) = inputs { + cluster_inputs.insert(cluster_id, inputs); + } + } + if cluster_inputs.is_empty() { + return None; + } + + let read_ts = self + .request(|tx| ClusterControllerRequest::RefreshWindowReadTs { tx }) + .await?; + Some(RefreshWindowInputsBatch { + read_ts, + cluster_inputs, + }) } async fn apply(&mut self, decisions: Vec) -> ApplyOutcome { @@ -309,35 +340,18 @@ impl Coordinator { ClusterControllerRequest::HasHydratableObjects { cluster_id, tx } => { let _ = tx.send(self.cluster_has_hydratable_objects(cluster_id)); } - ClusterControllerRequest::RefreshWindowInputs { cluster_id, tx } => { - // Gather the catalog- and storage-derived inputs on the loop, - // then complete the reply from a spawned task: the oracle - // read is a network round-trip (to the Postgres/CRDB-backed - // timestamp oracle) and must never run on the serial + ClusterControllerRequest::RefreshWindowClusterInputs { cluster_id, tx } => { + let _ = tx.send(self.refresh_window_catalog_inputs(cluster_id)); + } + ClusterControllerRequest::RefreshWindowReadTs { tx } => { + // The oracle read is a network round trip to the Postgres or + // CRDB-backed timestamp oracle. It must not run on the serial // coordinator loop. - match self.refresh_window_catalog_inputs(cluster_id) { - None => { - let _ = tx.send(None); - } - Some((compaction_estimate, refresh_mvs)) => { - let oracle = self.get_local_timestamp_oracle(); - // NOTE: this is one oracle read per scheduled cluster - // per tick, and the controller awaits each pull before - // the next, so the reads are sequential and the - // batching oracle cannot coalesce them. Fine at the - // tick cadence for realistic scheduled-cluster counts. - // TODO: hoist to one read per tick if that stops - // holding. - spawn(|| "cluster_controller_refresh_window_read_ts", async move { - let read_ts = oracle.read_ts().await; - let _ = tx.send(Some(RefreshWindowInputs { - read_ts, - compaction_estimate, - refresh_mvs, - })); - }); - } - } + let oracle = self.get_local_timestamp_oracle(); + spawn(|| "cluster_controller_refresh_window_read_ts", async move { + let read_ts = oracle.read_ts().await; + let _ = tx.send(read_ts); + }); } ClusterControllerRequest::Apply { decisions, tx } => { let outcome = if active { @@ -495,7 +509,7 @@ impl Coordinator { /// `None` if the cluster is missing, unmanaged, or not scheduled `ON /// REFRESH`. /// - /// The oracle read timestamp completing [`RefreshWindowInputs`] is + /// The oracle read timestamp completing [`RefreshWindowInputsBatch`] is /// deliberately not fetched here: this runs on the coordinator loop, and /// the oracle read is a network round-trip the request handler performs on /// a spawned task instead. @@ -506,7 +520,7 @@ impl Coordinator { fn refresh_window_catalog_inputs( &self, cluster_id: ClusterId, - ) -> Option<(Duration, Vec)> { + ) -> Option { use mz_catalog::memory::objects::CatalogItem; let cluster = self.catalog().try_get_cluster(cluster_id)?; @@ -549,7 +563,10 @@ impl Coordinator { .system_config() .cluster_refresh_mv_compaction_estimate(); - Some((compaction_estimate, refresh_mvs)) + Some(RefreshWindowClusterInputs { + compaction_estimate, + refresh_mvs, + }) } /// Apply one batch of decisions under their compare-and-append guards. diff --git a/src/cluster-controller/src/ctx.rs b/src/cluster-controller/src/ctx.rs index e27a5f2ba4974..d2a0d2080f1dc 100644 --- a/src/cluster-controller/src/ctx.rs +++ b/src/cluster-controller/src/ctx.rs @@ -24,7 +24,7 @@ //! controller drives what is fetched. Read methods are batched so a separate-task //! deployment can bound its round-trips to the Coordinator. -use std::collections::BTreeSet; +use std::collections::{BTreeMap, BTreeSet}; use std::time::Duration; use async_trait::async_trait; @@ -175,16 +175,36 @@ impl RefreshWindowDecision { } } -/// The live signals the on-refresh strategy reads to decide whether a scheduled -/// cluster is inside a refresh window: the current read timestamp, the -/// Persist-compaction time estimate, and the bound REFRESH MVs' frontiers and -/// schedules. +/// The catalog and storage inputs for one scheduled cluster's refresh window. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RefreshWindowClusterInputs { + /// How long after a refresh an MV is estimated to still need Persist + /// compaction, which also keeps the cluster on. + pub compaction_estimate: Duration, + /// The REFRESH MVs bound to the cluster. + pub refresh_mvs: Vec, +} + +/// Refresh-window inputs gathered for one reconciliation phase. /// -/// Pulled on demand only for scheduled clusters. A MANUAL cluster carries `None` -/// and is never probed. +/// The top-level timestamp makes sharing one oracle read across every included +/// cluster structural. A cluster absent from `cluster_inputs` has unavailable +/// inputs and must not be reconciled during the phase. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct RefreshWindowInputsBatch { + /// The local oracle read timestamp for every cluster in the batch. + pub read_ts: Timestamp, + /// The available catalog and storage inputs, keyed by cluster. + pub cluster_inputs: BTreeMap, +} + +/// The fulfilled live signal the on-refresh strategy uses for one cluster. +/// +/// Pulled on demand only for scheduled clusters. A MANUAL cluster carries +/// `None` and is never probed. #[derive(Clone, Debug, PartialEq, Eq)] pub struct RefreshWindowInputs { - /// The local oracle read timestamp the window decision is taken against. + /// The shared local oracle read timestamp the window decision uses. pub read_ts: Timestamp, /// How long after a refresh an MV is estimated to still need Persist /// compaction, which also keeps the cluster on. @@ -416,21 +436,24 @@ pub trait ClusterControllerCtx: Send { /// burst winds down via its linger. async fn has_hydratable_objects(&mut self, cluster_id: ClusterId) -> bool; - /// The refresh-window live signals for one scheduled cluster: the read - /// timestamp, the compaction estimate, and the bound REFRESH MVs' write - /// frontiers and schedules. Returns `None` when the cluster is missing, - /// unmanaged, or no longer scheduled `ON REFRESH` at pull time. The - /// controller only asks about clusters it observed as scheduled, so `None` - /// means a concurrent DDL moved the cluster mid-tick, and the schedule's - /// membership in the compare-and-append witness rejects any decision - /// derived from the stale observation. + /// The refresh-window live signals for the given scheduled clusters. + /// Returns one shared read timestamp plus the available per-cluster catalog + /// and storage inputs. Omits a cluster when its inputs are unavailable, + /// including when it is missing, unmanaged, or no longer scheduled `ON + /// REFRESH` at pull time. Returns `None` when the batch fails or no requested + /// cluster has valid inputs. /// /// Pulled on demand the same way as [`Self::hydrated_replicas`]: the - /// controller probes a cluster only when the on-refresh strategy needs the + /// controller includes a cluster only when the on-refresh strategy needs the /// signal (i.e. the cluster is scheduled), so a steady MANUAL cluster never - /// pays for it. - async fn refresh_window_inputs(&mut self, cluster_id: ClusterId) - -> Option; + /// pays for it. Implementations fetch the shared read timestamp once after + /// gathering the per-cluster inputs, so oracle latency does not scale with + /// the cluster count. The controller skips any omitted cluster for the + /// reconciliation phase. + async fn refresh_window_inputs( + &mut self, + cluster_ids: &[ClusterId], + ) -> Option; /// Apply a tick's batch of decisions under their compare-and-append guards. /// Each decision carries the [`ExpectedClusterState`] it was derived from; diff --git a/src/cluster-controller/src/lib.rs b/src/cluster-controller/src/lib.rs index 7b86b58eade7a..9fac1055f9a58 100644 --- a/src/cluster-controller/src/lib.rs +++ b/src/cluster-controller/src/lib.rs @@ -46,7 +46,7 @@ use mz_ore::soft_panic_or_log; use crate::ctx::{ ApplyOutcome, ClusterControllerCtx, ClusterState, CreateReason, Decision, ObservedReplica, ReconfigurationAudit, ReconfigurationRecord, ReconfigurationStatus, ReconfigurationWrite, - ReplicaShape, StateWrite, + RefreshWindowInputs, ReplicaShape, StateWrite, }; use crate::strategy::{ BaselineStrategy, ConfigSignals, DesiredReplica, GracefulReconfigurationStrategy, @@ -131,7 +131,10 @@ impl ClusterController { // that is probably about to go stale. let mut rejected = BTreeSet::new(); for state in &states { - let write = self.merge_state_writes(state, &signals[&state.cluster_id], &config, now); + let Some(signals) = signals.get(&state.cluster_id) else { + continue; + }; + let write = self.merge_state_writes(state, signals, &config, now); if write.is_empty() { continue; } @@ -166,8 +169,10 @@ impl ClusterController { if rejected.contains(&state.cluster_id) { continue; } - let decisions = - self.collect_replica_decisions(state, &signals[&state.cluster_id], &config, now); + let Some(signals) = signals.get(&state.cluster_id) else { + continue; + }; + let decisions = self.collect_replica_decisions(state, signals, &config, now); if decisions.is_empty() { continue; } @@ -325,10 +330,12 @@ impl ClusterController { /// /// Each strategy names its needs as a pure function of the durable state /// and the tick's config signals ([`Strategy::signal_request`]), so the - /// kernel stays ignorant of when a strategy engages. Signals are fetched per - /// cluster and only where requested: a steady cluster is never probed, - /// keeping the ctx seam pay-for-what-you-use. The returned map has an entry - /// for every state. + /// kernel stays ignorant of when a strategy engages. Signals are fetched + /// only where requested: a steady cluster is never probed, keeping the ctx + /// seam pay-for-what-you-use. Refresh-window inputs are fetched as one batch + /// so every scheduled cluster shares one oracle read per phase. The returned + /// map omits a state when one of its required inputs was unavailable, which + /// causes the reconciliation phase to skip that cluster. async fn fetch_signals( &self, ctx: &mut dyn ClusterControllerCtx, @@ -336,6 +343,7 @@ impl ClusterController { config: &ConfigSignals, ) -> BTreeMap { let mut signals = BTreeMap::new(); + let mut refresh_window_clusters = Vec::new(); for state in states { let request = self .strategies @@ -360,10 +368,37 @@ impl ClusterController { } } if request.refresh_window { - live.refresh_window = ctx.refresh_window_inputs(state.cluster_id).await; + refresh_window_clusters.push(state.cluster_id); } signals.insert(state.cluster_id, live); } + if !refresh_window_clusters.is_empty() { + match ctx.refresh_window_inputs(&refresh_window_clusters).await { + Some(batch) => { + let read_ts = batch.read_ts; + let mut cluster_inputs = batch.cluster_inputs; + for cluster_id in refresh_window_clusters { + let Some(inputs) = cluster_inputs.remove(&cluster_id) else { + signals.remove(&cluster_id); + continue; + }; + let live = signals + .get_mut(&cluster_id) + .expect("signal entry inserted for requested cluster"); + live.refresh_window = Some(RefreshWindowInputs { + read_ts, + compaction_estimate: inputs.compaction_estimate, + refresh_mvs: inputs.refresh_mvs, + }); + } + } + None => { + for cluster_id in refresh_window_clusters { + signals.remove(&cluster_id); + } + } + } + } signals } diff --git a/src/cluster-controller/src/tests.rs b/src/cluster-controller/src/tests.rs index 19e51834eac15..d311bf69d8445 100644 --- a/src/cluster-controller/src/tests.rs +++ b/src/cluster-controller/src/tests.rs @@ -29,7 +29,8 @@ use crate::ClusterController; use crate::ctx::{ ApplyOutcome, AutoScalingPolicy, AvailabilityZones, BurstAudit, ClusterControllerCtx, ClusterSchedule, ClusterState, CreateReason, Decision, ObservedReplica, ReconfigurationAudit, - ReconfigurationStatus, RefreshWindowInputs, ReplicaShape, StateWrite, + ReconfigurationStatus, RefreshWindowClusterInputs, RefreshWindowInputs, + RefreshWindowInputsBatch, ReplicaShape, StateWrite, }; use crate::strategy::{ConfigSignals, DesiredReplica, LiveSignals, Strategy}; @@ -161,10 +162,10 @@ struct FakeCtx { /// `has_hydratable_objects` pull, keeping that pull load-bearing for /// the seam tests. has_hydratable_objects: BTreeMap, - /// Refresh-window inputs the fake returns per cluster when the controller - /// probes a scheduled cluster. An on-refresh test sets this to drive the - /// window decision. - refresh_window: BTreeMap, + /// Refresh-window inputs the fake returns for a controller probe. + refresh_window: Option, + /// The cluster ids in each batched refresh-window probe. + refresh_window_probes: Vec>, } impl FakeCtx { @@ -181,10 +182,36 @@ impl FakeCtx { hydrated: BTreeSet::new(), hydration_probes: 0, has_hydratable_objects: BTreeMap::new(), - refresh_window: BTreeMap::new(), + refresh_window: None, + refresh_window_probes: Vec::new(), } } + fn set_refresh_window(&mut self, cluster_id: ClusterId, inputs: RefreshWindowInputs) { + let RefreshWindowInputs { + read_ts, + compaction_estimate, + refresh_mvs, + } = inputs; + let batch = self + .refresh_window + .get_or_insert_with(|| RefreshWindowInputsBatch { + read_ts, + cluster_inputs: BTreeMap::new(), + }); + assert_eq!( + batch.read_ts, read_ts, + "one batch must use one shared read timestamp" + ); + batch.cluster_inputs.insert( + cluster_id, + RefreshWindowClusterInputs { + compaction_estimate, + refresh_mvs, + }, + ); + } + /// All create decisions across every applied batch. fn creates(&self) -> Vec<&Decision> { self.applied @@ -243,9 +270,28 @@ impl ClusterControllerCtx for FakeCtx { async fn refresh_window_inputs( &mut self, - cluster_id: ClusterId, - ) -> Option { - self.refresh_window.get(&cluster_id).cloned() + cluster_ids: &[ClusterId], + ) -> Option { + self.refresh_window_probes.push(cluster_ids.to_vec()); + let batch = self.refresh_window.as_ref()?; + let cluster_inputs: BTreeMap<_, _> = cluster_ids + .iter() + .filter_map(|cluster_id| { + batch + .cluster_inputs + .get(cluster_id) + .cloned() + .map(|inputs| (*cluster_id, inputs)) + }) + .collect(); + if cluster_inputs.is_empty() { + None + } else { + Some(RefreshWindowInputsBatch { + read_ts: batch.read_ts, + cluster_inputs, + }) + } } async fn apply(&mut self, decisions: Vec) -> ApplyOutcome { @@ -2390,6 +2436,114 @@ fn on_refresh_compaction_window_keeps_cluster_on() { ); } +#[mz_ore::test(tokio::test)] +async fn on_refresh_batches_window_inputs_once_per_phase() { + let c1 = cluster(1); + let c2 = cluster(2); + let manual = cluster(3); + let (scheduled1, _) = scheduled_state(c1, "100cc", 1, 0, Vec::new(), None); + let (scheduled2, _) = scheduled_state(c2, "100cc", 1, 0, Vec::new(), None); + let mut ctx = FakeCtx::new(vec![ + scheduled1, + scheduled2, + state(manual, "100cc", 0, Vec::new()), + ]); + let closed_window = window_inputs(100, 0, Some(200), refresh_at(50)); + ctx.set_refresh_window(c1, closed_window.clone()); + ctx.set_refresh_window(c2, closed_window); + + controller().reconcile(&mut ctx).await; + + assert_eq!( + ctx.refresh_window_probes, + vec![vec![c1, c2], vec![c1, c2]], + "one batch per phase, excluding the MANUAL cluster", + ); +} + +#[mz_ore::test(tokio::test)] +async fn on_refresh_unavailable_window_inputs_skip_only_affected_cluster() { + let scheduled = cluster(1); + let manual = cluster(2); + let available = cluster(3); + let (unavailable_state, _) = scheduled_state( + scheduled, + "100cc", + 1, + 0, + vec![observed(replica(1), "r0", "100cc")], + None, + ); + let (available_state, _) = scheduled_state(available, "100cc", 0, 0, Vec::new(), None); + let mut ctx = FakeCtx::new(vec![ + unavailable_state, + state(manual, "100cc", 1, Vec::new()), + available_state, + ]); + ctx.set_refresh_window(available, window_inputs(100, 0, Some(50), refresh_at(1000))); + + controller().reconcile(&mut ctx).await; + + assert_eq!(ctx.refresh_window_probes, vec![vec![scheduled, available]]); + assert_eq!(ctx.states[&scheduled].replication_factor, 1); + assert_eq!(ctx.states[&scheduled].replicas.len(), 1); + assert!(ctx.applied.iter().flatten().all(|decision| { + let cluster_id = match decision { + Decision::CreateReplica { cluster_id, .. } + | Decision::DropReplica { cluster_id, .. } + | Decision::UpdateClusterState { cluster_id, .. } => cluster_id, + }; + *cluster_id != scheduled + })); + let created_clusters: BTreeSet<_> = ctx + .creates() + .into_iter() + .map(|decision| match decision { + Decision::CreateReplica { cluster_id, .. } => *cluster_id, + _ => unreachable!("creates returns only create decisions"), + }) + .collect(); + assert_eq!(created_clusters, BTreeSet::from([manual, available])); + assert_eq!(ctx.states[&manual].replicas.len(), 1); + assert_eq!(ctx.states[&available].replicas.len(), 1); + assert!(ctx.drops().is_empty()); +} + +#[mz_ore::test(tokio::test)] +async fn on_refresh_unavailable_window_input_batch_skips_scheduled_clusters() { + let stale_rf = cluster(1); + let running = cluster(2); + let manual = cluster(3); + let (stale_rf_state, _) = scheduled_state(stale_rf, "100cc", 1, 0, Vec::new(), None); + let (running_state, _) = scheduled_state( + running, + "100cc", + 0, + 0, + vec![observed(replica(1), "r0", "100cc")], + None, + ); + let mut ctx = FakeCtx::new(vec![ + stale_rf_state, + running_state, + state(manual, "100cc", 1, Vec::new()), + ]); + + controller().reconcile(&mut ctx).await; + + assert_eq!(ctx.refresh_window_probes, vec![vec![stale_rf, running]]); + assert_eq!(ctx.states[&stale_rf].replication_factor, 1); + assert_eq!(ctx.states[&running].replicas.len(), 1); + let creates = ctx.creates(); + assert_eq!(creates.len(), 1, "the MANUAL cluster still reconciles"); + assert!(matches!( + creates[0], + Decision::CreateReplica { cluster_id, .. } if *cluster_id == manual + )); + assert_eq!(ctx.states[&manual].replicas.len(), 1); + assert!(ctx.drops().is_empty()); +} + #[mz_ore::test(tokio::test)] async fn on_refresh_creates_in_window_through_seam() { // End-to-end through the ctx seam: a scheduled cluster with a stale rf=1 and @@ -2398,8 +2552,7 @@ async fn on_refresh_creates_in_window_through_seam() { let c = cluster(1); let (state, _signals) = scheduled_state(c, "100cc", 1, 0, Vec::new(), None); let mut ctx = FakeCtx::new(vec![state]); - ctx.refresh_window - .insert(c, window_inputs(100, 0, Some(50), refresh_at(1000))); + ctx.set_refresh_window(c, window_inputs(100, 0, Some(50), refresh_at(1000))); let controller = controller(); controller.reconcile(&mut ctx).await; @@ -2451,8 +2604,7 @@ async fn on_refresh_schedule_alter_rejects_in_flight_decision() { None, ); let mut ctx = FakeCtx::new(vec![state]); - ctx.refresh_window - .insert(c, window_inputs(100, 0, Some(200), refresh_at(50))); + ctx.set_refresh_window(c, window_inputs(100, 0, Some(200), refresh_at(50))); ctx.witness_check = true; // The `ALTER` flips only the schedule (rf, size, azs, logging unchanged), so // the rejection is attributable solely to the witness `schedule` field. @@ -2499,8 +2651,7 @@ async fn on_refresh_unchanged_schedule_passes_witness() { None, ); let mut ctx = FakeCtx::new(vec![state]); - ctx.refresh_window - .insert(c, window_inputs(100, 0, Some(200), refresh_at(50))); + ctx.set_refresh_window(c, window_inputs(100, 0, Some(200), refresh_at(50))); ctx.witness_check = true; let controller = controller(); @@ -2619,6 +2770,7 @@ async fn on_refresh_graceful_record_settles_then_normalizes() { // so the first tick cuts over without waiting for hydration. state.reconfiguration = Some(record_on_timeout("200cc", 2, 500, OnTimeout::Commit)); let mut ctx = FakeCtx::new(vec![state]); + ctx.set_refresh_window(c, window_inputs(100, 0, Some(200), refresh_at(50))); let controller = controller(); controller.reconcile(&mut ctx).await; diff --git a/test/cluster/mzcompose.py b/test/cluster/mzcompose.py index ab42df766d60f..73c6dbfa93af4 100644 --- a/test/cluster/mzcompose.py +++ b/test/cluster/mzcompose.py @@ -21,14 +21,14 @@ from collections.abc import Callable from copy import copy from datetime import datetime, timedelta -from statistics import quantiles +from statistics import median, quantiles from textwrap import dedent from threading import Event, Thread import psycopg import requests import websocket -from psycopg import Cursor +from psycopg import Cursor, sql from psycopg.errors import ( DatabaseError, InternalError_, @@ -7828,3 +7828,184 @@ def workflow_test_metrics_null_label(c: Composition) -> None: assert c.sql_query("SELECT 1", reuse_connection=False)[0][0] == 1 finally: c.sql("DROP CLUSTER sql198_unmgd CASCADE", port=6877, user="mz_system") + + +def workflow_test_controller_oracle_stall( + c: Composition, parser: WorkflowArgumentParser +) -> None: + """Scheduled-cluster count must not determine unrelated reconciliation latency.""" + parser.add_argument("--latency-ms", type=int, default=500) + parser.add_argument("--scheduled-clusters", type=int, default=8) + args = parser.parse_args() + if args.latency_ms < 100: + parser.error("--latency-ms must be at least 100") + if args.scheduled_clusters < 8: + parser.error("--scheduled-clusters must be at least 8") + oracle_port = 26258 + + def set_latency(toxi: str, latency_ms: int) -> None: + requests.delete(f"{toxi}/proxies/oracle/toxics/lat") + if latency_ms > 0: + response = requests.post( + f"{toxi}/proxies/oracle/toxics", + json={ + "name": "lat", + "type": "latency", + "attributes": {"latency": latency_ms, "jitter": 0}, + }, + ) + assert response.status_code == 200, response.text + + with c.override( + Materialized( + external_metadata_store=True, + options=[ + f"--timestamp-oracle-url=postgres://root@toxiproxy:{oracle_port}" + "?options=--search_path=tsoracle", + ], + ) + ): + c.up("toxiproxy") + toxi = f"http://localhost:{c.default_port('toxiproxy')}" + requests.delete(f"{toxi}/proxies/oracle") + response = requests.post( + f"{toxi}/proxies", + json={ + "name": "oracle", + "listen": f"0.0.0.0:{oracle_port}", + "upstream": "postgres-metadata:26257", + "enabled": True, + }, + ) + assert response.status_code == 201, response.text + c.up("materialized") + + c.sql( + "ALTER SYSTEM SET cluster_controller_tick_interval = '100ms'", + port=6877, + user="mz_system", + ) + mz = c.sql_cursor() + mz.execute("SET transaction_isolation = 'serializable'") + + def converge_ms(replication_factor: int) -> float: + start = time.monotonic() + mz.execute( + sql.SQL("ALTER CLUSTER cc_probe SET (REPLICATION FACTOR {})").format( + sql.Literal(replication_factor) + ) + ) + while True: + mz.execute( + "SELECT count(*) FROM mz_cluster_replicas r JOIN mz_clusters c " + "ON r.cluster_id = c.id WHERE c.name = 'cc_probe'" + ) + if mz.fetchall()[0][0] == replication_factor: + return (time.monotonic() - start) * 1000 + assert ( + time.monotonic() - start < 120 + ), f"cc_probe never reached rf {replication_factor}" + time.sleep(0.05) + + def convergence_samples() -> list[float]: + samples = [] + replication_factor = 1 + for _ in range(3): + samples.append(converge_ms(replication_factor)) + replication_factor = 1 - replication_factor + return samples + + def await_scheduled_replicas() -> None: + start = time.monotonic() + while True: + mz.execute( + "SELECT count(DISTINCT c.id), count(*) " + "FROM mz_clusters c JOIN mz_cluster_replicas r " + "ON r.cluster_id = c.id " + "WHERE c.name LIKE 'cc\\_sched%' ESCAPE '\\'" + ) + cluster_count, replica_count = mz.fetchall()[0] + if ( + cluster_count == args.scheduled_clusters + and replica_count == args.scheduled_clusters + ): + return + if time.monotonic() - start >= 120: + mz.execute( + "SELECT c.name, count(r.id) " + "FROM mz_clusters c LEFT JOIN mz_cluster_replicas r " + "ON r.cluster_id = c.id " + "WHERE c.name LIKE 'cc\\_sched%' ESCAPE '\\' " + "GROUP BY c.name ORDER BY c.name" + ) + replica_counts = mz.fetchall() + raise AssertionError( + "scheduled cluster replica counts after 120s: " + f"{replica_counts}. Expected exactly " + f"{args.scheduled_clusters} cc_sched clusters with " + "exactly one replica each" + ) + time.sleep(0.05) + + set_latency(toxi, 0) + mz.execute( + "CREATE CLUSTER cc_probe " + "(SIZE 'scale=1,workers=1', REPLICATION FACTOR 0)" + ) + + no_latency_samples = convergence_samples() + no_latency_ms = median(no_latency_samples) + converge_ms(0) + + set_latency(toxi, args.latency_ms) + control_samples = convergence_samples() + control_ms = median(control_samples) + set_latency(toxi, 0) + latency_increase_ms = control_ms - no_latency_ms + minimum_increase_ms = args.latency_ms / 2 + assert latency_increase_ms >= minimum_increase_ms, ( + f"injecting {args.latency_ms}ms oracle latency increased the control " + f"measurement by only {latency_increase_ms:.0f}ms, expected at least " + f"{minimum_increase_ms:.0f}ms. The oracle may be bypassing toxiproxy" + ) + + converge_ms(0) + mz.execute("CREATE TABLE cc_sched_t (x int)") + for i in range(args.scheduled_clusters): + cluster_name = sql.Identifier(f"cc_sched{i}") + mz.execute( + sql.SQL( + "CREATE CLUSTER {} (SIZE 'scale=1,workers=1', " + "SCHEDULE = ON REFRESH " + "(HYDRATION TIME ESTIMATE = '60 seconds'))" + ).format(cluster_name) + ) + mz.execute( + sql.SQL( + "CREATE MATERIALIZED VIEW {} IN CLUSTER {} " + "WITH (REFRESH = EVERY '1 second') AS " + "SELECT count(*) FROM cc_sched_t" + ).format(sql.Identifier(f"cc_sched{i}_mv"), cluster_name) + ) + + await_scheduled_replicas() + set_latency(toxi, args.latency_ms) + stalled_samples = convergence_samples() + stalled_ms = median(stalled_samples) + set_latency(toxi, 0) + + excess_ms = stalled_ms - control_ms + ceiling_ms = 4 * args.latency_ms + print(f"cc_probe 0 -> 1 replica at {args.latency_ms}ms oracle latency:") + print(f" no injected latency : {no_latency_ms:.0f}ms") + print(f" 0 scheduled clusters : {control_ms:.0f}ms") + print(f" {args.scheduled_clusters} scheduled clusters : {stalled_ms:.0f}ms") + print( + f" excess : {excess_ms:.0f}ms (ceiling {ceiling_ms}ms, " + "expected bounded oracle round-trips per controller phase)" + ) + assert excess_ms < ceiling_ms, ( + f"{args.scheduled_clusters} unrelated ON REFRESH clusters delayed the " + f"probe cluster's reconciliation by {excess_ms:.0f}ms " + f"(ceiling {ceiling_ms}ms)" + )