compute: a per-process registry of published index arrangements

[antiguru]

Second of eight PRs splitting #37770. Stacks on #38386. Tracked by CPU-215.

Publishing an arrangement produces handles, but a reader on another thread needs a way to find them. This adds a registry keyed by GlobalId and worker ordinal, shared across all timely workers of a process the way the persist client cache is. Worker i publishes into slot i and a reader on worker i of another runtime looks up the same slot, which is sound only because both sides shard keys by key.hashed() % peers.

A slot can be created before it is filled. get_or_create lets whichever side touches an id first create the slot, backed by unbacked publication points, and the other adopt it in place, so a reader never overwrites a publisher's arrangement nor imports over a slot a later publish replaces. publish is the one entry point for a publisher: it adopts the slot's points and marks the id dirty. Readers learn about publication and seal through a dirty-id inbox that unparks the worker's thread rather than by polling.

Inert: nothing constructs a registry.

[antiguru]

Do we want to wake a specific worker or thread? For the latter, just use a handle to the thread and call unpark.

[antiguru]

A specific worker, and switched to the thread as you suggest.

register_waker is called from handle_compute_command on the interactive worker's own thread (server.rs), so std::thread::current() is available at the registration site, exactly as at the peek-offload one. Waker now holds a Thread and mark calls unpark().

Two reasons beyond consistency. Every timely allocator's await_events bottoms out in std::thread::park, so the unpark is not a coincidence of the current allocator choice being the thread one. And, as the offload path's comment already notes, a root-path SyncActivator additionally marks the worker's dataflows schedulable, which is work this wake does not need: the server loop drains take_dirty on its own, it does not need an operator rescheduled.

What is lost is SyncActivationError as a "worker has gone away" signal, which the doc called a backstop. It is not a state we survive into: a panic on either runtime's worker aborts the process through the shared-fate hook, so there is no live registry with a departed worker behind it. That paragraph is gone.

This also removed the ParkedWorker harness in the tests, about 80 lines. It existed only to keep activate() on its non-erroring path, and unpark has no erroring path. The three dirty-set tests now register thread::current(), which the harness doc already said was sufficient for what they assert. sync_activator_fires_cross_thread stays, retargeted at shared_trace's ImportQueue, which is the remaining user of a real activation.

Posted by Claude Code.

[antiguru]

Could this be inlined?

[antiguru]

Inlined. note_frontier was a bare delegation to notify, and its only other content was a doc sentence naming the caller. The frontier probes in render.rs and logging/initialize.rs now call notify directly, and the seal-signal sentence moved onto notify, where the rest of the caller contract already lives.

Posted by Claude Code.

[antiguru]

Should wakers be a read-write lock?

[antiguru]

No, every access to wakers mutates, so an RwLock would only ever take the write lock. notify and notify_all insert into the dirty set and set the coalescing flag, take_dirty takes the set and clears the flag, register_waker installs an entry. There is no read-only path to share.

The lock that could plausibly be an RwLock is map, which handles only reads. It stays a Mutex because get_or_create and reexport need the read and the write to be one critical section: the whole point is that whichever side touches a slot first creates it and the other observes the same Arc, which a read lock upgraded to a write lock would not give.

Posted by Claude Code.

[def-]

QA LLM Review

1. HIGH -- reexport overwrites a reader's placeholder slot, stranding its import forever

src/compute/src/sharing.rs:161

reexport writes slots[worker_index] = Some(arr) unconditionally, so when a reader has already created a placeholder for to and built an import over it, that point is dropped from the registry and never adopted. The import stays pinned at the minimum frontier for the life of the dataflow, which is a silent permanent stall, and it breaks exactly the convergence contract get_or_create documents two functions above.

Details

The placeholder path exists because the reader may touch an id before any publisher does. That applies to a re-exported id just as much as to a normally published one: from the interactive runtime's side, to is an ordinary index id, and it has no way to know the maintenance runtime will satisfy it by aliasing from. So the sequence is reachable whenever the interactive runtime processes a CreateDataflow importing to ahead of the maintenance runtime rendering the dataflow that exports it — the same command-arrival-order hazard join_over_point_adopted_late_matches_direct covers for the normal path.

The re-export producer is Context::export_index's ArrangementFlavor::Trace(gid, ..) arm (src/compute/src/render.rs:751), which today just clones the TraceBundle under the new id. Note that arm cannot be fixed by calling adopt instead: the flavor carries Arranged<RowRowEnter<..>>, not Arranged<TraceAgent<Tr>>, so registry-level aliasing really is the only mechanism available and the clobber has to be handled here.

Fix: reexport must not drop an occupied slot on the floor. Either back the reader's existing point in place — a Published-to-Published binding, the analogue of adopt, so handles already captured by value follow the alias — or, if that is too invasive for now, have reexport detect the occupied slot and surface it (return it to the caller, or assert) rather than swap the Arc. Worth a regression test mirroring get_or_create_converges_on_one_slot: mint a handle off get_or_create(to, ..), then reexport(&from, to, ..), and assert the handle observes from's rows.

2. MEDIUM -- Two aliased ids share one publication point's per-id compaction state

src/compute/src/sharing.rs:281

After reexport, from and to name the same SharedIndexArrangement, but each id has its own TraceAgent in the TraceManager and its own controller compaction frontier. note_allow_compaction assigns into the single writer_logical (last writer wins) and note_standing_hold joins into the single standing_hold, so the two ids' independent frontiers collide, and reads at an as_of the trace can still serve get refused.

Details

TraceBundle::clone at src/compute/src/render.rs:753 gives to an independent agent, so handle_allow_compaction drives allow_compaction(from, f1) and allow_compaction(to, f2) separately and the trace's real since is the meet of both. The published since is meet(publisher_after, writer_logical) with writer_logical holding only whichever of f1/f2 landed last, so it can sit strictly above the real since, and Published::handle_at then returns Err for an as_of the arrangement still supports. That Err is documented as a protocol-ordering failure callers report loudly, so the visible effect is a hard failure on a legitimate read.

The standing_hold join is the sharper half. Its whole purpose is to keep the publisher's agent at the frontier the importing runtime has applied for this id. Joining the two ids' applied frontiers lifts it to the higher one, so if the maintenance runtime has already advanced from's own agent, nothing holds the trace at the as_of the interactive runtime may still present for from, and it compacts past it. Both need per-id state: either keep writer_logical/standing_hold keyed by id inside the point, or give the alias its own point rather than sharing one.

[antiguru]

Both confirmed. Fixed in 0225ab1 by removing reexport and the alias map rather than repairing them.

A re-exported index now gets its own publication point. ArrangementSharingRegistry::publish is the single publisher entry point: it adopts the slot's points, so a placeholder a reader has already imported is backed in place exactly as on the primary path, and it notifies. #38389's re-export arms re-import the shared traces and call it, the same shape publish_logging_index already used. That gives each id its own writer_logical and standing hold, which settles the second finding as well, and it removes the alias fan-out from notify.

Posted by Claude Code.