compute: a shared-trace primitive for cross-thread arrangement reads

[antiguru]

First of eight PRs splitting #37770. Stacks on #38396. Tracked by CPU-215.

An arrangement is normally readable only from the timely worker that maintains it, because its batches are Rc-backed and its trace handle is neither Send nor Sync. This adds a publication point carrying Arc-backed batches together with the trace's since and upper, so a reader on any thread can mint a Send handle for the same arrangement and import it as a snapshot at a chosen as_of.

A publication point is differential's TraceBox for readers that are not agents of the trace. It accumulates their holds in a MutableAntichain per axis, and each handle adjusts that accumulation as a delta the way a TraceAgent does. The standing hold and the publisher's own hold at the chain coverage are ordinary holds in those accumulations, so a shared arrangement compacting no faster than the slowest runtime's command stream follows from a registered hold rather than from an invariant asserted after the fact. The controller's own frontier stays out, since it is another agent's hold on the same trace and so belongs to the meet the trace already computes, which is what the publisher publishes as since.

Inert: nothing in the crate calls it. The module is pub so that its accessors are reachable for dead-code analysis while the only callers are its own tests.

PublishArrangement requires Tr::Batch: Send + Sync and the tests instantiate it over RowRowSpine, so Arc-backed spines are a compile prerequisite rather than a preference. This PR retargets to main once #38396 merges.

[def-]

QA LLM Review

1. MEDIUM -- a live import permanently pins the published arrangement's physical compaction

src/compute/src/shared_trace.rs:1075

The read hold import_snapshot_at retains for the life of the import advances only on the logical axis, and Clone seeds its physical hold from the mint-time since rather than the chain coverage. The accumulated physical frontier therefore sticks at that stale since forever, the publisher never forwards anything higher, and the published spine stops merging: batches pile up in Spine::pending, one per seal, for the life of the import.

Details

Measured on a RowRowSpine published through adopt, with a live import_snapshot_at (until empty) consumed by as_collection, over 40 seal ticks against an identical unimported control:

after register: physical_holds={0: [4]}          coverage_hold=[4]  accumulated=[4]  since=[0]
after import:   physical_holds={2: [0]}          coverage_hold=[4]  accumulated=[0]  since=[0]
end:            physical_holds={2: [0]}          coverage_hold=[40] accumulated=[0]  since=[0]
chain_len: imported=39  control=5

Registration id 2 is the hold clone at shared_trace.rs:1075. register/register_at do install the hold at the chain coverage (id 0 above sits at [4], as shared_trace.rs:151 documents), but Clone at shared_trace.rs:625 writes self.physical into the new registration, and self.physical is initialised to since at shared_trace.rs:482 and shared_trace.rs:537. Nothing then moves it: shared_trace.rs:1167 follows acknowledged on the logical axis only, and the TraceFrontier clone that a join would advance is dropped at build time for every consumer that keeps only the stream. So physical_compaction's meet is pinned at the since observed when the handle was minted, agent.set_physical_compaction joins and cannot be pulled back down, and consider_merges never drains pending again because pending[0].upper() <= physical_frontier stays false.

Cost is unbounded rather than constant: retractions in stranded pending batches never consolidate, and every cursor_through builds a CursorList over a batch count that grows one per seal. On an index with a live shared import and a one-second seal cadence that is thousands of batches per hour.

Suggested fix, verified against the same probe (chain folds to 5, exactly matching the control, and the hold tracks to [40]):

                                 if let Some(hold) = hold.as_mut() {
                                     hold.set_logical_compaction(acknowledged.borrow());
+                                    hold.set_physical_compaction(acknowledged.borrow());
                                 }

acknowledged is the right value on both axes: it is exactly the frontier below which this import will never cut again, which is what the physical hold is supposed to express. Worth separately reconciling Clone at shared_trace.rs:625 with the coverage-seeded hold register/register_at install, since as written a clone silently lowers a registration's physical hold to since and the two paths disagree about the documented invariant at shared_trace.rs:151.

[antiguru]

Confirmed and fixed in 285af14. The measurement reproduces exactly: 39 batches against an unimported control's 5, over 40 seals.

Both of the report's points were real, and the second one is the root cause rather than a separate cleanup. A handle carries two physical frontiers that 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 (mz_join_core asserts exactly that). 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 set_physical_compaction both wrote the reported frontier into the hold, so the hold silently dropped to the weaker value, and since the accumulation is a meet, one such registration is a floor under every other hold.

The 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 raised it afterwards.

The fix keeps the two frontiers in separate fields, has Clone inherit the hold, has the setter join into both, and takes the suggested set_physical_compaction(acknowledged) on the import's hold. acknowledged is right on the physical axis for the reason given: it is exactly the frontier below which that import will never cut again.

Two regression tests, each verified red without its half of the fix:

• live_import_does_not_pin_merging — the two-arm chain-length comparison, 39 against 5 before, folding to the control after. One note on reproducing it: the minting handle has to be dropped after building the import, as render::import_shared_index does. A live mint holds its own coverage-seeded registration and pins the floor by itself, which masks the bug under test. My first attempt at this test failed for that reason rather than the one it was written for.
• clone_inherits_the_hold_not_the_reported_frontier — asserts the registered hold directly, since the reported frontier cannot be read back through TraceReader. This is what pins the Clone half; the merge test alone does not, because the added physical advance repairs that scenario regardless of what Clone seeded.

The coverage_hold this PR already gives the publisher caps the forwarded physical frontier at the chain coverage, so advancing a reader hold to acknowledged cannot push the forwarded value past what the chain carries even though acknowledged tracks the stream frontier, which leads the coverage by up to a scheduling round.

The stack above this PR is rebased and pushed. Full run at the tip: 124 tests, workspace cargo check --all-targets, clippy, rustdoc with -D warnings.

(Posted by Claude Code.)