cluster-controller: batch ON REFRESH oracle reads

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
 

src/adapter/src/coord/cluster_controller.rs

@@ -14,17 +14,18 @@
 //! 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
 //! materialized by `reconcile_builtin_cluster_replicas` at catalog open, which
 //! 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<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>,
@@ -180,11 +184,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 {
@@ -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<RefreshMvInfo>)> {
+    ) -> Option<RefreshWindowClusterInputs> {
         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.

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<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.
@@ -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<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
     }