compute: serve fast-path peeks on the interactive runtime

[antiguru]

Seventh of eight PRs splitting #37770. Stacks on #38391. Tracked by CPU-216.

The interactive runtime holds no local traces, so an index peek there resolves against the sharing registry. IndexPeek gains an IndexTraces source: Local pins a TraceBundle for the peek's life, Shared resolves handles from the registry on every attempt and holds nothing in between. Both hand the scan the same enum trace types, so one IndexPeek, one IndexPeekScan, and one PendingPeek::Index serve both runtimes and the walk, gate, offload, and cancel paths are written once.

A shared peek that is not ready lives in pending_work, keyed by the index it waits on, not in queued_peeks. A publication or seal marks that id dirty and wakes the worker, which re-examines exactly the peeks parked under the changed ids, so wakeups scale with what changed rather than with total pending work. Fast-path persist peeks still land in pending_peeks on every runtime, because their persist-read task wakes the worker with no dirty signal.

Interactive dataflows build immediately in command arrival order rather than deferring until their dependency is published. An import over an unbacked publication point produces no data and holds its output frontier at the minimum until a publisher adopts the same slot, so late binding replaces the deferral.

The interactive runtime reports only its transient collections' frontiers, since it shares the identity of every non-transient collection with maintenance and the controller keeps one frontier stream per collection. For the same reason its logging is forced off: it serves introspection peeks from maintenance's published copies. The consequence is that nothing the interactive runtime does appears in introspection, tracked as CPU-222. That wants resolving before the flag is turned on anywhere it would hide real work.

Reconciliation drops pending_work. The standing holds in the registry deliberately survive: one is per collection, carries no dataflow identity, and only rises, so clearing it would drop the arrangement's bound to the minimum until replayed compactions raised it again.

Reachable only on a runtime holding the Interactive role, which requires the dyncfg that is still off everywhere.

[antiguru]

Function moved for no particular reason.

[antiguru]

Agreed, no reason. Moved back to where #38389 introduced it, right after new(). This was churn imported by taking compute_state.rs wholesale from the original branch. (Posted by Claude Code.)

[linear-code]

CPU-215

[antiguru]

I wonder if this should be a function on the dataflow description. We're repeating the condition from the multiplexer.

[antiguru]

Agreed, and it is now DataflowDescription::is_peek_dataflow() in mz-compute-types, next to is_single_time(). #38388's router and this tripwire ask the same method of the same description, so the assert can finally check what it claims to: previously a drift between the two spellings was exactly what it was meant to catch, and it was written as a second copy of the thing that might drift.

The predicate also got stricter on the way: is_single_time() in place of !until.is_empty(), per the discussion on the router. Details in the reply on #38388.

Posted by Claude Code.

[antiguru]

This looks like duplication with the regular index peek logic, is that right? The peek offload work aimed at reducing duplication, and canonicalizing all peek structure (which it not has done completely, but that's a different story.) So my ask here would be to check if we can combine the walk, maybe by making them generic over the type of index they're walking.

[antiguru]

Right that it was duplicated, though not in the walk. The walk was already shared: both build a PeekScan and hand it to the generic walk_scan, which is what the peek-offload canonicalisation bought. What was copied is the ~25 lines of frontier gating in front of it, and the only differences were cosmetic: where (oks, errs) come from, and TraceBundle::compaction_frontier() versus an explicit antichain_join of the same two frontiers, which is literally what that method computes.

Extracted as gate_peek, generic over both traces via the existing PeekOksTrace/PeekErrsTrace markers, returning a PeekGate of Open/NotReady/Compacted. That required pinning Time = Timestamp on the two marker traits, which they should have carried anyway: both only ever describe traces over mz_repr::Timestamp, and only the cursor GATs were constrained before.

IndexPeek::collect_finished_data is deliberately left alone. It is main's method with main's tests in index_peek_tests.rs, which call it to exercise the walk while bypassing the gate, so folding it in would have churned code this stack has no business touching.

Posted by Claude Code.

[antiguru]

Is the design still part of the stack?

[antiguru]

It never was in this stack. design.md lives in #38239, which is open and design-only, and whose description says it lands first so the implementation PRs can cite it rather than restate the argument. So the reference resolves iff #38239 merges ahead of this stack, and today it points at a file on no landed branch.

That ordering was tracked nowhere, which is why it could go unnoticed. Filed as CPU-239 on the Interactive read isolation project, naming this call site as the reason: the standing-hold argument is written down only there.

Left the reference as it is. Inlining the argument here would duplicate the design rather than cite it, which is the thing the design PR exists to avoid.

Posted by Claude Code.

[def-]

QA LLM Review

1. MEDIUM -- IndexTraces::resolve clones both trace handles on every peek attempt, including on the default single-runtime path

src/compute/src/compute_state/index_traces.rs:45

The Local arm of resolve hands out oks.clone() / errs.clone(), and seek_fulfillment calls it before the frontier gate, so every attempt at every index peek now pays two TraceAgent clones plus two Drops. That is new cost on the Solo/Maintenance path that ships today, and it repeats on every sweep for a peek whose frontiers are not ready yet, so it scales with pending peeks times worker activations rather than with peeks answered.

Details

Before this PR the walk borrowed the pinned bundle (self.trace_bundle.oks_errs_mut()) and cloned nothing. The clone buys nothing on the local path either: PendingPeek::index (src/compute/src/compute_state.rs:1822) already pinned that bundle at the peek timestamp for the peek's life, and the clone just inherits and re-registers the same holds.

Each TraceAgent::clone/Drop pair emits two differential TraceShare events and runs four TraceBox::adjust_*_compaction updates plus the Spine::set_*_compaction calls they trigger. The differential/arrange logger is unconditionally registered in production (src/compute/src/logging/initialize.rs:272), so those events are real: four extra log records per peek attempt, flowing into the mz_arrangement_sharing logging dataflow. A +1/-1 pair that straddles a logging-interval boundary also makes that relation transiently over-report by one.

This also means the PR summary's "Reachable only on a runtime holding the Interactive role" does not hold for this hunk.

A shape that avoids it: give IndexTraces a with_traces(&mut self, f: impl FnOnce(&mut PeekOks, &mut PeekErrs) -> R) -> Option<R>. The Local arm stores PeekOks::Local/PeekErrs::Local once (built in PendingPeek::index) and passes &mut to them; the Shared arm mints its per-attempt handles into locals and passes &mut to those, keeping the "a parked shared peek holds nothing" property. gate_peek, PeekScan::new and walk_traces already take &mut, so only walk_traces' by-value parameters have to change.

2. MEDIUM -- the new wake tests hand-roll the dirty mark, so nothing covers the publication/seal wiring parked peeks depend on

src/compute/src/compute_state/tests.rs:770

interactive_peek_resolves_on_seal_via_note_frontier and interactive_peek_resolves_on_publication_not_on_bare_tick both publish with PublishArrangement::adopt(&oks, &slot.oks, "peek oks", || {}) (lines 637-638, 802-803) — a no-op on_seal — and then call registry.notify(...) by hand (lines 639, 804, 864). They therefore test resolve_dirty's contract but not the mechanism that produces the dirty mark in production, which is ArrangementSharingRegistry::publish passing move || registry.notify(id, worker_index) into adopt and calling notify itself after adoption.

Details

Concrete regression they would pass through: drop or break that on_seal wiring and a shared peek parked in pending_work is never re-examined, so every interactive fast-path peek that is not immediately ready hangs until the client cancels. The assertion message at line 740, "publication must have marked the id dirty", is checking the test's own notify call, not the publish path.

Nothing else in the tree closes the gap. sharing/tests.rs::seal_signal_dirties_its_worker also calls registry.notify directly, and insert_marks_dirty_and_take_drains cannot separate the publication mark from the seal marks because its execute_directly run produces both — removing either source still leaves it green.

Two smaller things in the same area: the doc comment at line 766 and the inline comment at line 863 refer to note_frontier and to "export_index's frontier hook", neither of which exists; the hook is the on_seal closure publish hands to adopt. Routing these tests through registry.publish(id, &oks, &errs) (as sharing/tests.rs::publish_index_into does) and letting the seal drive take_dirty would make the names accurate and cover the wiring.

[antiguru]

If this was keyed just by id, could we avoid the scaffolding to check interactive and peer_published, and instead just go by existence of state for id?

[antiguru]

Yes, and it goes further than removing the checks here. The scaffolding existed only because the interactive runtime installed empty copies of the logging indexes (initialize_logging forced logging off but still built the dataflow and inserted collections and traces). Those copies served nothing: peeks on the logging indexes read the maintenance runtime's publications from the registry, report_frontiers skipped them, and only the maintenance runtime publishes them.

initialize_logging now returns early on the interactive runtime, so it hosts only the transient dataflows the multiplexer routes to it, and "no collection for this id" is exactly "the peer publishes this id". handle_allow_compaction reduces to: unhosted id, record the standing hold and return, otherwise the normal path. report_frontiers loses its transient-only filter as well, since there is nothing left to filter.

The role appears once, in a soft_assert_or_log that the maintenance runtime never sees compaction for an unknown collection, which drop_collection used to panic on and would otherwise now pass silently.

Without a compute_logger the interactive runtime's own events are not logged, which was already the case with the copies (logging was forced off). That is the CPU-222 gap and there is a TODO pointing at it.

Posted by Claude Code.

[antiguru]

I think a match would read nicer here. Also check other places that could be a match instead of if.

[antiguru]

Done, match self.compute_state.role with Maintenance | Solo as the local arm. The other role checks in this file went away with the empty logging copies (see the thread above), and the two that remain in render.rs and logging/initialize.rs are early returns, which read fine as if.

Posted by Claude Code.

[antiguru]

Instead of making the decision here, could we generalize the is_subscribe_or_copy field and only condition on it? This would avoid the additional complexity here.

[antiguru]

Went the other way: the condition is gone rather than generalised. With the interactive runtime installing no logging dataflow (see the handle_allow_compaction thread), every collection it holds is a transient dataflow it hosts, so there is nothing left for report_only_transient to skip and is_subscribe_or_copy keeps its one meaning.

Posted by Claude Code.

[def-]

QA LLM Review

1. HIGH -- reconciliation panics on the interactive runtime at every controller reconnect

src/compute/src/server.rs:928

initialize_logging now returns early on Interactive (src/compute/src/compute_state.rs:1202), so that runtime installs no logging traces. reconcile still pads them unconditionally from the CreateInstance in its own command history, and traces.remove(id).expect("logging trace exists") finds nothing. Every controller reconnect after the first therefore panics the interactive worker, which the process-wide panic hook turns into a replica abort, i.e. a crash loop.

Details

CreateInstance is broadcast to both runtimes by the multiplexer (src/compute-client/src/multiplex.rs:165), ComputeCommandHistory::reduce retains it (src/compute-client/src/protocol/history.rs:112), and config.logging.index_logs is always populated (src/compute-client/src/controller/instance.rs:1234; LoggingConfig's own comment says it must be, even when logging is disabled). So old_instance_config is Some with a non-empty index_logs on the interactive runtime, and the loop body runs for ids that were never inserted. The first connection is fine only because the padding block sits inside if let Some(compute_state) = &mut self.compute_state, which is None before the first CreateInstance.

The base branch did not have this: initialize_logging there built the logging dataflow with logging forced off but still called traces.set(id, trace), so the padding found its traces.

Gate the block on the role that installs the traces:

// The interactive runtime installs no logging dataflow, so it has no logging traces to pad.
if compute_state.role() != ComputeRuntimeRole::Interactive {
    if let Some(config) = old_instance_config {
        for id in config.logging.index_logs.values() {
            let trace = compute_state
                .traces
                .remove(id)
                .expect("logging trace exists");
            let padded = trace.into_padded();
            compute_state.traces.set(*id, padded);
        }
    }
}

Keeping the expect inside the gate matters: a genuinely missing trace on maintenance should still trip it.

2. MEDIUM -- a seal on any published index wakes every interactive worker, whether or not a peek waits on it

src/compute/src/server.rs:551

Registering the waker (server.rs:599) is what makes ArrangementSharingRegistry::mark live, and mark inserts and unparks for every id a publisher notifies, with no reference to what any worker is actually waiting on. Each maintenance seal of each published index therefore unparks each interactive worker, so instead of sleeping to server_maintenance_interval the interactive runtime turns its loop roughly as often as the replica seals batches, and each turn takes the process-global registry mutex in take_dirty even when pending_work is empty.

Details

The cost is not just the wakeups. That mutex is the one maintenance's publishers take from publish_chain's on_seal (src/compute/src/shared_trace/publish.rs), on the arrange operator's critical path, so interactive-side polling contends directly with index maintenance. It scales with the number of published indexes times interactive workers, not with pending peeks: on an idle replica with no peeks outstanding at all, every index's frontier advance still costs one wake and one lock acquisition per interactive worker, and resolve_dirty then walks a dirty set holding every id that sealed.

pending_work is already keyed by the id a peek waits on, which is the information the registry lacks. Letting a worker declare which ids it waits on (registered under the same lock acquisition that parks the peek, so a mark cannot slip between the map read and the registration) would make mark skip both the insert and the unpark for an id nobody is parked on, and let take_dirty be skipped entirely while pending_work is empty.