compute: report how far readers hold a shared arrangement back

doc/user/data/metrics.yml

@@ -844,6 +844,18 @@ metrics:
   - response_type
   source: src/compute-client/src/metrics.rs
   visibility: internal
+- name: mz_compute_shared_arrangement_held_count
+  help: How many of the shared arrangements this worker publishes are kept from compacting by the importing runtime or a reader.
+  labels:
+  - worker_id
+  source: src/compute/src/metrics.rs
+  visibility: internal
+- name: mz_compute_shared_arrangement_hold_gap_ms
+  help: The largest gap, over the shared arrangements this worker publishes, between the compaction frontier it was told to apply and the one it applied.
+  labels:
+  - worker_id
+  source: src/compute/src/metrics.rs
+  visibility: internal
 - name: mz_connection_status
   help: Count of completed network connections, by status
   labels:

src/compute/src/compute_state.rs

@@ -1380,6 +1380,25 @@ impl<'a> ActiveComputeState<'a> {
                 .replica_expiration_remaining_seconds
                 .set(remaining)
         }
+
+        // Only the publishing runtime is held back, and only it has slots in the registry under its
+        // own worker ordinal. Reporting from the interactive runtime as well would repeat the same
+        // numbers under a second `role` label, so its series stays at zero: an extra zero leaves a
+        // `sum` or a `max` over the label correct, where a duplicate would not.
+        if self.compute_state.role() != ComputeRuntimeRole::Interactive {
+            let (gap, held) = self
+                .compute_state
+                .sharing_registry
+                .hold_gaps(self.timely_worker.index());
+            self.compute_state
+                .metrics
+                .shared_arrangement_hold_gap_ms
+                .set(gap);
+            self.compute_state
+                .metrics
+                .shared_arrangement_held_count
+                .set(u64::cast_from(held));
+        }
     }
 
     /// Gives `peek` a turn on the worker if this activation's budget has one left, and queues it

src/compute/src/metrics.rs

@@ -89,6 +89,10 @@ pub struct ComputeMetrics {
 
     // subscribes
     subscribe_snapshots_skipped_total: IntCounter,
+
+    // shared arrangements
+    shared_arrangement_hold_gap_ms: raw::UIntGaugeVec,
+    shared_arrangement_held_count: raw::UIntGaugeVec,
 }
 
 /// Applies the per-role const label to `opts`, unless `role` is `Solo`.
@@ -305,6 +309,16 @@ impl ComputeMetrics {
                 name: "mz_subscribe_snapshots_skipped_total",
                 help: "The number of collection snapshots that were skipped by the subscribe snapshot optimization.",
             ), role)),
+            shared_arrangement_hold_gap_ms: registry.register(with_role(metric!(
+                name: "mz_compute_shared_arrangement_hold_gap_ms",
+                help: "The largest gap, over the shared arrangements this worker publishes, between the compaction frontier it was told to apply and the one it applied.",
+                var_labels: ["worker_id"],
+            ), role)),
+            shared_arrangement_held_count: registry.register(with_role(metric!(
+                name: "mz_compute_shared_arrangement_held_count",
+                help: "How many of the shared arrangements this worker publishes are kept from compacting by the importing runtime or a reader.",
+                var_labels: ["worker_id"],
+            ), role)),
         }
     }
 
@@ -357,6 +371,12 @@ impl ComputeMetrics {
         let shared_row_heap_capacity_bytes = self
             .shared_row_heap_capacity_bytes
             .with_label_values(&[&worker]);
+        let shared_arrangement_hold_gap_ms = self
+            .shared_arrangement_hold_gap_ms
+            .with_label_values(&[&worker]);
+        let shared_arrangement_held_count = self
+            .shared_arrangement_held_count
+            .with_label_values(&[&worker]);
 
         WorkerMetrics {
             worker_label: worker,
@@ -385,6 +405,8 @@ impl ComputeMetrics {
             replica_expiration_timestamp_seconds,
             replica_expiration_remaining_seconds,
             shared_row_heap_capacity_bytes,
+            shared_arrangement_hold_gap_ms,
+            shared_arrangement_held_count,
         }
     }
 }
@@ -454,6 +476,10 @@ pub struct WorkerMetrics {
     pub(crate) replica_expiration_remaining_seconds: raw::Gauge,
     /// Heap capacity of the shared row.
     shared_row_heap_capacity_bytes: UIntGauge,
+    /// The largest compaction gap over the shared arrangements this worker publishes.
+    pub(crate) shared_arrangement_hold_gap_ms: UIntGauge,
+    /// How many of those arrangements are held back at all.
+    pub(crate) shared_arrangement_held_count: UIntGauge,
 }
 
 impl WorkerMetrics {

src/compute/src/shared_trace/publish.rs

@@ -121,6 +121,15 @@ where
         self.shared.upper()
     }
 
+    /// The `(applied, requested)` logical compaction frontiers of this point.
+    ///
+    /// `applied` is what the publishing runtime's trace compacted to, `requested` what its
+    /// controller stream asked for. They differ by exactly what the importing runtime's standing
+    /// hold, or a live reader, is keeping the publisher from shedding.
+    pub(crate) fn logical_frontiers(&self) -> (Antichain<Tr::Time>, Antichain<Tr::Time>) {
+        (self.shared.since(), self.shared.writer_since())
+    }
+
     /// Why this point would refuse an `as_of`. See [`Diagnostics`].
     pub(crate) fn diagnostics(&self) -> Diagnostics<Tr::Time> {
         Diagnostics {

src/compute/src/shared_trace/tests.rs

@@ -637,6 +637,69 @@ fn handle_at_mints_at_as_of_or_refuses() {
     });
 }
 
+/// A point reports both the frontier its writer was told to compact to and the one the holds let
+/// it apply.
+///
+/// The two differ by exactly what the importing runtime is withholding, which is the coupling this
+/// publication introduces, and the pair is the only place that coupling is observable.
+#[mz_ore::test]
+fn logical_frontiers_report_what_the_holds_withhold() {
+    timely::execute_directly(move |worker| {
+        let (mut writer, published, mut input) = worker.dataflow::<Timestamp, _, _>(|scope| {
+            let (input, collection) = scope.new_collection::<(Row, Row), Diff>();
+            let arranged = collection.mz_arrange::<
+                ColumnationChunker<_>,
+                RowRowBatcher<_, _>,
+                RowRowBuilder<_, _>,
+                RowRowSpine<_, _>,
+            >("gap oks");
+            let writer = arranged.trace.clone();
+            let published = adopt_fresh(&arranged);
+            (writer, published, input)
+        });
+        for t in 0..5 {
+            tick(
+                worker,
+                &mut input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+        }
+
+        // The importing runtime is four behind the controller's request, so the writer applies
+        // four and reports ten.
+        let held = Antichain::from_elem(Timestamp::from(4_u64));
+        let target = Antichain::from_elem(Timestamp::from(10_u64));
+        published.note_standing_hold(&held);
+        writer.set_logical_compaction(target.borrow());
+        tick(
+            worker,
+            &mut input,
+            Timestamp::from(10_u64),
+            Timestamp::from(11_u64),
+        );
+        assert_eq!(
+            published.logical_frontiers(),
+            (held, target.clone()),
+            "the applied frontier must report the hold and the requested one the controller"
+        );
+
+        // The importing runtime catches up and the gap closes.
+        published.note_standing_hold(&target);
+        tick(
+            worker,
+            &mut input,
+            Timestamp::from(11_u64),
+            Timestamp::from(12_u64),
+        );
+        assert_eq!(
+            published.logical_frontiers(),
+            (target.clone(), target),
+            "a caught-up importer must leave no gap"
+        );
+    });
+}
+
 /// A consumer forwarding an empty input frontier releases its hold rather than recording an
 /// empty one.
 ///

src/compute/src/sharing.rs

@@ -327,6 +327,42 @@ impl ArrangementSharingRegistry {
         Some((slot.oks.handle(), slot.errs.handle()))
     }
 
+    /// How far the readers' holds keep the publishing runtime from compacting, over every
+    /// arrangement published on `worker_index`.
+    ///
+    /// Returns the largest gap in milliseconds between an arrangement's requested and applied
+    /// logical compaction frontiers, and how many arrangements have any gap at all. Aliases share
+    /// their target's points, so points are counted once rather than once per id. A point whose
+    /// requested frontier is empty is counted as held but contributes no gap: its collection is
+    /// being dropped and has no finite frontier left to subtract from.
+    pub(crate) fn hold_gaps(&self, worker_index: usize) -> (u64, usize) {
+        let inner = self.lock();
+        let mut seen = BTreeSet::new();
+        let mut max_gap = 0;
+        let mut held = 0;
+        for slots in inner.map.values() {
+            let Some(Some(slot)) = slots.get(worker_index) else {
+                continue;
+            };
+            if !seen.insert(Arc::as_ptr(slot)) {
+                continue;
+            }
+            let (applied, requested) = slot.oks.logical_frontiers();
+            match (applied.as_option(), requested.as_option()) {
+                (Some(applied), Some(requested)) => {
+                    let gap = u64::from(*requested).saturating_sub(u64::from(*applied));
+                    if gap > 0 {
+                        held += 1;
+                        max_gap = max_gap.max(gap);
+                    }
+                }
+                (Some(_), None) => held += 1,
+                _ => {}
+            }
+        }
+        (max_gap, held)
+    }
+
     /// The accumulated `oks` logical holds registered against `id` on `worker_index`, if published.
     ///
     /// Test-only. Minting a handle to observe the published frontiers cannot distinguish a live

src/timely-util/src/shared_trace.rs

@@ -87,6 +87,12 @@ struct State<B: BatchReader> {
     upper: Antichain<B::Time>,
     /// The inner trace's logical compaction frontier. Reads at times not beyond it are not accurate.
     logical: Antichain<B::Time>,
+    /// The frontier the writer asked for, before meeting the readers' holds.
+    ///
+    /// Equal to `logical` when nothing holds the trace back. Above it by exactly the amount a
+    /// reader is keeping the writer from compacting, which is the coupling this publication
+    /// introduces and the only place it is observable.
+    writer_logical: Antichain<B::Time>,
     /// The inner trace's physical compaction frontier.
     physical: Antichain<B::Time>,
     /// Readers' logical holds.
@@ -147,6 +153,7 @@ impl<B: BatchReader> Shared<B> {
                 chain: Vec::new(),
                 upper: minimum.clone(),
                 logical: minimum.clone(),
+                writer_logical: minimum.clone(),
                 physical: minimum,
                 remote_logical: MutableAntichain::new(),
                 remote_physical: MutableAntichain::new(),
@@ -171,6 +178,12 @@ impl<B: BatchReader> Shared<B> {
         self.lock().logical.clone()
     }
 
+    /// The logical compaction frontier the writer asked for, before the readers' holds were met
+    /// into it. At or beyond [`Shared::since`], and equal to it when no reader holds this point.
+    pub fn writer_since(&self) -> Antichain<B::Time> {
+        self.lock().writer_logical.clone()
+    }
+
     /// The published `(since, upper)`, read together.
     pub fn frontiers(&self) -> (Antichain<B::Time>, Antichain<B::Time>) {
         let state = self.lock();
@@ -300,6 +313,7 @@ impl<Tr: Trace> SharedSpine<Tr> {
             // The frontiers the `TraceReader` impl below last saw. Reading them from `inner` here
             // would need `&mut`, and they are equal.
             state.logical = self.local_logical.clone();
+            state.writer_logical = self.local_logical.clone();
             state.physical = self.local_physical.clone();
             state.live_queues()
         };
@@ -437,6 +451,7 @@ impl<Tr: Trace> SharedSpine<Tr> {
         let logical = self.local_logical.meet(&remote_logical);
         for state in guards.iter_mut() {
             state.logical = logical.clone();
+            state.writer_logical = self.local_logical.clone();
         }
         drop(guards);
         drop(attachment);