fixup: a handle's physical hold is not the frontier it reports

A handle carries two physical frontiers, and they are not interchangeable. The
one it reports through `get_physical_compaction` is seeded at the published
`since`, because a reported frontier may never lead the chain coverage. The one
it holds is seeded at that coverage, because a merge spanning the coverage
destroys the boundary the reader was seeded with.

`Clone` and the setter both wrote the reported frontier into the hold, which
silently lowers it. Since the accumulation is a meet, one such registration is a
floor under every other hold, so the published spine stops merging: batches pile
up in `Spine::pending`, one per seal, for as long as that registration lives.
The cost is unbounded rather than constant, since retractions in stranded
batches never consolidate and every `cursor_through` builds a `CursorList` over
all of them.

An import's read hold hit this on both counts. It is a clone, so it registered
at `since`, and it advanced only on the logical axis, so nothing ever raised it.
Measured against an unimported control over 40 seals: 39 batches against 5.

Keep the two frontiers in separate fields, have `Clone` inherit the hold, have
the setter join into both, and advance the import's hold on both axes.
`acknowledged` is the right value for the physical axis too: it is exactly the
frontier below which that import will never cut again.

Reported by the QA LLM review on #38386.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: antiguru

src/compute/src/shared_trace.rs

@@ -373,6 +373,16 @@ where
         state.logical_holds.values().cloned().collect()
     }
 
+    /// The physical holds currently registered against this publication point.
+    ///
+    /// Test-only. A handle reports a weaker physical frontier than it holds, so the hold cannot be
+    /// read back through `TraceReader::get_physical_compaction`.
+    #[cfg(test)]
+    pub(crate) fn physical_holds(&self) -> Vec<Antichain<Tr::Time>> {
+        let state = self.shared.state.lock().expect("shared trace poisoned");
+        state.physical_holds.values().cloned().collect()
+    }
+
     /// The number of batches in the published chain.
     ///
     /// Test-only. Counting through a handle would register a physical hold and perturb the merge
@@ -453,9 +463,21 @@ pub struct SharedTraceHandle<Tr: TraceReader> {
     /// This handle's own logical frontier, mirrored into `logical_holds[id]`. Kept locally so
     /// `get_logical_compaction` can return a borrow.
     logical: Antichain<Tr::Time>,
-    /// This handle's own physical frontier, mirrored into `physical_holds[id]`. Kept locally so
+    /// The physical frontier this handle *reports*, seeded at the published `since`. Kept locally so
     /// `get_physical_compaction` can return a borrow.
     physical: Antichain<Tr::Time>,
+    /// The physical frontier this handle *holds*, seeded at the chain coverage and mirrored into
+    /// `physical_holds[id]`.
+    ///
+    /// Distinct from `physical`, which is the weaker of the two on registration and must stay so:
+    /// a reported frontier may never lead the chain coverage (see `Self::register_at`), while the
+    /// hold has to start AT that coverage or a merge can eat the boundary this reader was seeded
+    /// with. Overwriting the hold with the reported value silently lowers it, which stops the spine
+    /// merging for the life of the handle.
+    ///
+    /// The logical axis needs no such split: both registrations install the same frontier a handle
+    /// reports.
+    physical_hold: Antichain<Tr::Time>,
 }
 
 impl<Tr: TraceReader> SharedTraceHandle<Tr>
@@ -464,7 +486,7 @@ where
 {
     /// Registers a fresh hold at the current published `since` and returns a handle for it.
     fn register(shared: SharedTraceRef<Tr>) -> Self {
-        let (id, since) = {
+        let (id, since, coverage) = {
             let mut state = shared.state.lock().expect("shared trace poisoned");
             let id = state.next_id;
             state.next_id += 1;
@@ -473,13 +495,14 @@ where
             // The cut floor starts at the chain coverage, NOT at `since`. See `register_at`.
             let coverage = seed_frontier::<Tr>(&state.chain, &state.upper);
             state.set_physical_hold(id, &coverage);
-            (id, since)
+            (id, since, coverage)
         };
         Self {
             shared,
             id,
             logical: since.clone(),
             physical: since,
+            physical_hold: coverage,
         }
     }
 
@@ -489,7 +512,7 @@ where
         shared: SharedTraceRef<Tr>,
         as_of: &Antichain<Tr::Time>,
     ) -> Result<Self, Antichain<Tr::Time>> {
-        let (id, since) = {
+        let (id, since, coverage) = {
             let mut state = shared.state.lock().expect("shared trace poisoned");
             if !timely::PartialOrder::less_equal(&state.since, as_of) {
                 return Err(state.since.clone());
@@ -516,7 +539,7 @@ where
             // will get, and permit a merge across the very first frontier that reader cuts at.
             let coverage = seed_frontier::<Tr>(&state.chain, &state.upper);
             state.set_physical_hold(id, &coverage);
-            (id, since)
+            (id, since, coverage)
         };
         Ok(Self {
             shared,
@@ -535,6 +558,7 @@ where
             // publisher forwards the published `since` as its physical target, so this reports the
             // grant.
             physical: since,
+            physical_hold: coverage,
         })
     }
 
@@ -601,10 +625,10 @@ where
         state.set_logical_hold(self.id, &self.logical);
     }
 
-    /// Mirrors this handle's physical frontier into the publication point.
+    /// Mirrors this handle's physical hold into the publication point.
     fn update_physical_hold(&self) {
         let mut state = self.shared.state.lock().expect("shared trace poisoned");
-        state.set_physical_hold(self.id, &self.physical);
+        state.set_physical_hold(self.id, &self.physical_hold);
     }
 }
 
@@ -622,14 +646,19 @@ where
             let id = state.next_id;
             state.next_id += 1;
             state.set_logical_hold(id, &self.logical);
-            state.set_physical_hold(id, &self.physical);
+            // The clone inherits this handle's HOLD, not the frontier it reports. Registering the
+            // reported frontier would install a hold below the coverage the source was seeded with,
+            // and since the accumulation is a meet, one such clone stops the spine merging for as
+            // long as it lives.
+            state.set_physical_hold(id, &self.physical_hold);
             id
         };
         Self {
             shared: Arc::clone(&self.shared),
             id,
             logical: self.logical.clone(),
             physical: self.physical.clone(),
+            physical_hold: self.physical_hold.clone(),
         }
     }
 }
@@ -705,7 +734,12 @@ where
     }
 
     fn set_physical_compaction(&mut self, frontier: AntichainRef<'_, Tr::Time>) {
-        self.physical = self.physical.join(&frontier.to_owned());
+        let frontier = frontier.to_owned();
+        self.physical = self.physical.join(&frontier);
+        // Join, never assign: the hold starts at the chain coverage, which leads what this handle
+        // reports, and a request below that coverage must not pull the hold down to it. `join` with
+        // the empty antichain is absorbing, which is what still lets an empty request release.
+        self.physical_hold = self.physical_hold.join(&frontier);
         self.update_physical_hold();
     }
 
@@ -1165,6 +1199,12 @@ where
                                 // slower of the two wins.
                                 if let Some(hold) = hold.as_mut() {
                                     hold.set_logical_compaction(acknowledged.borrow());
+                                    // Both axes. `acknowledged` is exactly the frontier below which
+                                    // this import will never cut again, which is what the physical
+                                    // hold expresses. Without this the hold pins the spine at the
+                                    // coverage it registered at and the chain grows one batch per
+                                    // seal for the life of the import.
+                                    hold.set_physical_compaction(acknowledged.borrow());
                                 }
                                 // Bound the read at `until`: once the trace's frontier reaches it, drop
                                 // the capability so a single-time read completes. Otherwise track the

src/compute/src/shared_trace/tests.rs

@@ -877,6 +877,159 @@ fn reader_floor_bounds_merges_and_no_reader_merges_freely() {
     });
 }
 
+/// A clone inherits its source's physical hold, not the weaker frontier its source reports.
+///
+/// `register`/`register_at` seed the hold at the chain coverage while reporting the published
+/// `since`, which is weaker. Registering the reported frontier for a clone would silently install a
+/// hold below the coverage the source was seeded with, and since the accumulation is a meet, that
+/// clone becomes a floor under every other hold for as long as it lives.
+#[mz_ore::test]
+fn clone_inherits_the_hold_not_the_reported_frontier() {
+    timely::execute_directly(move |worker| {
+        let (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<_, _>,
+            >("clone oks");
+            (adopt_fresh(&arranged), input)
+        });
+        for t in 0..4u64 {
+            tick(
+                worker,
+                &mut input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+        }
+
+        // The chain now covers 4, while the published `since` is still the minimum, so the two
+        // frontiers a handle carries are distinguishable.
+        let handle = published.handle();
+        let (since, _upper) = handle.frontiers();
+        let coverage = Antichain::from_elem(Timestamp::from(4_u64));
+        assert_eq!(
+            since,
+            Antichain::from_elem(Timestamp::MIN),
+            "no compaction was requested, so `since` should still be the minimum"
+        );
+
+        let clone = handle.clone();
+        drop(handle);
+        let holds = published.physical_holds();
+        assert_eq!(
+            holds,
+            vec![coverage],
+            "the clone's hold is not the coverage its source was seeded with, so it sits below \
+             every boundary the source still needed"
+        );
+        drop(clone);
+    });
+}
+
+/// A live import does not pin the published spine's physical compaction.
+///
+/// The import's own read hold is the only registration a consumer that keeps the stream leaves
+/// behind, so if that hold never rises the accumulated physical frontier never rises either, and
+/// the spine stops merging for the life of the import. The cost is unbounded rather than constant:
+/// one stranded batch per seal, whose retractions never consolidate, and a `CursorList` over all of
+/// them on every `cursor_through`.
+///
+/// Same two-arm shape as [`reader_floor_bounds_merges_and_no_reader_merges_freely`], and for the
+/// same reason: the observable is the difference between the arms' chain lengths, which is immune to
+/// the spine's particular merge policy.
+#[mz_ore::test]
+fn live_import_does_not_pin_merging() {
+    timely::execute_directly(move |worker| {
+        let (imported, control, mut imported_input, mut control_input) = worker
+            .dataflow::<Timestamp, _, _>(|scope| {
+                let (imported_input, imported_collection) =
+                    scope.new_collection::<(Row, Row), Diff>();
+                let imported_arranged = imported_collection.mz_arrange::<
+                    ColumnationChunker<_>,
+                    RowRowBatcher<_, _>,
+                    RowRowBuilder<_, _>,
+                    RowRowSpine<_, _>,
+                >("imported oks");
+                let (control_input, control_collection) =
+                    scope.new_collection::<(Row, Row), Diff>();
+                let control_arranged = control_collection.mz_arrange::<
+                    ColumnationChunker<_>,
+                    RowRowBatcher<_, _>,
+                    RowRowBuilder<_, _>,
+                    RowRowSpine<_, _>,
+                >("control oks");
+                (
+                    adopt_fresh(&imported_arranged),
+                    adopt_fresh(&control_arranged),
+                    imported_input,
+                    control_input,
+                )
+            });
+
+        // Seal a few times so the chain the import seeds from is non-empty.
+        for t in 0..4u64 {
+            tick(
+                worker,
+                &mut imported_input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+            tick(
+                worker,
+                &mut control_input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+        }
+
+        // A live import: empty `until`, so it never completes and its read hold lives as long as
+        // the dataflow. Keep only the stream and drop the trace, which is what `as_collection` and
+        // the reduce path do during construction, leaving the import's own hold as the only
+        // registration.
+        let handle = imported.handle();
+        worker.dataflow::<Timestamp, _, _>(|scope| {
+            let arranged = handle.import_snapshot_at(
+                scope.clone(),
+                "live import",
+                Antichain::from_elem(Timestamp::from(4_u64)),
+                Antichain::new(),
+            );
+            drop(arranged.trace);
+        });
+        // Drop the minting handle, as `crate::render::import_shared_index` does: the import owns
+        // its own clone, and a live mint would pin the floor at its own registration coverage and
+        // mask what this test is about.
+        drop(handle);
+
+        for t in 4..40u64 {
+            tick(
+                worker,
+                &mut imported_input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+            tick(
+                worker,
+                &mut control_input,
+                Timestamp::from(t),
+                Timestamp::from(t + 1),
+            );
+        }
+
+        let imported_len = imported.chain_len();
+        let control_len = control.chain_len();
+        assert!(
+            imported_len <= control_len + 2,
+            "imported chain {imported_len} against unimported {control_len}: the live import's \
+             read hold is not following the stream on the physical axis, so the spine cannot merge \
+             for as long as the import lives"
+        );
+    });
+}
+
 /// A fresh importer is seeded with the frontier its seeded chain covers, not the stream frontier
 /// that lags it.
 ///