Arc-backed spines for cross-thread arrangement sharing

[antiguru]

Summary

Adds the ability to read an arrangement from a thread other than the one that maintains it, in additive commits.

trace: add Arc-backed spine variants for cross-thread sharing

Adds arc_blanket_impls (a mirror of rc_blanket_impls using Arc) and the ArcOrdValSpine/ArcOrdKeySpine spines with their ArcOrdValBuilder/ArcOrdKeyBuilder, re-exported as ArcValSpine/ArcKeySpine/ArcValBuilder/ArcKeyBuilder.

This is purely additive. The default OrdValSpine/OrdKeySpine, the RcOrd*Builders, and the Val*/Key* aliases stay Rc-backed, so existing users are unaffected. An Arc'd batch whose contents are Send + Sync can be handed to and read from another thread, which the Rc variants cannot. Atomic reference counting is marginally more expensive (per batch clone/drop, never per record), so Rc remains the default and callers opt into Arc only when they need the sharing.

arrange: share arrangements across timely runtimes

Adds operators::arrange::sharing, a primitive for reading an arrangement from outside the owning worker, built on the Arc-backed spines:

• Arranged::publish attaches a publisher operator to an arrangement and returns a Published, whose handle() mints Clone + Send SharedTraceHandles.
• SharedTraceHandle implements TraceReader, so downstream operators drive its compaction and acquire cursors as with any trace handle.
• SharedTraceHandle::snapshot_at serves a consistent point/full-scan read from any thread, blocking until the publication frontier passes the query time and gating on since <= t so a read past compaction fails rather than returning stale rows.
• SharedTraceHandle::import replays the shared arrangement into another scope, so a second timely runtime can consume it as an ordinary Arranged.

The publisher forwards the meet of reader holds to its own TraceAgent (the sole writer of the trace's compaction), never the empty meet of zero holds. Handle clones carry independent holds. Compaction is applied to the agent outside the shared lock, since physical compaction can run an unbounded merge. The published since is the frontier the trace will actually hold after forwarding, so a concurrently registering handle cannot latch a floor below the trace's real compaction. A late importer registering after the publisher closed seeds its own terminal frontier rather than hanging.

arrange/sharing: import asserts equal total peers

Records the publisher's total peer count at publish time and asserts an importing scope's peers() matches it. Pairwise import (importer worker i reads publisher worker i) is sound only when both sides shard by key.hashed() % peers, which requires equal total peers (workers-per-process times processes), not just equal workers-per-process. A mismatch now panics at import time instead of silently reading the wrong shard.

arrange/sharing: publishing no longer pins compaction

Previously the publisher held its own TraceAgent at the publish-time since, which never advanced, so merely publishing an arrangement froze the trace's compaction for the life of the arrangement even with zero readers. The publisher now sources the writer-driven compaction frontier from the trace box (the meet of all other handles' holds, i.e. the box's counted meet minus the publisher's own contribution) and forwards that when no reader holds exist, and the meet of reader holds when they do. Publishing carries no independent compaction floor: only a live importer's own as_of hold holds the trace back, released on drop. The published since equals the trace's real compaction, so a live reader can still read at times it holds.

Motivation

This unblocks sharing an arrangement between two adjacent timely runtimes in one process (for example, a maintenance runtime and a latency-sensitive query runtime), and reading arrangement snapshots from non-timely threads.

Testing

• tests/trace.rs: reads an Arc'd batch from another thread.
• tests/sharing.rs: a cross-thread snapshot, a cross-dataflow import through a shared handle, publish_without_readers_does_not_pin_compaction, import_hold_pins_then_releases, and import_asserts_equal_peers.
• Full existing test suite passes; the Rc default path is unchanged.

🤖 Generated with Claude Code