compute-client: multiplex one command stream across two runtimes

[antiguru]

Third of eight PRs splitting #37770. Stacks on #38387. Tracked by CPU-216.

A replica running two compute runtimes still speaks one compute protocol to the controller. This adds the multiplexer that fans a single command stream out to both and merges their responses into one, so neither the controller nor the protocol learns that the replica is split.

Routing follows collection identity rather than command kind. Lifecycle commands go to both runtimes, a dataflow goes to the runtime that will host it, and AllowCompaction for a maintained collection is broadcast to both because the interactive runtime may be reading a published copy of it. Frontier reports are forwarded only from the owning runtime, since the controller keeps one frontier stream per collection and two reporters would race and regress it. Peek responses are forwarded verbatim, because exactly one runtime answers a peek.

A dataflow goes to interactive when it is a peek dataflow: DataflowDescription::is_peek_dataflow, added here next to is_single_time, meaning it installs transient collections, reads a single time, and drives no sink that outlives the read. Each clause earns its place. Transience because filter_response forwards interactive's frontier reports only for transient ids. Single-time because the shared-index import is a snapshot bounded one step past as_of and cannot feed a dataflow that runs further, and because transience alone says nothing about when a dataflow stops: a REFRESH AT materialized view is durable and still gets a finite until, an introspection subscribe is transient and never stops. The predicate lives on the description so that #38392's tripwire can ask the same question of the same value rather than restate it.

Routing reads only the description in front of it, never transient_owner. A peek dataflow reads published maintenance indexes, not another temporary collection, so that is sound today, but nothing enforces it. A soft assertion now fails loudly if a peek dataflow ever imports a transient id, whose producer might sit on the other runtime. Rerouting instead would make such a case work and therefore never surface. The durable fix is for the control plane to name the runtime in the description, recorded as a TODO(CPU-216) rather than folded in here, since it surfaces placement in the protocol.

Inert: nothing constructs a multiplexer. Independent of #38386 and #38387; it sits above them only to keep the stack linear.

[linear-code]

CPU-215

[antiguru]

We need to document that no object can depend on transient descs, otherwise a downstream dependency might render on the other runtime, which would be surprising. I think we're avoiding this by restricting to dataflows with a non-empty until, but still worth pointing out.

[antiguru]

Your worry is right, and it was unenforced. CreateDataflow decided routing from the description's own shape and never consulted transient_owner, which only owner_of reads for Schedule, AllowWrites, and AllowCompaction. So a chain of temporary objects worked only because the predicate is uniform, not because anything held it.

Now asserted rather than assumed: a dataflow routed to interactive must import no transient id. Loud rather than silently rerouted by owner_of, per your point about recognising the situation. Rerouting would make the case work and therefore never surface, and the placement decision you want the control plane to make would lose its forcing function. The TODO(CPU-216) on the assert records that alternative, naming the trade you identified: it is the control plane's concern, at the cost of surfacing placement in the protocol.

Transience alone was never the guarantee, as you say. Introspection subscribes are transient and unbounded, and they are excluded by the subscribe clause, not by transience. The routing comment now says what each clause buys instead of leading with transience.

Posted by Claude Code.

[antiguru]

Should it be !until.is_empty() or until = [as_of.step_forward()]?

[antiguru]

until = as_of.step_forward(), and that already has a name: DataflowDescription::is_single_time().

The whole predicate moved onto DataflowDescription as is_peek_dataflow(), next to it, and both this site and handle_create_dataflow's tripwire in #38392 now ask the same method. !until.is_empty() was weaker than the interactive path needs: import_shared_index bounds its snapshot one step past as_of regardless of desc.until, so a transient dataflow with until beyond as_of + 1 would render with silently truncated inputs. Not reachable today, because optimize::peek sets until = as_of + 1 exactly, but nothing enforced the coincidence.

Switching also settled the as_of = MAX case, which I had wrong. I first thought it needed excluding, because is_single_time() is true there while until is empty. It does not: render already filters try_step_forward for exactly this, giving the empty snapshot bound and the semantics of reading the final state. So the emptiness of until at MAX means end-of-time, not unbounded, and the read routes to interactive like any other single-time read. routing_excludes_copy_to_and_subscribe_and_unbounded now covers both: MAX with empty until goes to interactive, a finite as_of with empty until stays on maintenance.

Confirmed end to end rather than argued. #38393 gains read_at_max_timestamp.slt, which reads at 18446744073709551615 off constant-backed indexes (constant collections seal, so a read at MAX is answerable at all), both as a fast-path peek and as a join that has to build a peek dataflow. Passes with the flag on and off.

Posted by Claude Code.

[def-]

QA LLM Review

1. MEDIUM -- frontiers_dropped_from_non_owning_runtime is a coin flip: it misses a broken frontier filter ~half the time

src/compute-client/src/multiplex/tests.rs:842

Both mock channels have a message queued before the single recv(), so tokio::select! picks a runtime at random. When it picks maintenance first the assertion is satisfied without the filter ever running, and interactive's empty-frontier report is left unread in the channel. This is the only test covering filter_response's drop path, so a regression there escapes CI roughly half the time.

Details

Measured on this branch: with filter_response's interactive arm changed to Runtime::Interactive => true (i.e. the filter removed), the test passed 24 of 40 runs, and the whole multiplex::tests suite passed 9 of 20 runs.

Note also that the assertion cannot distinguish "dropped" from "delivered second": it only inspects the first response. Draining both would make the test deterministic:

h.inter_tx.send(/* interactive's empty frontier */).expect("send");
h.maint_tx.send(frontiers(id, 100)).expect("send maint");

// Exactly one frontier for `id` reaches the controller, and it is maintenance's.
let got = h.mux.recv().await.expect("recv");
assert!(
    matches!(&got, Some(ComputeResponse::Frontiers(g, f))
        if *g == id && f.write_frontier == Some(Antichain::from_elem(Timestamp::from(100u64)))),
    "expected maintenance frontier, got {got:?}",
);
// Nothing further: interactive's report was dropped, not queued behind maintenance's.
assert!(
    tokio::time::timeout(Duration::from_millis(50), h.mux.recv()).await.is_err(),
    "interactive's non-transient frontier must not be forwarded at all",
);

The regression this conceals is the one the test's own comment names: the controller keeps one frontier stream per collection, so interactive's empty copy of an introspection index reporting an empty write_frontier regresses a maintained collection's frontier to the empty antichain.

2. LOW -- module doc points at a design file that will not exist

src/compute-client/src/multiplex.rs:26

The broadcast-compaction rationale cites doc/developer/design/20260720_two_runtime_compute/broadcast-compaction.md. That path is absent from the tree, and the design PR in this stack adds only design.md to that directory while explicitly listing broadcast-compaction.md among the superseded working notes. Repoint the reference at design.md.

[antiguru]

Both fixed in 287fc37. The test drains past the first response and asserts nothing further arrives within 50 ms, so a removed filter fails it every time. The module doc no longer points at a design file; the broadcast rationale it cited is stated inline.

Posted by Claude Code.

[antiguru]

Add a comment: GenericClient::recv cancellation-safe by invariant.

[antiguru]

Added at the select!. The trait doc on GenericClient::recv is what mandates it, so the comment cites that.

Posted by Claude Code.