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 compares probe-cluster convergence
with zero and eight scheduled clusters under injected oracle latency.

Fixes SQL-569.

Author: aljoscha

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.
 //!
 //! Everything here is gated by [`ENABLE_CLUSTER_CONTROLLER`] (default on). With
 //! the gate off the task does not tick, so the legacy scheduling and graceful
@@ -26,7 +27,7 @@
 //! replicas are materialized by `reconcile_builtin_cluster_replicas` at catalog
 //! open, which derives the same target from the same config.)
 
-use std::collections::BTreeSet;
+use std::collections::{BTreeMap, BTreeSet};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -36,7 +37,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;
@@ -54,8 +56,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 *user* managed clusters the controller owns this tick.
@@ -81,13 +84,14 @@ pub enum ClusterControllerRequest {
         cluster_id: ClusterId,
         tx: oneshot::Sender<bool>,
     },
-    /// 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<Option<RefreshWindowInputs>>,
+        tx: oneshot::Sender<Option<RefreshWindowClusterInputs>>,
     },
+    /// One timestamp-oracle read for a completed refresh-window input batch.
+    RefreshWindowReadTs { tx: oneshot::Sender<Timestamp> },
     /// Apply a tick's batch of decisions under their compare-and-append guards.
     Apply {
         decisions: Vec<Decision>,
@@ -184,11 +188,38 @@ impl ClusterControllerCtx for CoordCtx {
 
     async fn refresh_window_inputs(
         &mut self,
-        cluster_id: ClusterId,
-    ) -> Option<RefreshWindowInputs> {
-        self.request(|tx| ClusterControllerRequest::RefreshWindowInputs { cluster_id, tx })
-            .await
-            .flatten()
+        cluster_ids: &[ClusterId],
+    ) -> Option<RefreshWindowInputsBatch> {
+        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<Decision>) -> ApplyOutcome {
@@ -326,36 +357,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
-                // coordinator loop. The legacy `check_refresh_policy` makes
-                // the same split.
-                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,
-                            }));
-                        });
-                    }
-                }
+            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.
+                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 {
@@ -514,7 +527,7 @@ impl Coordinator {
     /// REFRESH`. These are the same signals the legacy `check_refresh_policy`
     /// reads.
     ///
-    /// 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.
@@ -525,7 +538,7 @@ impl Coordinator {
     fn refresh_window_catalog_inputs(
         &self,
         cluster_id: ClusterId,
-    ) -> Option<(Duration, Vec<RefreshMvInfo>)> {
+    ) -> Option<RefreshWindowClusterInputs> {
         use mz_catalog::memory::objects::CatalogItem;
 
         let cluster = self.catalog().try_get_cluster(cluster_id)?;
@@ -568,7 +581,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.

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;
@@ -176,16 +176,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<RefreshMvInfo>,
+}
+
+/// 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<ClusterId, RefreshWindowClusterInputs>,
+}
+
+/// 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.
@@ -417,21 +437,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<RefreshWindowInputs>;
+    /// 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<RefreshWindowInputsBatch>;
 
     /// Apply a tick's batch of decisions under their compare-and-append guards.
     /// Each decision carries the [`ExpectedClusterState`] it was derived from;

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,17 +330,20 @@ 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,
         states: &[ClusterState],
         config: &ConfigSignals,
     ) -> BTreeMap<ClusterId, LiveSignals> {
         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
     }