diff --git a/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md new file mode 100644 index 0000000000000..2eb44cb4f0f52 --- /dev/null +++ b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md @@ -0,0 +1,480 @@ +# Shared arrangements across timely runtimes + +- Associated: + - [Two-runtime compute](../20260720_two_runtime_compute/README.md): the + compute-layer design this primitive exists to serve. + - [background.md](./background.md): how a single-runtime arrangement works + today (the status-quo baseline this design departs from) + - differential-dataflow branch `claude/spines-differential-arc-j93mho` + (Arc'd batches, shared-trace primitive) + - materialize branch `claude/spines-differential-arc-j93mho` + (Send + Sync batch assertions) + +## The Problem + +A replica runs one timely runtime. That runtime does two very different kinds +of work on the same worker threads: + +1. CPU-bound maintenance: keeping indexes and materialized views up to date, + including arrangement merges and operator work proportional to input volume. +2. Latency-sensitive reads: serving peeks against arrangements and rendering + short-lived dataflows for ad-hoc queries. + +These collide. A peek against an index must wait for the worker that owns the +relevant arrangement shard to come around to it, and that worker may be deep +in a join or a large merge. The result is tail latency on reads that is +proportional to how busy maintenance is, which is exactly when users notice. + +The underlying restriction is that arrangements are worker-local. Batches are +reference counted with `Rc`, trace handles are `Rc>`, and both +are pinned to the worker thread that built them. Nothing else in the process +can look at an arrangement, so all reads must be scheduled onto the owning +worker. + +The Arc'd-batches work on the associated branches removes the first half of +that restriction for the stock differential spines: their batches are now +`Arc`'d, so a batch can be handed to and read from any thread. Materialize's +batch *contents* (lgalloc regions, `CompactBytes`, columnation stacks) are +verified `Send + Sync`, though Materialize's own spine typedefs still wrap +those contents in `Rc` and must migrate to `Arc` before they can be shared +(tracked as step 4 of the prototype). What is missing beyond the wrapper is +the machinery to hand over entire arrangements, with correct frontiers and +compaction, to code running outside the owning worker. That machinery is the +subject of this design. + +## Success Criteria + +- A second timely runtime ("query runtime") in the same process can import an + arrangement maintained by the main runtime and use it as an ordinary + `Arranged` collection: render joins, reduces, and other operators against it + without copying or re-arranging the data. +- A non-timely thread (or the query runtime) can serve a point or full-scan + read against a consistent snapshot of an arrangement at a given time, + without scheduling work on the main runtime's workers. +- Readers hold back compaction only as far as necessary: the main runtime's + arrangement compacts to the meet of all registered read holds (their greatest + lower bound, so it never compacts past the least reader's hold), and no + further than the published `since` when there are none. Holds release + deterministically when readers drop. +- The main runtime's write path pays only constant overhead per batch + (an `Arc` clone and a mutex push), independent of batch size. +- No unsafe code in the sharing layer, and no changes to the `Trace`/ + `TraceReader` contracts that existing operators depend on. + +## Out of Scope + +- Cross-process sharing. Everything here is threads within one process. + Cross-process sharing is persist's job. +- Sharing the chunked columnar traces (`trace::chunk`, `columnar` in + differential). Their batch internals (`Rc`'d `ColChunk`, `OnceCell` page + cache) are not yet thread-safe. They keep `Rc` batches for now and can adopt + the same machinery once their internals move to `Arc`/`OnceLock`. +- Materialize integration (compute controller owning two runtimes, peek + routing policy, introspection of shared traces). This design describes the + differential-level primitive and the architecture it enables. Wiring it into + `mz-compute` is follow-up work. +- Scheduling policy inside either runtime (thread counts, pinning, priorities). + +## Solution Proposal + +Two adjacent timely runtimes in one process. The main runtime maintains +indexes and materialized views exactly as today. The query runtime renders +ephemeral dataflows and serves peeks. Arrangements cross the boundary through +a per-arrangement, per-worker *publication point*: a small mutex-guarded +structure through which the owning worker publishes its trace's batches and +frontiers, and from which readers on other threads import them. + +The primitive lives in differential dataflow as a new module, +`operators::arrange::sharing`, built entirely on public trace APIs. + +### The unit of sharing is the batch chain, not the spine + +A `Spine` cannot cross threads: it holds a timely `Logger` (`Rc`-based) and +in-progress mergers. It also should not cross threads, because it is the +mutable half of the arrangement and has exactly one writer. + +What crosses the boundary is the spine's *contents*: a chain of `Arc`'d +batches with contiguous descriptions, together with the trace's `since` +(logical compaction) and `upper` (seal) frontiers. Batches are immutable, so +a chain plus frontiers is a consistent, self-describing view of the +arrangement. When the main runtime later merges batches, readers holding the +old chain are unaffected: their `Arc`s keep the pre-merge batches alive, and +they simply hold that memory until they drop. + +### Publisher: an operator on the main runtime + +Sharing is initiated on the owning worker, from an existing arrangement: + +```rust +let arranged: Arranged> = collection.arrange(); +let published: Published> = arranged.publish(); +let handle: SharedTraceHandle = published.handle(); // Clone + Send +``` + +`publish` attaches a sink-like operator to `arranged.stream`. That stream +carries the non-empty batches the arrange operator emits, each under a +capability whose time lower-bounds the batch's updates. On each activation the +publisher, under the shared lock: + +- appends each newly received batch to every registered reader's replay queue + as `TraceReplayInstruction::Batch(batch, Some(cap_time))`, recording the + message's capability time as the replay hint, then a + `TraceReplayInstruction::Frontier(batch.upper())`, mirroring + `TraceWriter::insert`. It wakes each reader (see wakeup below). +- refreshes the published chain from `TraceReader::map_batches` on its trace + handle. This is the authoritative source of the published `chain`, `since`, + and `upper`, for three reasons the arrangement stream cannot satisfy: + `map_batches` includes the seal-only empty batches that `arrange_core` + inserts into the trace but never sends downstream (an idle arrangement with + a ticking frontier produces only these). It reflects the main runtime's + compaction so *new* readers start from the already-merged state rather than + replaying history. And its last batch's `upper` is the trace's true seal + frontier. The published `upper` is taken from the chain (the join of batch + uppers via `read_upper`), never from the operator's input frontier, which + lags the trace by at least one progress round. +- forwards compaction to its own `TraceAgent` (details in "Compaction + forwarding" below). + +The publisher owns a `TraceAgent` clone, so the trace cannot compact or be +dropped out from under remote readers: the publisher's read capability *is* +the aggregate lease for all of them. Dropping the publisher marks the state +`closed` and enqueues a final `Frontier(empty)` at the end of each reader's +queue, so readers drain everything already published before shutting down. + +Cost on the write path: one `Arc` clone and one `VecDeque` push per batch per +reader, plus a mutex acquisition per activation. No per-record work. + +#### Compaction forwarding + +> **Corrections required (from adversarial review and the compute integration).** +> This section describes the current `sharing.rs` behavior, which has two defects +> the [two-runtime compute design](../20260720_two_runtime_compute/README.md) +> depends on fixing. First, with no readers the publisher forwards its own +> `get_logical_compaction`, a frozen hold that never advances, so merely +> publishing pins the trace's compaction at publish-time `since` forever. +> Publishing must carry no independent compaction floor. The trace's writer and +> the controller drive `since`, and only a live importer's own `as_of` hold +> (released on drop) may hold it back. Second, `snapshot_at` must enforce a +> `since <= t` gate so a point read past compaction errors rather than returning +> stale rows. Read holds from `import` are correct and intended (they are how a +> long-lived read keeps its `as_of`, and how a subscribe would), they just must +> never sit below the controller's own hold. + +Readers register holds. The publisher aggregates them and advances its +`TraceAgent`, which is the only writer of the underlying trace's compaction +frontiers. Two hazards shape how: + +- *`set_physical_compaction` is not cheap.* On the spine it calls + `consider_merges`, which can complete an in-progress merge with unbounded + fuel synchronously. It must therefore run *outside* the shared lock, or a + large merge would stall every concurrent peek and import. The publisher + computes the target frontiers under the lock (updating `state.since` first, + so a reader registering concurrently cannot latch a stale lower `since`), + releases the lock, then calls `set_logical_compaction` / + `set_physical_compaction` on the agent. +- *Compaction is irreversible and the empty meet is destructive.* The meet of + an empty hold registry is the empty frontier, i.e. "compact everything". + Forwarding it releases the publisher's read capability permanently, after + which `batches_through` panics (it asserts a non-empty logical frontier) and + no future reader can re-establish a floor. So the publisher never forwards a + bare meet. With no readers it holds compaction at the published `since` and + `upper` and advances only the join of holds that actually exist. Logical and + physical holds are tracked and forwarded independently, because consumers + like `join` legitimately advance one far ahead of the other. + +### Shared state + +```rust +struct SharedTraceState { + /// The published chain, sourced from `map_batches`: contiguous + /// descriptions including seal-only empty batches. Its coverage is at + /// least `upper` (it may run ahead within a worker step, see below). + chain: Vec, + /// Logical compaction frontier of the published view. Reads at times not + /// beyond `since` are not accurate; peeks pick a time at or beyond it. + since: Antichain, + /// Seal frontier: the join of the chain's batch uppers. Batches strictly + /// below `upper` are complete and readable. + upper: Antichain, + /// Per-registration logical holds. The publisher forwards their meet, + /// falling back to `since` when empty (never the destructive empty meet). + logical_holds: BTreeMap>, + /// Per-registration physical holds, forwarded independently of logical. + physical_holds: BTreeMap>, + /// Per-registration replay queues and wakeups. Keyed by registration id, + /// not by handle: one handle may back several registrations. + readers: BTreeMap>, + /// Set when the publisher drops. Terminal `Frontier(empty)` is enqueued + /// per reader. Readers act on that queue entry, not on this flag directly. + closed: bool, +} + +/// Wakeup for a registered reader. Import registrations carry a timely +/// `SyncActivator`; peek waiters carry a `Condvar`/thread handle, since a +/// non-timely thread has no operator to activate. +enum ReaderWakeup { + Operator(SyncActivator), + Thread(Arc), +} + +type SharedTraceRef = Arc>>; +``` + +One instance per (arrangement, worker). Every operation under the lock is +short: push a batch, clone the chain, update an antichain, read holds. The one +expensive step, applying compaction to the `TraceAgent`, is deliberately done +outside the lock (see "Compaction forwarding"). Contention is bounded by +publish and read frequency, both per-batch, not per-record. + +Ids are assigned by a counter in the shared state. Distinguishing *reader ids* +(one hold-and-queue registration) from *handles* matters because holds must be +independent per consumer, addressed next. + +### Reader: handle, snapshot, and import + +`SharedTraceHandle` is `Clone + Send` and holds a `SharedTraceRef` +plus its own registration id, minted from the shared counter. It implements +`TraceReader` directly: + +- `batches_through(upper)` locks, finds a clean cut of the published chain at + `upper` (skipping empty batches, as `Spine::batches_through` does), and + returns `Arc` clones of the covering batches. The lock is released before + any cursor work. Reading never blocks the publisher longer than a chain + clone. +- `set_logical_compaction(frontier)` writes this registration's entry in + `logical_holds`. The publisher applies the join of all entries on its next + activation, so compaction is asynchronous but conservative: the trace never + compacts beyond a registered hold. +- `set_physical_compaction(frontier)` writes `physical_holds` likewise. + Logical and physical are independent: `join` advances a trace's logical + compaction to its *opposing* input's frontier while keeping its physical + hold at its *own* acknowledged batch boundary, and the two routinely + diverge. The "physical equals logical" default applies only to readers that + never call `set_physical_compaction`. + +`clone()` is where independence is enforced. It cannot share a registration +id, because `import` returns `Arranged { trace: handle.clone() }` and +`Arranged` is itself `Clone`, so two downstream operators (say a join and a +reduce) each drive compaction on their own clone. If they shared one hold +slot, the faster one would release the slower one's hold and the trace would +compact under it, corrupting the slower reader. So `clone()` mints a fresh +registration id under the lock and copies the source's current holds into it, +exactly as `TraceAgent::clone` registers an independent counted hold in +`TraceBox`. Dropping a handle removes only its own registration's holds and +queue. + +A subtlety compaction inherits from the trace: a reader registering late gets +a *logical* hold at the current published `since`, but its effective *physical* +floor is whatever physical frontier the publisher has already forwarded (near +`upper`), because physical compaction cannot retreat. This is harmless for +import-based readers, since every `cursor_through` frontier they use is derived +from chain state at or after their registration. + +Two read paths sit on top: + +1. **Peeks.** `handle.snapshot()` returns a `TraceSnapshot`: an owned + `(chain, since, upper)` triple. It exposes `cursor()` over a `CursorList` + of the batches' cursors (`Arc` implements `Navigable`). A peek picks a + time `t` with `since ≤ t` and `upper` not `≤ t`, so all updates at `t` are + complete. If `upper` has not yet passed `t`, the peek registers a + `ReaderWakeup::Thread(Condvar)` and waits on it under the shared lock, + which closes the lost-wakeup window: the publisher signals the condvar + under the same lock whenever it advances `upper`. This works from any + thread, timely or not. + +2. **Imports.** `handle.import(scope)` on a query-runtime worker builds a + source operator, mirroring `TraceAgent::import_core`: + - registration, under one lock acquisition, mints a registration id, + installs a replay queue keyed by that id and a + `ReaderWakeup::Operator(worker.sync_activator_for(...))`, and seeds the + queue with the current chain as `Batch` instructions (hint + `Time::minimum()`, as `new_listener` does for historical batches) + followed by `Frontier(upper)`. Taking the chain and installing the queue + atomically is what guarantees no batch is missed or duplicated: later + batches land in the queue, earlier ones are in the seed. + - on activation the source drains its queue. It emits each non-empty batch + under a capability *delayed to that batch's hint* (a lower bound on the + batch's times), and applies each `Frontier` by downgrading its + capabilities, exactly as `TraceAgent` import replay does. Empty batches + carry no hint and are not emitted, only their frontier is applied. Note + the hint is a lower bound, never the batch's `upper`: emitting at `upper` + would let the query runtime's frontier advance past updates still inside + the batch and break `join`. + - the queue is owned by the source operator, not the handle. The + registration is a guard captured in the source closure, so dropping the + import dataflow deregisters it and releases its holds even while the + query worker and other handle clones live on. Tying queue lifetime to the + handle instead would leak: a `SyncActivator` for a live worker keeps + succeeding, so a dropped-dataflow-but-live-worker reader would never be + detected, its queue would grow without bound, and its holds would wedge + compaction. `SyncActivator` failure detects only a fully-gone worker, so + it is a backstop, not the primary deregistration path. + - the result is `Arranged { stream, trace: handle.clone() }`, so every + existing arrangement-consuming operator works unchanged against the + shared trace. + +#### Initial state + +The shared state exists at `publish()` time so `handle()` can hand out +`Clone + Send` handles that peek or import before the publisher's first +activation. It initializes to `chain = []`, `since =` the publisher agent's +`get_logical_compaction()` at publish time, and `upper = +Antichain::from_elem(Time::minimum())`. The last point is a trap worth naming: +the `Antichain::new()` default is the *empty* frontier, which reads as +"complete through the end of time", so every peek's wait condition would be +vacuously satisfied and a peek would read the empty chain as a valid snapshot +and return wrong, empty results instead of blocking. + +### Partitioning: pairwise import + +Arrangements are sharded across workers by a deterministic exchange of the +key. A shared arrangement is therefore N publication points, one per +main-runtime worker. For rendered dataflows the query runtime must preserve +that sharding: it runs with the same number of workers, and query worker `i` +imports main worker `i`'s publication point. Keys then live on the same query +worker that any downstream `arrange` of the same key would route them to, so +joins between two imported arrangements of the same key are co-located without +any exchange. + +The soundness of pairwise import rests on the exchange being a deterministic +function of the key and worker count alone. Confirming that Materialize's +arrangement exchange is exactly that function (and not one that also depends +on, say, dataflow-local state) is part of the Materialize integration work, +and equal worker counts between runtimes is a hard requirement the import path +asserts. + +This pairing is a correctness requirement for imports, not a preference. The +implementation asserts `peers` match at import time. Peeks are exempt: a +peek may read any or all publication points from any thread, since it does +not feed exchanged operators. + +### Consistency and correctness argument + +- *No torn reads.* The chain, `since`, and `upper` are only ever updated + together under the lock, and readers copy all three under the same lock. + Every snapshot is a frontier-consistent view. +- *No missed or duplicated batches on import.* Registration atomically takes + the current chain and installs the queue. Batches arriving after + registration land in the queue. Batches before it are in the chain. The + publisher appends to queues under the same lock it uses to refresh the + chain. +- *Compaction safety.* The publisher's `TraceAgent` capability lower-bounds + everything it has published, and it advances only to the meet of registered + holds (never the empty meet, never below `since`). `state.since` is updated + under the lock before compaction is applied to the agent outside the lock, + so a concurrently registering reader cannot latch a `since` the trace has + already passed. A reader reading at a time not beyond `since` therefore + always sees accurate times. +- *Progress.* `upper` is the join of the published chain's batch uppers, which + is the trace's true seal frontier (the operator's input frontier only lags + it). Within a worker step the chain may briefly cover *more* than the last + published `upper`, never less, so the invariant is "chain coverage ≥ + `upper`", and peeks that bound times by `t < upper` stay correct. Import + capabilities downgrade only on `Frontier` instructions, so query-runtime + frontiers are always justified by the main runtime's progress. +- *Shutdown.* Publisher drop marks `closed` and enqueues a terminal + `Frontier(empty)` at the tail of each reader's queue. Readers act on that + queue entry, after draining every batch ahead of it, so nothing published is + dropped. Import registrations deregister via their source-closure guard when + the import dataflow drops, which releases holds even while the query worker + lives. `SyncActivator` failure (whole query worker gone) is only a backstop. + +### Memory considerations + +A snapshot pins its chain until dropped. A long-running peek over a large +arrangement holds pre-merge batches alive while the main runtime replaces +them. The cost is bounded by chain size at snapshot time, and is exactly the +cost the main runtime would pay if it had `cursor_through` open locally. +lgalloc supports freeing from foreign threads (its `Handle` is +`Send + Sync`), so batches dropped on the query side release correctly. +Accounting attribution moves to whoever drops last, which introspection needs +to be aware of eventually (out of scope here). + +## Minimal Viable Prototype + +The prototype is the differential branch itself, in four steps, each with +tests: + +1. Arc'd batches (done): the stock `ord_neu` spines (`OrdValSpine`, + `OrdKeySpine` over the `Vector` layout) use `Arc`, the full test suite + passes, and a test reads a batch from a foreign thread. +2. Materialize batch contents verified `Send + Sync` (done): compile-time + assertions in `mz-row-spine` over the production `Row`/`Timestamp`/`Diff` + layouts. Note this covers the *contents*, not the `Rc` wrapper. +3. The `sharing` module (this design): publisher, handle, snapshot, import, + built and tested against the stock Arc'd spines. Tests run a main runtime + and a query runtime as two `timely::execute` instances in one process and + verify each of: + - a peek snapshot sees exactly the published collection at a chosen time. + - an imported arrangement renders a correct dataflow in the query runtime + and tracks ongoing updates. + - two consumers of one import hold compaction independently. + - read holds keep the main trace from compacting, and release on drop. + - publisher shutdown drains queued batches, then closes importers. +4. Materialize adoption of Arc spines (done): Materialize's production spines + now wrap batches in `Arc` with `ArcBuilder`. `RowRowSpine`, `RowValSpine`, + `RowSpine`, `ValRowSpine` in `mz-row-spine` and `ColValSpine`, `ColKeySpine` + in `mz-compute` typedefs migrated from `Rc`/`RcBuilder`. `ErrSpine` and the + other error and generic spines are aliases of these, so they follow. The + only consumers that needed touching were those that named the batch wrapper + directly: the arrangement-size logger (`Rc::downgrade`/`Weak` over batches) + and the storage sink's stock `OrdValSpine` builder alias. The step-2 + assertions guaranteed the contents were ready, so only the wrapper changed. + Materialize depends on `differential-dataflow` through a `[patch.crates-io]` + entry pointing at the fork until an upstream release carries + `arc_blanket_impls`. + +## Alternatives + +- **Share the whole spine behind a mutex.** Rejected: `Spine` is `!Send` + (timely `Logger`), holds in-progress mergers, and has a single-writer + ownership story that a shared mutex muddles. It would also serialize + maintenance merge work against reads. The chain-of-batches view gives + readers everything they can use. +- **Ship batches over a channel and rebuild a spine on the query side.** + Rejected: the query runtime would pay merge CPU and hold a second copy of + merged batches, and it could not benefit from the main runtime's + compaction. It also duplicates `since`/`upper` management badly. +- **Serialize batches across the boundary.** Strictly worse in-process: + copies all data, loses the columnar in-memory layout, and equals what + persist already offers across processes. +- **Priority scheduling inside one runtime.** Timely scheduling is + cooperative per worker, so a peek behind a long-running operator step cannot + preempt it. Yielding more finely helps tail latency but couples maintenance + throughput and read latency forever, and does nothing to isolate ephemeral + dataflow rendering. +- **`TraceRc`-style read-counted wrappers extended across threads.** The old + `TraceRc` machinery solved hold accounting worker-locally with + `Rc>`. Generalizing it directly to `Arc>` puts a + lock on the owning worker's hot read path too. The publisher/handle split + keeps the owning worker lock-free except at publication points. + +## Open questions + +- **Physical compaction policy.** Logical and physical holds are independent + (a reader that never sets a physical hold defaults it to its logical hold). + Whether the default keeps more batch boundaries alive than join actually + needs is a question to measure before refining. +- **Publisher wakeup latency for hold changes.** Reader hold changes take + effect on the publisher's next activation. Applying compaction outside the + lock (see "Compaction forwarding") means an idle arrangement with an + unscheduled publisher applies hold *releases* lazily, which delays freeing + memory but is never a correctness problem. If it matters, readers can wake + the publisher through a main-runtime `SyncActivator` registered alongside + the publication point. +- **Lease expiry.** RAII holds are correct in-process, but a query thread + stuck in a long computation holds compaction back. Do we want time-based + lease expiry with snapshot invalidation, or is same-process trust enough? + This design assumes the latter for now. +- **Wakeup of the publisher.** Reader hold changes take effect on the + publisher's next activation, which today means the next batch or frontier + movement. An idle arrangement applies hold releases lazily. If that + matters, the publisher can also be woken by readers through a main-runtime + `SyncActivator`. +- **How Materialize adopts it.** One replica process running two runtimes + needs controller work: which dataflows land where, how peeks choose a + runtime, how introspection reports shared traces and their memory. Separate + design. +- **Logging.** Shared-trace activity (publications, imports, holds) is + invisible to timely/differential logging today. Decide what to log and + where once the mz integration design exists. diff --git a/doc/developer/design/20260719_shared_arrangements_across_runtimes/background.md b/doc/developer/design/20260719_shared_arrangements_across_runtimes/background.md new file mode 100644 index 0000000000000..1c2c92df80a81 --- /dev/null +++ b/doc/developer/design/20260719_shared_arrangements_across_runtimes/background.md @@ -0,0 +1,269 @@ +# Background: how a single-runtime arrangement works today + +This is a status-quo companion to the [shared arrangements design](./README.md). +It records how differential dataflow builds, merges, compacts, reads, and +imports an arrangement inside one timely runtime. The design proposes crossing +a thread boundary. This document is the baseline it departs from, so that the +design reads as a delta and its assumptions are visible rather than implicit. + +Nothing here is a proposal. It is a description of existing behavior, grounded +in source. References point at `differential-dataflow/src/...` on the branch +behind TimelyDataflow/differential-dataflow#807 (the Arc-batches fork). Line +numbers drift, so treat them as pointers to the right function, not addresses. + +## The data model: immutable batches with a description + +An arrangement is a collection of updates `(data, time, diff)` organized into +*batches*. A batch is immutable once built. The `Batch` trait is documented as +"An immutable collection of updates" (`trace/mod.rs:236`), and the spine is +"based on collection and merging immutable batches of updates" +(`trace/implementations/spine_fueled.rs:9-11`). Merging never mutates its +inputs. `Merger::done` produces a new output batch (`trace/mod.rs:321-323`) and +hands the source batches back to the caller to drop. + +Every batch carries a `Description` (`trace/description.rs:68-97`) with three +frontiers: + +- `lower`: the inclusive lower bound of the times the batch describes. +- `upper`: the exclusive upper bound. Together `[lower, upper)` is the batch's + slice of time. `upper` is also the seal frontier: updates strictly below it + are complete. +- `since`: the compaction frontier at which the batch's times are observed. + When `since <= lower` the batch contents are exact. When `since` is in advance + of `lower`, a consumer must advance observed times by `since` before comparing + them (`trace/description.rs:50-56`). + +`lower`, `upper`, and `since` are the vocabulary for everything below. +Immutability is what makes sharing conceivable at all. A batch plus its +description is a self-contained, consistent fact that stays valid no matter what +the writer does next. + +## The arrange operator: one write path, two consumers + +`arrange_core` (`operators/arrange/arrangement.rs:352-520`) turns a stream of +updates into an arrangement. It feeds a `Batcher` and, as the input frontier +advances, seals batches whose bounds are the successive frontiers +(`arrangement.rs:361-374`). Each sealed batch goes to two places: + +1. Into the trace, via `writer.insert(batch.clone(), Some(cap_time))` + (`arrangement.rs:479`). +2. Onto the operator's output stream, via `output.session(...).give(batch)` + (`arrangement.rs:481-482`). + +These are the same batches. The stream doc states it plainly: "This stream +contains the same batches of updates the trace itself accepts" +(`arrangement.rs:48-49`). Downstream operators consume the stream for the +*incremental* batches they need (a join needs each side's deltas), and reach the +*accumulated* state through a trace handle. + +### Seal-only empty batches + +When the input frontier ticks but the batcher holds no data in advance of it, +the operator takes a different branch (`arrangement.rs:503-509`). It calls +`writer.seal(frontier)`, which builds an empty batch `[old_upper, new_upper)` +and inserts it into the trace but does **not** send it downstream. The comment +is explicit: an empty input batch with the new upper is fed "to the trace agent +(but not along the timely output)" (`arrangement.rs:449-452`). + +So an idle arrangement with a ticking frontier produces empty batches that enter +the trace (and, as the next section shows, the replay queues) but never travel +the output stream. This is why anything reconstructing the seal frontier must +read it from the trace, not from the stream. The stream lags the trace by these +empty batches, and by at least one progress round. + +## The spine: fueled, amortized, internal merging + +The trace is a `Spine` (`trace/implementations/spine_fueled.rs`). Inserting a +batch (`spine_fueled.rs:270-286`) appends to a pending list and calls +`consider_merges` (`spine_fueled.rs:399-437`), which places batches into +size-tiered layers and begins merges when a layer fills. + +Merging is *fueled*. `introduce_batch` grants `fuel = (8 << level) * effort` +(`spine_fueled.rs:474-477`) and applies it to every in-progress merge via +`apply_fuel` (`spine_fueled.rs:563-589`). A merge advances a little on each +insert rather than all at once, which is how the spine amortizes merge cost +against ingestion. The eight-units figure is justified in the source as four +units per real record plus four per virtual record from promoting smaller +batches (`spine_fueled.rs:468-473`). When a merge completes it cascades into the +next larger layer (`spine_fueled.rs:584-587`). + +The load-bearing fact for anything downstream: **merges are entirely internal to +the spine.** They are driven by `insert`, `exert`, and +`set_physical_compaction` → `consider_merges` (`spine_fueled.rs:205`). None of +them call `TraceWriter::insert`, and none of them touch the arrange operator's +output stream. A merge changes the physical layout of the trace. It produces no +event that leaves the spine. + +## Compaction: logical versus physical + +A trace exposes two independent compaction frontiers, and they mean different +things (`trace/mod.rs:82-129`). + +**Logical compaction** (`set_logical_compaction`) is permission to change update +times. From the trait doc (`trace/mod.rs:82-97`): + +> Logical compaction is the ability of the trace to change the times of the +> updates it contains. Update times may be changed as long as their comparison +> to all query times beyond the logical compaction frontier remains unchanged. +> Practically, this means that groups of timestamps not beyond the frontier can +> be coalesced into fewer representative times. +> By advancing the logical compaction frontier, the caller unblocks merging of +> otherwise equivalent updates, but loses the ability to observe historical +> detail that is not beyond `frontier`. + +**Physical compaction** (`set_physical_compaction`) is permission to merge +batches (`trace/mod.rs:107-121`): + +> Physical compaction is the ability of the trace to merge the batches of +> updates it maintains. Physical compaction does not change the updates or their +> timestamps, although it is also the moment at which logical compaction is most +> likely to happen. +> By advancing the physical compaction frontier, the caller unblocks the merging +> of batches of updates, but loses the ability to create a cursor through any +> frontier not beyond `frontier`. + +In the spine, `set_logical_compaction` only records the frontier +(`spine_fueled.rs:193-196`). `set_physical_compaction` records its frontier and +calls `consider_merges` (`spine_fueled.rs:200-206`). The actual coalescing +happens *when a merge begins*: `insert_at` passes the logical frontier into +`begin_merge` (`spine_fueled.rs:617-618`), the new batch's `since` becomes the +join of its inputs' `since` with that frontier (`ord_neu.rs:383-384`), and each +update time is advanced by the merged `since` before it is consolidated +(`ord_neu.rs:571-572`). Advancing `since` is what collapses distinct times at or +below it into equal representatives, so their diffs cancel or sum. + +Two consequences worth stating. Compaction only ever advances, and it is +irreversible: once times are coalesced the detail is gone. And logical +compaction does its actual work through physical merges. A trace whose physical +compaction is held back accumulates uncompacted detail even if its logical +frontier has advanced, because the merge that would coalesce has not run. + +## Reading a trace + +`TraceReader` (`trace/mod.rs`) is the read interface. Two methods matter for +sharing: + +- `batches_through(upper)` returns a clean cut of the batch chain: the non-empty + batches whose descriptions are covered by `upper`. It skips empty batches + (`spine_fueled.rs:139,146,155,166`) and panics if a non-empty batch straddles + the cut (`spine_fueled.rs:176-186`). The frontiers of the chain "form a total + order" (`spine_fueled.rs:116-117`). +- `map_batches(f)` visits the entire current batch chain, including empty + batches (`spine_fueled.rs:211-223`). It reflects merges: after the spine rolls + two batches into one, `map_batches` yields the merged batch. It is the + authoritative view of what the trace currently holds. + +A reader turns batches into a cursor. Because `map_batches` reflects merges, any +reader that re-reads the trace sees the current, compacted, merged state. A +reader that captured an older set of batch handles keeps those exact batches +alive and unchanged, because batches are immutable and reference counted. + +## Sharing within a worker: TraceAgent and TraceBox + +Within one worker, an arrangement is shared through `TraceAgent` +(`operators/arrange/agent.rs`), a handle onto a shared `TraceBox` +(`agent.rs:529-587`). This is the existing, single-thread analog of the +cross-thread sharing the design adds, and it is worth understanding as the model +being generalized. + +`TraceBox` wraps the trace plus two `MutableAntichain`s, one for logical and one +for physical holds, described as "accumulated holds on times for advancement" +and "for distinction" (`agent.rs:540-547`). A `MutableAntichain` counts +multiplicities of held times, and its `frontier()` is the meet, the greatest +lower bound, of all outstanding holds. Each live `TraceAgent` contributes its +own hold as a counted entry. + +Holds move through `adjust_logical_compaction` (`agent.rs:572-577`), which +replaces the old hold with the new one (add the new times, remove the old) and +pushes the resulting meet into the trace. Cloning an agent adds its holds +(`agent.rs:494-495`), dropping one removes them (`agent.rs:520-521`). The trace's +effective compaction frontier is therefore always the meet of holds across all +live handles. No single handle can compact the trace past another handle's hold. + +Notably, `TraceAgent::set_logical_compaction` does not require monotonic +advance. It joins the request with the agent's current frontier +(`agent.rs:43-50`) and moves forward with the consequence. The trace only ever +sees advancing frontiers even when a caller is careless. + +## Import and replay + +A downstream dataflow consumes an arrangement it did not build by *importing* +it. `import_core` (`agent.rs:270-318`) builds a timely `source` operator that +registers a *listener queue* against the trace and replays instructions from it. + +The queue speaks `TraceReplayInstruction`, a stream of `Batch(batch, hint)` and +`Frontier(frontier)` items. Two producers feed it: + +1. **Seeding.** When a listener registers, `new_listener` (`agent.rs:110-137`) + walks the current trace with `map_batches` and enqueues each existing batch + as `Batch(batch, Some(minimum))`, followed by one `Frontier(upper)` for the + last batch's upper. The hint for historical batches is the minimum time, not + each batch's lower. Crucially the seed comes from `map_batches`, so a new + importer starts from the *already merged and compacted* batch set, not from a + replay of history. + +2. **Ongoing inserts.** `TraceWriter::insert` (`writer.rs:50-82`) pushes, for + each newly sealed batch, the pair `Batch(batch, hint)` then + `Frontier(batch.upper())` to every registered queue (`writer.rs:66-75`), then + activates the listener. This runs once per new batch, including the empty + seal batches from an idle arrangement. The `hint` is "either `None` in the + case of an empty batch, or is `Some(time)` for a time less or equal to all + updates in the batch and which is suitable for use as a capability" + (`writer.rs:50-54`). + +The replay loop (`agent.rs:291-313`) drains the queue and translates it: +`Frontier(f)` downgrades the source's capability set, and `Batch(batch, hint)` +emits the batch under a capability *delayed to the hint* when the hint is +`Some` and the batch is non-empty. The hint being a lower bound, never the +batch's `upper`, is what keeps the imported frontier from advancing past updates +still inside a batch. Empty batches produce no output but their following +frontier still moves progress. + +There is no third producer. **Merges are never enqueued.** `TraceWriter` has no +merge entry point. Its only merge-adjacent method, `exert` (`writer.rs:43-48`), +drives spine merges but pushes nothing to any queue. An importer therefore never +receives a merge event. It receives new batches as they are sealed and rebuilds +equivalent state locally, and a *late* importer instead seeds from the current +merged chain. Both arrive at the same logical collection, differing only in +physical layout. + +## What any cross-thread sharing inherits from this baseline + +The design in [README.md](./README.md) generalizes `TraceAgent` and its listener +queue across a thread boundary. These status-quo properties are the ones it +must carry, work around, or consciously change. They are stated here as facts, +not decisions. + +- **Merges and compaction never leave the spine.** No event is produced when + the writer merges or coalesces. Readers that re-read the trace (via + `map_batches`) see the merged state. Readers replaying the insert stream do + not, and rebuild equivalent state on their own. Any sharing mechanism that + wants a remote reader to *benefit* from the writer's merges must route that + reader through a fresh read of the batch chain, not through the replay stream. + +- **The replay queue is producer and consumer coupled on one worker today.** + `insert` pushes to the queue and the same worker's scheduler drains it in the + same run of steps. The queue self-limits because production and consumption + advance together. Nothing in `writer.rs` or `agent.rs` bounds the queue + length. Decoupling the producer and consumer onto independently scheduled + runtimes removes the coupling that kept it bounded. + +- **Compaction is monotone and irreversible, and the empty meet is + destructive.** The trace only advances. A hold set that empties to the empty + frontier means "compact everything", which is unrecoverable. Hold accounting + that today lives in `TraceBox` as counted antichains has to be reconstructed + wherever the holds now live. + +- **Logical compaction realizes through physical merges.** Holding physical + compaction back keeps uncompacted detail resident even when the logical + frontier has advanced. A remote reader that holds physical compaction affects + the writer's ability to shrink its own footprint. + +- **Immutability plus reference counting is the whole basis for sharing.** A + batch handed out stays valid and unchanged for as long as a handle to it + lives, wherever that handle lives. The only thing pinning batches to the + writing thread is the reference-count wrapper. Today that is `Rc` + (`trace/mod.rs:327-439`, `rc_blanket_impls`). The Arc-batches work adds an + identical `arc_blanket_impls` (`trace/mod.rs:447-558`) whose sole difference + is `Arc` for `Rc`, so a batch whose contents are `Send + Sync` can be read + from any thread. diff --git a/doc/developer/design/20260720_two_runtime_compute/README.md b/doc/developer/design/20260720_two_runtime_compute/README.md new file mode 100644 index 0000000000000..7b1ffec9c569d --- /dev/null +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -0,0 +1,480 @@ +# Two-runtime compute: isolating reads from maintenance + +- Associated: + - [background.md](./background.md): how a compute replica serves maintenance + and reads today (the status-quo baseline this design departs from). + - [Shared arrangements across timely runtimes](../20260719_shared_arrangements_across_runtimes/README.md): + the differential-level primitive this design consumes. + - Implementation of the `Arc`-batches prerequisite: differential + TimelyDataflow/differential-dataflow#807, materialize + MaterializeInc/materialize#37743. File and line references to the `Arc` + spines below are to the state on those branches, not to `main`. + +## The Problem + +A compute replica does two very different kinds of work on the same timely +worker threads. It maintains indexes and materialized views, which is CPU-bound +and bursty (arrangement merges, operator work proportional to input volume). And +it serves reads: peeks against arrangements, and the short-lived dataflows +behind ad-hoc queries. + +These contend. An index peek runs its cursor walk synchronously on the worker +thread (`compute/src/compute_state.rs:982-988`, driven from the loop at +`compute/src/server.rs:417`). The worker only reaches peek servicing after it +returns from `step_or_park`, so a peek waits for the worker that owns the shard +to come back around its loop. That worker may be deep in a join or a large +merge, and timely operators run to completion. The result is read tail latency +proportional to how busy maintenance is, which is exactly when users are +watching. Persist peeks already sidestep this by offloading to an async task +(`compute_state.rs:1254-1325`). Index peeks, the common fast path, do not. + +The root cause is that reads and maintenance share worker threads, and an +arrangement is readable only from the worker that maintains it. The `Arc`-batches +work removes the second half of that restriction: arrangement batches become +`Arc`'d and their contents are asserted `Send + Sync` in `mz-row-spine` (added +by the Arc migration, #37743), so a batch can be read from another thread. This +design uses that to serve reads from a different set of threads than the ones +doing maintenance. + +## Success Criteria + +- A read is served by threads other than the maintenance worker threads. Its + latency floor is CPU contention with maintenance, not waiting for a + maintenance worker to return to its command loop. The read no longer queues + behind a long maintenance operator step. +- Correctness is unchanged. A read observes exactly the collection at its + timestamp, gated by the same `since <= timestamp < upper` window as today. +- The controller remains the single authority on compaction. No arrangement + compacts past a time a read still needs. A read never holds a maintenance + arrangement back below the controller's own read hold. A peek holds nothing (it + reads a point-in-time snapshot), and a longer-lived import holds only its own + `as_of`, released when it drops, which is the same shape as a read hold today. +- Maintenance throughput is not regressed. Publishing an arrangement for reading + costs the maintenance path only constant per-batch overhead, and publishing + alone never pins compaction. +- Generality is preserved. The read side imports an arrangement with an `as_of` + and replays its change stream, so a continuously maintained read (a subscribe) + can move to the interactive runtime later. The design does not bake in a + point-in-time-only mechanism that would foreclose that. + +Non-criterion, stated to head off a misreading: this does not make an individual +peek's cursor walk faster. It removes the walk, and the wait to be serviced, +from the maintenance worker's critical path. A tiny point-lookup that was +microseconds inline is still microseconds, now paid on an interactive thread +that was free to run it immediately instead of after the maintenance worker's +current step. + +## Out of Scope + +- Cross-process sharing. Everything here is threads within one replica process. + Cross-process is persist's job. +- Changing the fast-path versus slow-path peek decision, which lives in the + coordinator (`adapter/src/coord/peek.rs`). This design changes only where the + read executes on the replica, not how the coordinator plans it. +- Core-sharing policy between the two runtimes (pinning, priorities, + oversubscription). Worker counts are fixed by this design. How the resulting + threads share cores is deferred to measurement (see open questions). +- Multi-replica or cross-replica concerns. This is about one replica's internal + structure. + +## Architecture + +One replica process runs two compute timely runtimes, two `serve`/`ClusterSpec` +instances (`compute/src/server.rs:85-131`) sharing per-process resources exactly +as storage and compute already do today (`clusterd/src/lib.rs`, `background.md`): + +- The **maintenance runtime** owns the durable, maintained work: indexes, + materialized views, and subscribes. It renders and maintains their + arrangements as today, and additionally *publishes* each index arrangement so + other threads can read it. +- The **interactive runtime** owns the ephemeral, read-oriented work: peeks and + temporary dataflows. It reads maintenance arrangements through the published + handles, never touching a maintenance worker thread. + +Reads are served on the interactive side by threads that hold the published +arrangement directly and only contend for CPU with maintenance. That is the +whole point: a peek does not wait for a maintenance worker to come back around +its loop, because the thread serving it is not a maintenance worker. + +The shared trace supports two read modes, and the design uses both. A **snapshot** +mode serves a point-in-time read (a peek) by handing back a consistent cursor +over the current batch chain, held alive by its `Arc`s. An **import** mode +replays the arrangement's change stream from an `as_of` into a dataflow, exactly +as a same-worker `import_index` does today, and is the general path that a future +subscribe migration would use. Peeks take the snapshot mode because it is +pin-free and needs no dataflow. Temporary dataflows take the import mode. + +### Serving a peek + +A fast-path index peek is served by taking a consistent snapshot of the +published arrangement and running the existing cursor walk against it, on an +interactive thread: + +1. The routing layer delivers the `Peek` to the interactive side without going + through a maintenance worker's command loop. +2. The interactive side looks up the published handle for the target + `GlobalId`, gates on the published `upper`/`since`, and takes a `Send` + snapshot: a clone of the current batch chain (a handful of `Arc` clones) plus + the frontiers. +3. It runs `PeekResultIterator` (`compute/src/compute_state/peek_result_iterator.rs`) + over the snapshot and produces the `PeekResponse`. + +The snapshot's `Arc`'d batches are immutable, so once taken, the maintenance +runtime may merge and compact freely while the walk proceeds. No lock is held +across the walk, and there is no torn read. + +Whether the walk runs on the interactive runtime's timely workers or on a +dedicated reader pool alongside it is an implementation choice. A dedicated pool +isolates peeks even from the interactive runtime's own temporary-dataflow +rendering, which matches the "a thread that just grabs the trace and runs the +peek" model most directly. Either way the walk is off the maintenance threads. + +### Serving a temporary dataflow + +A slow-path peek or other ad-hoc query is rendered as a temporary dataflow on +the interactive runtime. It imports the maintenance arrangements it needs as +ordinary `Arranged` collections through the import mode, replaying the change +stream from the dataflow's `as_of`, then runs joins, reduces, and the rest with +no maintenance-thread involvement. Its result is peeked and the dataflow is +dropped, as transient dataflows are today. + +The import registers a real read hold at its `as_of` on the shared trace, which +is released when the dataflow drops. This is deliberate, not a wart. It is the +same read-hold shape a same-worker import has today, and it is what lets a +continuously maintained read hold its `as_of` for as long as it runs. It is the +reason the design keeps the change-stream import rather than a point-in-time-only +snapshot: a subscribe cannot be served from a frozen snapshot, it needs the +ongoing stream, so foreclosing the import mode would foreclose ever moving +subscribes here. The hold is bounded and safe for the same reason a peek's is +(next section). + +## What decouples, and what does not + +The latency win is real and comes from the serving thread not being a +maintenance worker. A peek at a timestamp at or below the arrangement's +published `upper`, which is the common case, is served immediately, bounded only +by CPU availability. It does not queue behind a maintenance operator step. + +The one residual dependency is frontier freshness, and it is not new in kind. A +peek at a timestamp *beyond* the published `upper` must wait for the maintenance +runtime to publish a newer `upper`. Publication happens on the maintenance +worker's activation and is cheap (a chain refresh plus a frontier update, no +per-record work), so it advances whenever the worker steps, independently of +whether the worker is free to do a full walk. This is the same kind of wait a +read at the frontier edge already incurs today (waiting for the trace `upper` to +advance), not a wait for the maintenance worker to be free to *serve* the read. +So fresh-edge reads are gated on maintenance frontier progress as before, and +everything at or behind the published frontier is fully decoupled. + +## Memory + +A snapshot pins the specific batches it captured until the read finishes. While +it is held, the maintenance runtime may merge those batches into a new one, so +the pre-merge inputs (pinned by the reader) and the merged output (in the live +trace) coexist for the read's duration. This is bounded by the arrangement size +and is transient, and it is the intended cost of letting a reader hold a +consistent view while the writer compacts. + +It is not a new doubling. A fueled merge already holds both its input and output +batches in memory while the merge is in progress, independent of any reader. A +held snapshot extends the lifetime of its captured batches, it does not create a +second copy of the whole arrangement. + +## Routing and response merging + +The controller sees one replica behind one protocol endpoint. A **new +process-level multiplexer** sits between the controller connection and the two +runtimes. It is a genuinely new component, not a reuse of the existing +`PartitionedComputeState` (`compute-client/src/service.rs:79`), which merges +*homogeneous* partitions (all workers of one runtime, waiting for every part and +taking a meet). The two runtimes are *heterogeneous*: each collection id lives on +exactly one runtime, and a peek is answered by exactly one runtime. The +multiplexer therefore does ownership-based demux and pass-through, not a meet or +an all-parts union. + +**Routing by target id.** The multiplexer dispatches on the id a command +targets: + +- `Peek`, `CancelPeek`, and `CreateDataflow` for a temporary dataflow go to the + interactive side. A temporary dataflow exports only `GlobalId::Transient` + (`compute-types/src/dataflows.rs:380-383`). +- `CreateDataflow` for maintained work (indexes, MVs, subscribes), `Schedule`, + `AllowWrites`, and `AllowCompaction` for a maintained id go to the maintenance + runtime. +- `AllowCompaction` (including the empty-frontier drop) for a transient id goes + to the interactive side that owns that temporary dataflow. +- Lifecycle commands (`CreateInstance`, `UpdateConfiguration`, + `InitializationComplete`) go to both. + +Note that routing is not literally "by the target id" for every command. A +fast-path `Peek` names a maintenance-owned index yet must be served on the +interactive side, and `CancelPeek` keys by `uuid`. So the multiplexer learns +transient-id ownership by observing `CreateDataflow`, and treats peeks as +interactive by command type. The one subtlety is that subscribes also export +transient ids but must stay on the maintenance runtime, since they are +continuously maintained rather than read once. A subscribe is distinguished by +carrying a subscribe sink in its `DataflowDescription`. The predicate is a +bespoke policy, "maintained work to maintenance, one-shot reads to interactive", +not a one-line id check. + +**Response merging.** The multiplexer merges both runtimes' responses into the +one stream the controller expects. `PeekResponse`s originate on the interactive +side, `Frontiers` for maintained collections on the maintenance runtime. Its +combining contract is its own (pass a peek response through from whichever side +produced it, report maintained frontiers from maintenance), which is why it +cannot simply reuse `PartitionedComputeState`. + +## Where the sharing hooks in + +**Mechanism: a publication point on the arrange stream, not an `Arc`-native +trace.** The tempting move is to make the trace itself cross-thread, an +`Arc>` trace box the interactive runtime imports from directly. +Rejected. Differential's `TraceAgent`/`TraceBox` and its whole import path are +`Rc`-based (`Rc>`, `Rc` listener queues, `Activator`), so an +`Arc`-native trace is an invasive change to differential's core handle, and it +puts a lock on the *maintenance worker's own* hot read path, which every local +dataflow import and every compaction call would then pay. Instead, the local +trace stays a lock-free `Rc` `TraceAgent`, and a separate publication point +carries the `Arc` batches, frontiers, importer queues, and holds across the +boundary. Only cross-thread readers pay the lock. This is additive and leaves +the owning worker's path untouched. + +**Maintenance insertion site: `export_index`** (`compute/src/render.rs:688`). +The `ArrangementFlavor::Local` arm has the `Arranged` in hand, with both its +change stream (`oks.stream`/`errs.stream`) and its trace, right before +`traces.set(idx_id, TraceBundle::new(oks.trace, errs.trace))` +(`render.rs:713-737`). That is where publication attaches: it sinks the arrange +output stream (the change stream that mirrors `TraceWriter::insert`) into a +publication point and records a `Send` handle in a per-process registry. The +`ArrangementFlavor::Trace(gid, ..)` re-export arm (`render.rs:739-743`), which +today just re-registers the existing trace under a new id, re-registers the same +published handle. `export_index_iterative` mirrors this. Publishing does not +change what `traces.set` stores, so the local trace and its compaction are +exactly as today. + +**The registry.** A new first-class per-process object keyed by `GlobalId`, one +entry per worker ordinal, shared into both runtimes the way the persist client +cache is shared today (created once, handed to both `serve` calls, +`compute_state.rs:114-116`). Interactive worker `i` finds maintenance worker +`i`'s handle. + +**Interactive insertion sites.** Two read entry points consume the registry. +Peeks: a lookup that takes a snapshot and runs the cursor walk, replacing the +`handle_peek` path's local `TraceManager` lookup (`compute_state.rs:673-699`) +with a registry lookup for the target id. Temporary dataflows: a new import path +analogous to `import_index` (`render.rs:591`) that sources the imported +`Arranged` from the registry handle (import mode) instead of the local +`TraceManager`. + +Publication cost is one `Arc` clone plus a mutex push per batch. + +The two runtimes step independently, so there is no happens-before between +"maintenance renders and publishes index X" and "interactive is asked to read +X". In steady state the controller only issues a peek after a `Frontiers` +response proves the index rendered, which implies it published. But on +reconciliation or replica restart the controller replays `CreateDataflow` and +`Peek` from history, and a peek can reach the interactive side before the +maintenance runtime has re-rendered and re-published. The design therefore +requires an explicit readiness rule: a read for an unregistered `GlobalId` +**blocks** until the publication point appears, woken when it registers, rather +than erroring or reading an empty snapshot as if valid. This block-until- +published handshake is new machinery the primitive does not provide today. + +## Compaction: controller authority and required primitive changes + +The controller stays the sole compaction authority. The maintenance runtime +drives an arrangement's `since` from `AllowCompaction` exactly as today, derived +from the controller's read holds (`instance.rs:1922-1981`). The read side must +never pin a maintenance arrangement back below that frontier. + +Correctness rests on three things, and only the first exists today: + +1. The controller already holds a maintenance arrangement back for as long as a + read needs it. A peek carries a read hold at its timestamp, and a temporary + dataflow is created behind a read hold at its `as_of`. Command reordering + across the two runtimes cannot defeat this, because the controller gates the + *emission* of `AllowCompaction`, not its delivery: it never emits a frontier + past a time it is still holding. So while the read is outstanding, the + maintenance `since` cannot advance past the read timestamp. +2. The batches a read observes are immutable. A snapshot peek pins its captured + batches by `Arc`, and an import replays immutable batches into its dataflow, + so in either mode maintenance may merge and compact concurrently without + affecting an in-progress read. +3. A `since <= timestamp` gate on the read path, so a read at a time the + arrangement has already compacted past returns the same error it does today + (`compute_state.rs:1536-1544`) rather than a silently coalesced result. + +The two read modes differ in how they hold, and both are safe: + +- A **snapshot peek holds nothing on the trace.** Its `Arc`'d batches are its + only hold, and they defer only the *frees*, never maintenance's compaction + *decisions*. It cannot pin maintenance no matter how slow it is. +- An **import holds a real read hold at its `as_of`**, released when the dataflow + drops. This is intended. It is the same read-hold shape a same-worker import + has today, and it is what a subscribe needs to keep its `as_of` alive while it + runs. It cannot pin maintenance *below the controller's own hold*, because the + controller holds the same `as_of` for as long as the read exists (criterion 1). + A wedged importer that never drops is no worse than a stuck read today: the + read it serves never completes, so the controller's hold is stuck at the same + frontier regardless. Dropping the import is how the hold is freed, which is + exactly the "droppable to free the read hold" property we want. + +**Two changes to the shared-arrangements primitive are required, and the +primitive does not do them yet.** Both were found by adversarial review of +`sharing.rs` and are gating work for this design: + +- **The publisher must not pin compaction.** As written, the publisher holds a + `TraceAgent` clone whose logical/physical hold is sourced from its own + `get_logical_compaction` and never advances (`sharing.rs:412,462`), so merely + publishing an index freezes its compaction at publish-time `since` for the life + of the index, even with no readers. Publishing must not carry an independent + compaction floor. The maintenance handle and controller drive `since`, and only + a live import's own `as_of` hold (released on drop) may hold the trace back. +- **`snapshot_at` must enforce the `since <= timestamp` gate** (criterion 3). It + currently waits only for `upper` to pass the time and returns whatever `since` + the snapshot has (`sharing.rs:185-203`), so it could serve stale rows once + compaction actually moves. The gate must be added on the read path. + +Note the earlier framing that "reads register no hold" was too strong. That is +true and desirable for the snapshot peek path, but an import legitimately holds +its `as_of`. The property the design actually needs is the weaker and correct +one: publishing pins nothing, and every hold that does exist is either the +controller's or a live import's own `as_of`, both released on completion. + +## Internal to compute, one protocol endpoint + +The split is invisible to the controller, by choice. Teaching the controller +about two runtimes (an explicit sub-runtime axis in `ReplicaState`, +`instance.rs:3082-3097`) would enable smarter placement and per-runtime +introspection, but it is a real protocol and controller change and is not needed +for the mechanism to work. The cost of staying internal is that peek routing and +per-runtime memory attribution are not visible to the controller, which +introspection will eventually want (see open questions). + +## Equal peers is a hard requirement + +Co-location of an imported arrangement with interactive-side operators relies on +both runtimes routing a key to the same worker ordinal, which holds only if they +have the same total peer count. Peer count is `workers_per_process * +num_processes` (`compute/src/extensions/arrange.rs`), so both factors must match +between the two compute runtimes, not just workers-per-process. This is stronger +than the storage-versus-compute worker-count assertion that exists today +(`clusterd/src/lib.rs:426-429`), which does not cover it. The design requires a +hard assertion of equal peers at replica configuration, and the primitive's +`import` should assert matching peers at import time (it does not today, contrary +to an earlier claim in the primitive design). + +## Resource sharing and the single-instance assumption + +Two compute runtimes in one process share the per-process resources the +storage/compute split already shares: the persist client cache, tracing handle, +and metrics registry (`compute_state.rs:114-116`). Three process-global +assumptions must be reconciled, since they are written for "a replica process +hosts a single instance" (`compute_state.rs:495-496`): + +- `mz_row_spine::DICTIONARY_COMPRESSION` (`compute_state.rs:497`): a process + global both runtimes would write. They must agree, or it must move to + per-runtime state. +- lgalloc and pager configuration (`compute_state.rs:255-361`): initialized once + per process. The second runtime must not re-initialize it. +- Metrics (`compute_state.rs:538-540`): already shared with storage, so the + pattern exists, but per-runtime labels are needed to tell maintenance and + interactive metrics apart. + +## Failure model + +The two runtimes share fate, exactly as the replica process behaves today. They +live in one process, so a fatal error in either brings the process down and the +replica restarts as a whole. There is no partial-failure path, no fallback that +reroutes interactive reads onto the maintenance runtime, and no independent +restart of one runtime. A read in flight when the process dies fails with the +process and is retried by the controller against the restarted replica, which is +the existing behavior. + +Shared fate is a requirement, not just a convenience. It is what makes an +import's read hold safe without a lease-expiry mechanism. A wedged importer can +hold maintenance compaction back only as long as its process lives, and shared +fate bounds that to the life of the whole replica, which is exactly the bound a +stuck read has today. For it to hold, a panic on any worker or reader thread of +either runtime must abort the whole process. The implementation must confirm the +panic path is abort-equivalent for both runtimes. If a single thread could panic +without taking the process down, a wedged importer's hold and a half-served read +would leak, and that fallback path would need its own correctness argument. + +## New components this design requires + +Called out explicitly, because the mechanism does not reduce to existing objects: + +- The process-level command/response multiplexer over two runtimes (not + `PartitionedComputeState`). +- The per-process publication registry keyed by `GlobalId`, plus the + block-until-published handshake. +- Two interactive read entry points off the registry: a peek path that snapshots + and walks (a new `PendingPeek` variant or reader-pool equivalent, replacing the + local `TraceManager` lookup in `handle_peek`, `compute_state.rs:673-699`), and a + temporary-dataflow import path analogous to `import_index` (`render.rs:591`) + that imports the change stream from the registry handle. +- The two primitive changes in "Compaction": publisher-does-not-pin and the + `snapshot_at` since gate. Plus the `import` equal-peers assertion and the + `batches_through` straddle fix (open questions). +- A compile-time `Send` assertion over the peek walk. `peek_stash.rs:44-48` + documents that `PeekResultIterator` is `!Send` today because the trace reader + is `Rc`. The `Arc` migration should flip it, but the design depends on it, so + it must be asserted, including the batch cursor and the columnation containers, + not just the batch wrapper. + +## Open questions + +- **Core sharing between runtimes.** Worker counts are fixed (equal peers). How + the resulting threads share cores, and whether oversubscription is acceptable + given the interactive side is often blocked on reads, is deferred to + measurement. +- **Import queue backpressure.** The change-stream replay queue is bounded today + only because producer and consumer share a worker step. Across runtimes they + are independently scheduled, so a lagging interactive importer can grow the + queue without bound. A bound plus publisher backpressure is likely needed. This + is the price of keeping the general change-stream import (peeks avoid it by + using the snapshot mode, which has no queue). It is a liveness and memory + concern more than a compaction one: a lagging importer still holds only its own + `as_of`, bounded by the controller's hold and by shared fate, so it falls + behind on its own updates rather than compacting-wedging maintenance. It grows + in importance if long-lived imports (subscribes) move here. Captured in the + primitive design. +- **Introspection and memory attribution.** Under one endpoint the controller + cannot see which runtime holds what memory, and a shared arrangement reachable + from both runtimes' arrangement-size loggers can be double-counted or + misattributed depending on drop order. Per-replica memory relations + (`mz_arrangement_sizes` and friends) need a per-runtime view eventually. +- **Cancellation and exactly-one response.** With the walk off the maintenance + thread, a `CancelPeek` can race a completing read. The merge point on the + interactive side must guarantee exactly one `PeekResponse`, including for + point-lookups that produce no intermediate batch at which to observe the + cancel. +- **`batches_through` straddle handling.** The primitive's `batches_through` + includes a batch that straddles the requested cut rather than panicking as the + spine does (`sharing.rs:258-276`). Confirm every `cursor_through` frontier that + reaches a `SharedTraceHandle` is batch-aligned, or restore a fail-stop, so an + import cannot silently return updates past the cut. + +## Alternatives + +- **Priority scheduling inside one runtime.** Yield more finely so peeks + interleave with maintenance. Timely scheduling is cooperative per worker, so a + peek behind a long operator step cannot preempt it, and this couples + maintenance throughput to read latency permanently. It also does nothing for + temporary-dataflow rendering. +- **Offload only the walk, keeping the snapshot on the maintenance worker.** A + smaller change: the maintenance worker gates and snapshots on its own loop, + then hands the walk to a pool. Rejected, because the worker must still come + around its loop to take the snapshot, so the read still queues behind a long + maintenance step. It removes the walk cost but not the wait to be serviced, + which is the dominant tail. Serving the read from an interactive thread that + holds the published handle avoids both. +- **Serve all peeks from Persist.** Peeks can read persist shards directly + (`PeekTarget::Persist`), bypassing arrangements. This exists for some peeks, + but it loses the in-memory arrangement's latency and freshness and does not + help index-backed reads that must reflect the maintained arrangement. +- **A second process instead of a second runtime.** Isolates reads fully but + cannot share arrangements in memory, so it copies data across the process + boundary, which is what persist already offers. In-process sharing is the + point. diff --git a/doc/developer/design/20260720_two_runtime_compute/background.md b/doc/developer/design/20260720_two_runtime_compute/background.md new file mode 100644 index 0000000000000..9eab9757b0a2b --- /dev/null +++ b/doc/developer/design/20260720_two_runtime_compute/background.md @@ -0,0 +1,291 @@ +# Background: how a compute replica serves maintenance and reads today + +This is a status-quo companion to the [two-runtime compute design](./README.md). +It records how a Materialize compute replica is structured today, how it stores +and compacts arrangements, how it serves peeks, and how the controller drives +it. The design proposes splitting maintenance and reads across two timely +runtimes. This document is the baseline it departs from. + +Nothing here is a proposal. References point at `/home/user/materialize/src` at +the time of writing. Line numbers drift, so treat them as pointers to the right +function. + +## A clusterd process already runs two timely runtimes + +The most important status-quo fact for this design: a `clusterd` process today +already hosts two independent timely runtimes, one for storage and one for +compute. `clusterd::run` starts both, `mz_storage::serve` and +`mz_compute::server::serve`, from one process (`clusterd/src/lib.rs:441`, +`:471`), each with its own `TimelyConfig`. There is an explicit +`// TODO: unify storage and compute servers to use one timely cluster.` +(`clusterd/src/lib.rs:502`). + +So "two timely runtimes in one process, sharing per-process resources" is not +new infrastructure. What is new in this design is that both runtimes would be +*compute* and would *share arrangements*, which storage and compute do not. + +Two invariants the co-resident runtimes already rely on carry directly into this +design: + +- **Equal worker counts are asserted across runtimes.** `clusterd` refuses to + start unless storage and compute have equal workers per process + (`clusterd/src/lib.rs:426-429`, "storage and compute must have equal + workers-per-process"). This is the precondition a pairwise per-worker + arrangement import needs. +- **Key routing is a deterministic `hash % peers`.** The arrangement exchange + routes by `key.hashed()` (`compute/src/extensions/arrange.rs:133`), so a given + key lands on the same worker ordinal in any runtime with the same worker + count. This is what makes worker-local, non-networked pairwise import feasible. + +### The lifecycle seam + +Timely threads are spawned in exactly one place, `ClusterSpec::build_cluster` +(`cluster/src/client.rs:182-279`), which calls timely's `execute_from` +(`client.rs:260`) and stores the `WorkerGuards` in a `TimelyContainer` +(`client.rs:50-61`). Compute's `serve` (`compute/src/server.rs:85-131`) builds a +`Config` that `impl ClusterSpec` and calls `build_cluster` once +(`server.rs:121`), producing one compute runtime. + +Resources group into three scopes, which is the map for where a second runtime +would attach: + +- **Per process (shared today):** `Arc` (created once in + `clusterd/src/lib.rs:394`, passed to both servers, "intentionally shared + between workers" `compute_state.rs:114-116`), `TxnsContext`, + `ConnectionContext`, `Arc`, `MetricsRegistry`. Also several + process-global statics that assume a single compute instance: + `mz_row_spine::DICTIONARY_COMPRESSION` (`compute_state.rs:497`), lgalloc and + pager config (`compute_state.rs:255-361`), documented as "a replica process + hosts a single instance" (`compute_state.rs:495-496`). +- **Per runtime (one per `ClusterSpec`/`execute_from`):** the `TimelyContainer` + (`cluster/src/client.rs:50-61`), the networking mesh from + `initialize_networking` (`cluster/src/communication.rs:100`), and the + `WorkerConfig`. +- **Per worker (per thread):** `ComputeState` including the `TraceManager` + (`compute_state.rs:86`, `:97`), the command channel, and the `&mut + TimelyWorker`. + +## Per-worker state and the command loop + +Each worker owns a `ComputeState` (`compute/src/compute_state.rs:86`, +"Worker-local state that is maintained across dataflows"), constructed lazily on +the first `CreateInstance` command (`compute/src/server.rs:431-443`). It holds +the collections, the `traces: TraceManager` (`compute_state.rs:97`), the +`pending_peeks` map (`compute_state.rs:108-109`), and the command history. + +The worker loop is `Worker::run_client` (`compute/src/server.rs:372-422`). Each +iteration runs periodic maintenance (`traces.maintenance()`, +`report_frontiers()`, `server.rs:394-400`), steps timely +(`step_or_park`, `server.rs:411`), applies pending commands +(`handle_pending_commands`, `server.rs:414`), then services reads: + +```rust +compute_state.process_peeks(); +compute_state.process_subscribes(); +compute_state.process_copy_tos(); +``` +(`server.rs:416-420`). Command dispatch is +`ActiveComputeState::handle_compute_command` +(`compute_state.rs:446-478`), a match over `ComputeCommand` variants. + +Commands reach workers through a dataflow-based command channel: the controller +sends only to worker 0, which broadcasts over the timely fabric so every worker +sees the same ordered stream (`compute/src/command_channel.rs:10-23`). +`CreateDataflow` is split per worker by `partition_among` +(`command_channel.rs:156-199`), everything else is replicated +(`command_channel.rs:201-204`). Each runtime has its own worker 0. + +## Arrangements: storage, rendering, import, compaction + +### Storage + +Arrangements live in `TraceManager`, a `BTreeMap` +(`compute/src/arrangement/manager.rs:33-36`), held on `ComputeState` +(`compute_state.rs:97`). A `TraceBundle` (`manager.rs:234-239`) bundles an ok +trace, an err trace, and lifetime tokens: + +```rust +pub struct TraceBundle { + oks: PaddedTrace>, + errs: PaddedTrace>, + to_drop: Option>, +} +``` + +The handle type is a `TraceAgent` over the spine (`RowRowAgent = TraceAgent< +RowRowSpine>`, `typedefs.rs:98`). The ok arrangements are backed by +`RowRowSpine` and errors by `ErrSpine`. On `main` these spines wrap batches in +`Rc` (`row-spine/src/lib.rs:51`, `typedefs.rs:107`). The `Arc`-batches migration +(#37743) changes the wrapper to `Arc` and adds `Send + Sync` assertions over the +batch contents. Either way the coupling that keeps a trace single-runtime is +*above* the batch layer: the `TraceAgent` handle and the `to_drop: Rc` +token (`manager.rs:238`) are `Rc`-based and `!Send`. + +### Rendering and registration + +`CollectionBundle::arrange_collection` arranges the ok stream into a +`RowRowSpine` (`compute/src/render/context.rs:1129-1212`) and the err stream into +an `ErrSpine` (`context.rs:1104-1111`). `Context::export_index` +(`compute/src/render.rs:688-758`) registers the resulting handles into the trace +manager: + +```rust +compute_state.traces.set(idx_id, TraceBundle::new(oks.trace, errs.trace).with_drop(needed_tokens)); +``` +(`render.rs:734-737`). + +### Import (the single-runtime analog of this design) + +A new dataflow that reads an existing index imports it worker-locally. +`Context::import_index` (`render.rs:591-682`) looks the trace up in the same +worker's `TraceManager` (`render.rs:603`), asserts the import is legal against +compaction ("Index has been allowed to compact beyond the dataflow as_of", +`render.rs:604-607`), and calls `import_frontier_core` to bring the trace into +the new dataflow scope restricted to `[as_of, until)` (`render.rs:611-625`). No +exchange happens. This is a worker-local trace-handle handoff. It is the exact +operation a second runtime needs a cross-runtime analog for. + +### Compaction and read holds + +Two frontiers per trace: + +- **Logical compaction** (the `since`) is driven by the controller. + `AllowCompaction` (`compute_state.rs:466`) calls + `TraceManager::allow_compaction` (`manager.rs:79-84`), which sets + `set_logical_compaction` on both traces. An empty frontier means drop the + collection (`compute_state.rs:662-671`). +- **Physical compaction** is driven by periodic maintenance: + `TraceManager::maintenance` sets `set_physical_compaction(upper)` on each trace + (`manager.rs:54-71`), called from the worker loop (`server.rs:396`). + +There is no separate read-hold object on the replica. The read hold *is* the +trace handle plus its logical-compaction frontier. A peek pins a hold by cloning +the `TraceBundle` and calling `set_logical_compaction(timestamp)` on it +(`compute_state.rs:1231-1252`). The cloned handle keeps data readable until the +peek finishes and drops. The authority for how far a trace may compact lives in +the controller (next section), which sends `AllowCompaction`. + +## The peek path + +Where the fast-path versus slow-path decision happens: in the adapter or +coordinator, not in compute. `PeekPlan` is `FastPath` or `SlowPath` +(`adapter/src/coord/peek.rs:442-446`). A fast-path `PeekExisting` reads an +existing arrangement (`peek.rs:799-820`). A slow-path plan ships a transient +index dataflow first via `create_dataflow`, then peeks it and drops it +(`peek.rs:855-926`, teardown at `:1002-1004`). Both paths send compute the *same* +`Peek` command with `PeekTarget::Index { id }` (`compute-client/src/protocol/ +command.rs:411-425`, `:449-474`). + +On the replica, `Peek` is handled by `handle_peek` (`compute_state.rs:673-699`), +which clones the `TraceBundle` out of the manager (`compute_state.rs:678`), +builds a `PendingPeek::Index`, and calls `process_peek`. + +The actual read. `IndexPeek::seek_fulfillment` (`compute_state.rs:1504-1544`) +first gates on frontiers, then walks the trace. The gate contract +(`compute_state.rs:1505-1516`): + +> To produce output at `peek.timestamp`, we must be certain that it is no longer +> changing. A trace guarantees that all future changes will be greater than or +> equal to an element of `upper`. If an element of `upper` is less or equal to +> `peek.timestamp`, then there can be further updates that would change the +> output. + +So the peek is not ready until `upper` passes `timestamp` +(`compute_state.rs:1527-1534`), and it errors if `since` is beyond `timestamp` +(`compute_state.rs:1536-1544`). The serving window is `since <= timestamp < +upper`. The cursor walk is `PeekResultIterator` +(`compute/src/compute_state/peek_result_iterator.rs`), which walks a `CursorList` +over the trace's batch cursors and accumulates diffs only for times +`<= peek_timestamp` (`peek_result_iterator.rs:257-262`). Literal constraints use +`cursor.seek_key` for point lookups (`peek_result_iterator.rs:91-105`). + +**Peek reads run on the timely worker thread.** `process_peeks` +(`compute_state.rs:1058-1065`) is called from the worker loop +(`server.rs:417`) and calls `seek_fulfillment` inline (`compute_state.rs:982-988`), +so a large cursor walk blocks the timely step loop and contends with +maintenance. The one exception is the Persist peek (`PendingPeek::persist`, +`compute_state.rs:1254-1325`), which spawns an async task and re-activates the +worker via a `SyncActivator` when done. Index arrangement reads do not offload. +This offload is the precedent the design's stage 1 builds on. + +## Ephemeral dataflows + +There is no compute-layer "ephemeral query" concept. A slow-path peek's one-off +dataflow is an ordinary `CreateDataflow` whose exports are all transient. +`DataflowDescription::is_transient()` is "all export ids are transient" +(`compute-types/src/dataflows.rs:380-383`). A maintained index or MV exports a +persistent `GlobalId::User`, kept alive by controller read holds. A transient +peek dataflow exports only `GlobalId::Transient` and is created then dropped +around one peek. The only rendering consequence is that compute-event logging is +skipped for transient dataflows (`render/context.rs:116-123`). Subscribes are the +other transient class, maintained continuously rather than read once. + +## The controller protocol + +The protocol is documented at `compute-client/src/protocol.rs:10-135` +(asynchronous, in-order, three stages). `ComputeCommand` +(`compute-client/src/protocol/command.rs:38`) carries the lifecycle: +`CreateInstance`, `CreateDataflow` (`command.rs:144`), `Schedule` +(`command.rs:154`), `AllowCompaction { id, frontier }` (`command.rs:209-214`), +`Peek` (`command.rs:244`), `CancelPeek` (`command.rs:259`). `ComputeResponse` +(`compute-client/src/protocol/response.rs:29`) carries `Frontiers` +(`response.rs:61`), `PeekResponse` (`response.rs:79`, exactly one per peek), +`SubscribeResponse`, and `CopyToResponse`. + +### Replica addressing + +The controller talks to a replica as one logical client, internally partitioned +across the replica's processes. Each replica has one async `ReplicaTask` +(`compute-client/src/controller/replica.rs:159`) and a `Partitioned` client with +one partition per control address (process) (`replica.rs:42`, `:184-191`). The +partitioning state `PartitionedComputeState` (`compute-client/src/service.rs:79`) +exists both on the controller (dispatching between processes) and inside each +process (dispatching between workers) (`service.rs:66-70`). `split_command` +broadcasts `Hello`/`UpdateConfiguration` to all parts and sends everything else +to part 0 (`service.rs:365-382`). `absorb_response` merges: frontiers by meet +across parts (`service.rs:179`, `:417`), peek responses by union once all parts +report (`service.rs:219`). + +**There is no protocol-level notion of two runtimes or sub-replicas inside one +process.** The controller's model of a replica is `ReplicaState` +(`instance.rs:3082-3097`): one client, one config, one epoch, a flat +`BTreeMap`. The only multiplexing it knows is +processes (partition axis) and, below that, workers (in-process command channel). + +### Frontier reporting and compaction control + +The replica reports frontiers from `ComputeState::report_frontiers` +(`compute_state.rs:825-918`): write frontier from `traces.oks_mut().read_upper` +(`:842-847`), emitted as `ComputeResponse::Frontiers`. The controller consumes +them in `handle_frontiers_response` (`instance.rs:2042-2080`), advancing the +global write frontier and relaxing read holds per read policy +(`maybe_update_global_write_frontier`, `instance.rs:1870-1919`). + +Compaction is decided by controller read holds, not by the replica. +`apply_read_hold_change` (`instance.rs:1922-1981`) recomputes `new_since` from +the capability frontier, checks it does not regress, propagates it to compute and +storage dependencies, and emits `ComputeCommand::AllowCompaction { id, frontier: +new_since }`. The `since <= upper` relationship holds because the since is +derived from the write frontier through the read policy +(`instance.rs:1840-1860`). A peek's read hold is what keeps `AllowCompaction` +from advancing `since` past the peek timestamp, which is exactly the guarantee +the peek's frontier gate relies on. + +## Load-bearing facts this design inherits + +- Two timely runtimes per process is already how `clusterd` works, with equal + worker counts asserted and `hash % peers` routing. The preconditions for + pairwise import exist. +- The arrangement-sharing boundary is `ComputeState::traces` / `TraceManager`, + accessed worker-locally today via `export_index` / `import_index`. Batches are + already `Arc` and `Send + Sync`. The `TraceAgent` handle and `to_drop` token + are still `Rc`. +- Index peek reads run on the worker thread and contend with maintenance. + Persist peeks already offload to an async task, which is the pattern a + worker-offloaded index peek would follow. +- The controller drives compaction through its own read holds, addresses a + replica at process granularity, and has no notion of sub-runtimes. A split can + be made transparent to it, or modeled explicitly, and the compaction authority + is the controller either way. +- Several process-global statics assume one compute instance per process. Two + compute runtimes in one process must reconcile them.