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

[antiguru]

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

An arrangement is normally readable only from the timely worker that maintains it. Its batches already cross threads, being Arc-backed through mz_row_spine::ArcBatch, but its trace handle is a TraceAgent, which is Rc<RefCell<..>> and pinned to one thread, and its spine does merge work whenever it is touched, including inside roll_up and complete_at, which finish merges synchronously. A spine behind a mutex would hold that mutex for whole merges.

SharedSpine in mz-timely-util is a Trace wrapper that sidesteps this. It owns the inner spine on the arranging worker's thread and, once attached to a publication point, mirrors the spine's batch chain, upper, and compaction frontiers into the point inside every mutation. Batches are immutable and reference counted, so the view is the trace's contents, not a copy, and the lock is held for a chain rebuild, one reference count per spine level, never for merge work. Readers on any thread cut the chain through TraceReader, register compaction holds against the point, and import it as an arrangement with the analogue of TraceAgent::import_frontier_core. The writer applies the meet of its own TraceBox frontier and the readers' holds to the inner spine, so a shared arrangement compacts no faster than its slowest reader, and a reader that moves a hold wakes the arrange operator through a SyncActivator.

Every spine in typedefs is wrapped, so any arrangement can be published. Unattached, the wrapper costs one branch per trace call. The Materialize-side glue is Published, a publication point plus the standing hold (a reader with no physical hold, tracking the frontier the importing runtime has applied), PublishArrangement::adopt, which attaches an arrangement's trace to a point through TraceAgent::trace_box_unstable, and SharedTraceHandle, the reader carrying the publisher's peer count so imports refuse mismatched scopes.

Logical and physical compaction carry different frontiers throughout. Logical decides which times stay distinguishable, physical which batches may merge, and since is never the right physical bound. A reader's physical hold starts at the chain coverage, the frontier through which the published chain is complete, because that is where a seeded reader makes its first cut.

The published since is the trace's own compaction frontier, read off the spine after each mutation, so no controller frontier is forwarded through the registry. The point closes when the trace drops, which is when its last TraceAgent drops. Production keeps one in the trace manager. Tests keep one for as long as they read.

Inert: nothing in the crate calls it. The module is pub(crate) with a single #![allow(dead_code, unused_imports)] carrying a TODO(CPU-215), since crate::sharing arrives in the next PR of the stack. Exporting a crate-internal primitive on the public surface to keep dead-code analysis quiet would be the worse trade.

🤖 Generated with Claude Code

https://claude.ai/code/session_01VDm7opomJLxbNUEP3r9BLk

[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.)

[def-]

QA LLM Review

1. MEDIUM -- empty_logical_request_releases_the_hold can no longer fail

src/compute/src/shared_trace/tests.rs:687

Rewriting the assertion against the accumulated MutableAntichain made it vacuous: the handle registers at since and the standing hold sits at the same minimum time, so logical_holds() == standing_hold() holds identically whether or not the empty request released anything. The test named for the release property can no longer observe it, and no other test in the file does either.

Details

Verified by mutation. With set_logical_compaction changed to ignore an empty request outright, so the handle's hold is never released:

let next = self.logical.join(&frontier.to_owned());
if next.is_empty() { return; }   // hold is never released

all 17 shared_trace tests still pass. The concealed regression is real: an unreleased logical hold sits in logical_compaction for the life of the handle, the publisher forwards that frontier to its agent every activation, and the published arrangement stops compacting past it. That is exactly what the reduce operator triggers on every dataflow whose input finishes, per the test's own doc.

The accumulation is a meet, so a single hold at or above the standing hold is unobservable through logical_holds(). Releasing the standing hold first makes the accumulation carry the handle's contribution alone:

published.note_standing_hold(&Antichain::new());
assert!(published.logical_holds().is_empty());

let mut hold = published.handle();
assert!(!published.logical_holds().is_empty());
hold.set_logical_compaction(Antichain::new().borrow());
assert_eq!(published.logical_holds(), Antichain::new(), "...");

That version passes on the current code and fails on the mutation above.

[linear-code]

CPU-215

[antiguru]

Confirmed and fixed in c59f8e4. The test now releases the standing hold first, so the accumulation carries the handle's contribution alone, and it fails under the mutation described above.

The same commit moves SharedTraceHandle::frontiers, which had no production caller, into the test module.

Posted by Claude Code.

[antiguru]

We could turn capabilities into a blank CapabilitySet, without wrapping it in an Option: if it is empty (capabilities.is_empty() through the Deref impl) should be equivalent to the current let Some(...) = capabilities.as_mut(), and the None assignment equivalent to downgrading to the empty frontier. This would remove some implementation complexity.

[antiguru]

Done. capabilities is a CapabilitySet::from_elem(capability), the activation guard asks is_empty(), and the bound check downgrades to the empty frontier instead of assigning None. The as_mut().unwrap() and the extra nesting are gone.

Posted by Claude Code.

[antiguru]

We're not removing elements from chain, which means that we might retain batches that will be merged by the maintenance runtime, but kept alive here. This would be a memory problem.

[antiguru]

The chain is not appended to. Every activation rebuilds it from agent.map_batches and assigns state.chain = chain, so a batch the spine has merged away leaves the published chain at the publisher's next activation. Readers hold only Arc<SharedTrace>, and a cursor clones the chain for the duration of one read.

Two retention windows did exist, one of them systematic. The publisher read the chain, published it, and then called agent.set_physical_compaction, which can complete merges synchronously. The chain it had just published therefore named the pre-merge batches until the next activation, on every activation where the forwarded floor let the spine merge. That is fixed: the chain is re-read after the compaction call.

The other window is merges the arrange operator finishes on its own between publisher activations, through the activator the spine holds. Nothing downstream observes those, so the published chain pins that spine's pre-merge batches until the next stream or frontier event, which for a maintained collection is the next tick. For a collection whose upper has stopped moving there is no next tick, so the pre-merge set stays pinned alongside the merged batch until the dataflow drops. Bounded at one extra copy of that arrangement, and only for sealed collections. Closing it would need the publisher to poll while the chain has more than one batch, which I have not done.

Importer queues hold their batches until the importer drains, which is the unbounded-queue caveat already in the design.

Posted by Claude Code.

[antiguru]

The publisher changed shape in the latest push, so the earlier review threads on publish.rs (chain retention, stream versus trace upper, forwarding of the controller's frontier) describe code that no longer exists.

The sink operator is gone. SharedSpine in mz-timely-util is a Trace wrapper: every spine in typedefs is one, and once an arrangement is attached to a publication point the trace mirrors its chain, upper, and compaction frontiers into the point inside each mutation. The published since is the spine's own frontier, read back after the TraceBox applies the meet of the local agents' holds and the readers' holds, so nothing is forwarded through the registry any more (note_allow_compaction and Diagnostics::writer_logical are deleted in the later PRs). A mutex around the spine itself was rejected because roll_up and complete_at finish merges synchronously, so such a lock is held for whole merges; measured on a churn workload, writer mutations peaked at 276 ms while readers of the wrapper acquired cursors in 1.5 µs at p99.

Consequences elsewhere in the stack: PR 04 and PR 08 lose the publisher operator from the introspection goldens, and the arrangement-size doubling in introspection-sources.td went with the sink. Tests keep a TraceAgent alive while they read, since the point now closes when the trace drops, which is where the trace manager holds it in production.

Posted by Claude Code.

[antiguru]

Fix import order.

[antiguru]

Done, mz_row_spine moved up into the external group after mz_repr. The same misplacement was already on main in this file; fixed since the line was touched anyway.

Posted by Claude Code.

[antiguru]

Fix import order.

[antiguru]

Done, same move as in linear_join.rs.

Posted by Claude Code.

[def-]

QA LLM Review

1. MEDIUM -- the published since is written in a second lock acquisition, so a reader can latch a frontier the trace has already compacted past

src/timely-util/src/shared_trace.rs:389

apply_holds reads the readers' holds under the point's lock, releases it, and only then advances the inner trace's logical frontier and runs the merges that coalesce to it. state.logical is written afterwards, by a second acquisition in publish_frontiers/publish_chain. For the whole of that gap, which spans the merge (the PR's own measurement puts writer mutations at up to 276 ms), the point still advertises the pre-compaction since, so a reader thread can mint a handle at an as_of whose times the spine is coalescing away and read silently wrong data there.

Details

The gap is explicit in the code: shared_trace.rs:393-400 reads remote_logical/remote_physical and drops the guard; 404 applies the new logical frontier to the inner trace; 411-415 then applies the physical frontier, whose Spine::consider_merges completes merges synchronously, advancing update times to the logical frontier just set. Only after all of that does publish_frontiers (shared_trace.rs:328) or publish_chain (shared_trace.rs:345) take the lock again and write state.logical.

Shared::reader_at (shared_trace.rs:216) gates on state.logical and documents that "Checks and registers under one lock acquisition, so a returned reader holds a frontier the trace can still honour". That holds against the published value, but the published value is the stale one for the duration above, so Published::handle_at returns Ok for an as_of in [since_old, since_new).

The effect outlives the window. The reader registers its hold at the stale frontier in remote_logical, so the next apply_holds computes meet(local, stale) and calls inner.set_logical_compaction with a frontier below the one already applied. Spine::set_logical_compaction accepts a rewind without complaint (unlike the physical axis, which apply_holds:411 guards), so the published since regresses to a value the trace can no longer honour and stays there for the life of that handle.

The publisher this replaces did not have the gap: it read the accumulated holds and wrote state.since inside one acquisition and applied the result to its agent afterwards, so a reader either was counted in that round's meet or observed the new since.

Fix: compute both target frontiers and write state.logical in the same acquisition in which the holds are read, before touching the inner trace. publish_frontiers then has nothing left to do on the logical axis.

[antiguru]

Confirmed and fixed in fb18e16.

The report is right about both the window and the durable effect. apply_holds now takes one acquisition: it reads remote_logical/remote_physical, computes the meet against the local frontiers, and writes state.logical under that same guard, before the inner trace is touched at all. A reader that registers before us is counted in the meet; one that registers after sees the frontier we are about to apply, and reader_at refuses an as_of below it. So the interval where the point advertised a since the spine was coalescing away is gone, and with it the stale hold that pulled the applied frontier back down.

Publishing the target before applying it means the point can briefly advertise a since ahead of the trace's own. That refuses a reader the trace could still have served, which is the safe direction; the reverse is the one the report describes.

One thing the fix had to add that the report does not mention. A single trace can back several publication points, since one arrangement is published under every id that re-exports it, so the meet runs across all of them and they are now locked together. They are acquired in address order to make that deadlock-free, and the critical section is antichain arithmetic only, never merge work.

Posted by Claude Code.