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)" + )