From cf0da2af3a83fa0407fd069e0b62fd98cdbc091c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 14:02:19 +0000 Subject: [PATCH 1/6] design: shared arrangements across timely runtimes The differential-level primitive design (README.md) plus a status-quo background (background.md) describing how a single-runtime arrangement works today. Moved out of the Arc-migration implementation branch so design review is separate from the code change. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../README.md | 464 ++++++++++++++++++ .../background.md | 269 ++++++++++ 2 files changed, 733 insertions(+) create mode 100644 doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md create mode 100644 doc/developer/design/20260719_shared_arrangements_across_runtimes/background.md 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..66a8f29dbd124 --- /dev/null +++ b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md @@ -0,0 +1,464 @@ +# Shared arrangements across timely runtimes + +- Associated: + - [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 + +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. From fd4fd81a4b8bfb02b60143293bff0ec66fe8eef4 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 14:05:47 +0000 Subject: [PATCH 2/6] design: two-runtime compute for read isolation New design package for splitting a compute replica's maintenance work from its read serving. README.md covers the full arc in two stages: stage 1 offloads fast-path index peek reads off the worker thread (no second runtime, needs only Arc batches + a Send snapshot), stage 2 stands up a second query timely runtime that imports maintenance arrangements for slow-path peeks and ad-hoc query dataflows. background.md grounds it in how a compute replica is structured today. Cross-linked with the shared-arrangements primitive design, which this design consumes. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../README.md | 2 + .../20260720_two_runtime_compute/README.md | 309 ++++++++++++++++++ .../background.md | 291 +++++++++++++++++ 3 files changed, 602 insertions(+) create mode 100644 doc/developer/design/20260720_two_runtime_compute/README.md create mode 100644 doc/developer/design/20260720_two_runtime_compute/background.md diff --git a/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md index 66a8f29dbd124..6aa3cb62e9cdb 100644 --- a/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md +++ b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md @@ -1,6 +1,8 @@ # 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` 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..747a124a48363 --- /dev/null +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -0,0 +1,309 @@ +# 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. + +## 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`), so it waits for the worker that owns the relevant +shard to come around to it. That worker may be deep in a join or a large merge. +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](../20260719_shared_arrangements_across_runtimes/README.md) +removes the second half of that restriction: arrangement batches are now `Arc`'d +and their contents are `Send + Sync` (`row-spine/src/lib.rs:175-177`), so a batch +can be read from another thread. This design uses that to move reads off the +maintenance threads. + +## Success Criteria + +- A read (a peek, and later an ad-hoc query dataflow) does not wait behind + maintenance work on a worker thread. Read tail latency is decoupled from + maintenance load. +- 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. +- Maintenance throughput is not regressed. Publishing an arrangement for reading + costs the maintenance path only constant per-batch overhead. +- The change is stageable: an early increment ships read isolation without the + full second-runtime architecture. + +## 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. +- Scheduling policy between the two runtimes (thread counts, pinning, + priorities). Discussed under open questions, not decided here. +- Multi-replica or cross-replica concerns. This is about one replica's internal + structure. + +## Solution overview: two stages + +The design is a full arc, delivered in two stages. Stage 1 gets most of the +latency win with a small, self-contained change and no second runtime. Stage 2 +is the full two-runtime architecture that also isolates ad-hoc query dataflows. + +The staging matters because the two stages need very different amounts of the +[shared-arrangements primitive](../20260719_shared_arrangements_across_runtimes/README.md). +Stage 1 needs only `Arc` batches and a `Send` snapshot. Stage 2 needs the +publish, import, and cross-thread hold machinery. + +## Stage 1: off-worker peek reads + +**Idea.** Keep one timely runtime. On an index peek, the worker does only the +cheap, gated part on-thread, then hands the expensive cursor walk to a reader +thread pool. This mirrors what Persist peeks already do +(`compute_state.rs:1254-1325`). + +**Mechanism.** When a `Peek { target: Index { id }, timestamp, .. }` arrives, the +worker: + +1. Gates as today: it waits until the trace `upper` passes `timestamp` and + checks `since <= timestamp` (`compute_state.rs:1517-1544`). This stays + on-thread because it is a frontier comparison, not a data scan. +2. Takes a `Send` snapshot of its own trace. A snapshot is a clone of the + current batch chain (a handful of `Arc` clones) plus the `since` and `upper` + frontiers. This is cheap and does not block on maintenance. +3. Hands the snapshot, the `SafeMfpPlan`, the `RowSetFinishing`, and the peek + `uuid` to a reader pool. The pool runs the existing `PeekResultIterator` + cursor walk (`compute/src/compute_state/peek_result_iterator.rs`) against the + snapshot and produces the `PeekResponse`. +4. The pool delivers the response back through the existing response path, + re-activating the worker via a `SyncActivator` exactly as the Persist peek + task does. + +**Why this is correct and cheap.** + +- The snapshot's `Arc`'d batches are immutable. Once taken, the maintenance + worker may merge and compact freely. The reader pool reads the pinned + pre-merge batches. No torn read, no lock held across the walk. +- The controller's per-peek read hold already keeps the trace `since <= + timestamp` until the `PeekResponse` is received + (`compute-client/src/controller/instance.rs:1922-1981`), so the snapshot is + accurate at `timestamp`. The existing `since` gate stays as the safety net. +- `PeekResultIterator` already operates over a `CursorList` and a `Vec` of + batches, which become `Send` once batches are `Send + Sync`. The walk moves + off-thread almost unchanged. + +**What stage 1 needs from the primitive.** Only the `Arc` batches (done in #807 +and #37743) and a `Send` snapshot type. It does *not* need a publication point, +import, or cross-thread hold registration. The worker snapshots its own trace on +demand. There is no second runtime and no cross-runtime compaction to reconcile. + +**What stage 1 does not solve.** Slow-path peeks still ship a transient dataflow +onto the maintenance runtime, and ad-hoc query dataflows still run there. Their +rendering and operator work still contends with maintenance. Stage 1 isolates +the fast-path read, which is the common case and the sharpest latency problem, +but not query dataflow rendering. + +### Cancellation and lifecycle (stage 1) + +`CancelPeek` must cancel an in-flight offloaded read. The reader pool task holds +a cancellation flag keyed by `uuid`, checked between output batches, mirroring +how `pending_peeks` is drained today (`compute_state.rs:701-705`). A snapshot +held by a running task keeps its batches alive independently of the trace, so +cancellation is about not sending a response, not about memory safety. + +## Stage 2: the query runtime + +**Idea.** Stand up a second compute timely runtime in the same process, the +*query runtime*, alongside the *maintenance runtime*. The maintenance runtime +owns indexes and MVs as today and publishes their arrangements. The query +runtime serves slow-path peeks and renders ad-hoc query dataflows by importing +those arrangements, so query rendering never competes with maintenance for +threads. + +This is where the +[shared-arrangements primitive](../20260719_shared_arrangements_across_runtimes/README.md) +is used in full: `Arranged::publish` on the maintenance side, `import` and +`snapshot` on the query side, with cross-thread hold aggregation driving +compaction. + +**Topology.** One replica process, two compute `serve` instances (two +`ClusterSpec` runtimes, `compute/src/server.rs:85-131`), equal worker counts. +Query worker `i` imports maintenance worker `i`'s publication point. Because both +runtimes route keys by the same `hash % peers` +(`compute/src/extensions/arrange.rs:133`), a key lives on the same worker ordinal +in both, so an imported arrangement is co-located with any query-side operator +on the same key, with no cross-worker exchange. This reuses the equal-worker +invariant `clusterd` already asserts between storage and compute +(`clusterd/src/lib.rs:426-429`). + +**Publication.** When the maintenance runtime renders an index +(`export_index`, `compute/src/render.rs:688`), it also publishes the arrangement +and records the resulting `SharedTraceHandle`s in a per-process, per-worker +registry keyed by `GlobalId`, so the query runtime's worker `i` can find worker +`i`'s handle. Publication cost is one `Arc` clone plus a mutex push per batch, so +maintenance throughput is unaffected. + +**Query-side reads.** + +- A slow-path peek is served by the query runtime: instead of shipping a + transient index dataflow onto the maintenance runtime, the query runtime + renders it, importing the base arrangements it needs, and peeks its own result. +- An ad-hoc query dataflow (the general form) is rendered entirely on the query + runtime, importing maintenance arrangements as ordinary `Arranged` collections + through `SharedTraceHandle::import`. + +**Command routing.** The replica process demultiplexes the controller's command +stream between the two runtimes: `Peek`, `CancelPeek`, and transient query +dataflows go to the query runtime; `CreateDataflow` for maintained +indexes/MVs, `Schedule`, `AllowCompaction`, and `AllowWrites` go to the +maintenance runtime. Both runtimes see the lifecycle commands (`CreateInstance`, +`UpdateConfiguration`, `InitializationComplete`). Each runtime keeps its own +worker-0 broadcast (`compute/src/command_channel.rs`), so ordering within a +runtime is preserved. + +**Response merging.** The process merges responses from both runtimes into the +single stream the controller expects. `PeekResponse`s originate in the query +runtime, `Frontiers` for maintained collections in the maintenance runtime. This +extends the existing per-process `PartitionedComputeState` merge +(`compute-client/src/service.rs:79`) to span two runtimes rather than only the +workers of one. + +## Compaction reconciliation + +This is the subtlest correctness point, and it differs by stage. + +**Stage 1.** No reconciliation. There is one runtime and one trace. The +controller drives `since` through `AllowCompaction` exactly as today. The +offloaded read is protected by the controller's per-peek read hold plus the +snapshot's immutable batches. + +**Stage 2.** Two things now hold a maintenance arrangement back from compaction: + +1. The controller, through `AllowCompaction` to the maintenance runtime, derived + from its read holds (`instance.rs:1922-1981`). +2. The query runtime, through the primitive's hold aggregation: an imported + arrangement registers logical and physical holds that the publisher forwards + to the maintenance `TraceAgent`. + +These must not fight. The controller remains the authority: it already holds a +read capability for the query being served (a peek hold, or the read hold behind +a slow-path dataflow's `as_of`). The query runtime's holds are a *finer-grained, +in-process realization* of that same requirement. The design keeps them +consistent by construction: + +- The query runtime never needs a hold *below* what the controller has already + guaranteed. It imports at an `as_of` the controller vouched for, so its holds + sit at or above the controller's `since`. +- The maintenance runtime's `TraceAgent` receives the *meet* of the controller's + `AllowCompaction` frontier and the query runtime's forwarded holds. Because + compaction only advances, the effective `since` is the lower of the two, which + is always `<=` any time either party still needs. + +The open risk is a query-runtime hold that lags the controller: if the query +runtime wedges while holding a hold, it pins maintenance compaction. That is the +lease-expiry question the primitive design already flags, surfaced here as a +concrete operational concern (see open questions). + +## Controller transparency + +Two options for how much the controller knows about the split. + +- **Transparent.** The controller still sees one replica and one response + stream. The process routes commands and merges responses internally. Minimal + protocol change. This is the recommended starting point, and it is enough for + both stages: the split is an internal replica detail, and the controller's + compaction authority is unchanged because `AllowCompaction` still targets the + maintenance runtime's traces. +- **Explicit.** The controller models two runtimes and routes to them directly, + enabling smarter placement and per-runtime introspection. This is a real + protocol and controller change (`ReplicaState` would grow a sub-runtime axis, + `instance.rs:3082-3097`). Deferred until transparency proves insufficient. + +The recommendation is transparent for both stages. The cost is that peek routing +and per-runtime memory attribution are hidden from the controller, which +introspection will eventually want (open question). + +## 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`, `background.md`). 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 must agree on it, 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 query + metrics apart. + +## Thread allocation + +Two runtimes of `N` workers each put `2N` worker threads on the same cores. The +query runtime is mostly blocked waiting on reads and imports, so it is not `N` +cores of steady load, but the design must not silently double the thread count +and oversubscribe. Options (thread split, smaller query pool, cooperative +yielding) are a scheduling decision left to the open questions, not fixed here. +Stage 1 avoids this entirely (one runtime plus a bounded reader pool). + +## Open questions + +- **Thread and core allocation between runtimes.** How many worker threads does + the query runtime get, and how are cores shared with maintenance? Needs + measurement. +- **Lease expiry for query-runtime holds (stage 2).** A wedged query runtime + pins maintenance compaction through its forwarded holds. Do we want a + time-based lease with snapshot invalidation, or is same-process trust enough? + The primitive design assumes trust for now. +- **Import queue backpressure (stage 2).** The primitive's replay queue is + bounded today only because producer and consumer share a worker step. Across + runtimes they are independently scheduled, so a lagging query runtime can grow + the queue without bound. A bound plus publisher backpressure is likely needed + before long-lived query dataflows are safe. Captured in the primitive design. +- **Introspection and memory attribution.** Under transparency the controller + cannot see which runtime holds what memory. Shared arrangements are counted by + whoever drops last. Introspection needs a per-runtime view eventually. +- **Slow-path versus stage 1.** Stage 1 offloads fast-path index peeks but not + slow-path peeks. Is it worth teaching the coordinator to prefer the fast path + more aggressively once stage 1 lands, deferring stage 2? +- **Failure and restart.** If the query runtime dies, peeks in flight fail and + should be retried on maintenance as a fallback. If maintenance dies, the whole + replica restarts as today. The fallback path needs specifying. + +## Alternatives + +- **Priority scheduling inside one runtime.** Yield more finely so peeks can + 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 + query dataflow rendering. Stage 1's offload achieves read isolation without + this coupling. +- **Serve all peeks from Persist.** Peeks can read persist shards directly + (`PeekTarget::Persist`), bypassing arrangements. This already 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 would copy data across the process + boundary, which is what persist already offers. In-process sharing is the whole + point. +- **Do only stage 1, never stage 2.** Viable if fast-path peek isolation turns + out to cover the observed latency problem. Stage 2 is justified only if ad-hoc + query dataflow rendering is shown to contend materially with maintenance. 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..bb2b9ccb037c4 --- /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`. Both spines are already `Arc`-backed +after the batch migration (`row-spine/src/lib.rs:51`, `typedefs.rs:107`), and +their batch contents are asserted `Send + Sync` +(`row-spine/src/lib.rs:175-177`). The remaining single-runtime coupling 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. From 4562ee2c50343ee528701599e8527dc41a9e9274 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 14:29:10 +0000 Subject: [PATCH 3/6] design: fold in two-runtime compute decisions Lock in the direction: the split stays internal to compute behind one protocol endpoint, with a routing layer dispatching by target id (peeks and temporary dataflows to the interactive runtime, indexes/MVs/subscribes to the maintenance runtime). The interactive runtime never pins maintenance compaction, it tracks maintenance's since or holds nothing, so the controller stays the sole compaction authority and no interactive lease can wedge maintenance. Equal worker counts are a categorical requirement so cross-runtime exchange never arises. Rename the read-serving runtime to the interactive runtime throughout. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../20260720_two_runtime_compute/README.md | 240 ++++++++++-------- 1 file changed, 134 insertions(+), 106 deletions(-) diff --git a/doc/developer/design/20260720_two_runtime_compute/README.md b/doc/developer/design/20260720_two_runtime_compute/README.md index 747a124a48363..c262ed3ba01a3 100644 --- a/doc/developer/design/20260720_two_runtime_compute/README.md +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -36,7 +36,7 @@ maintenance threads. ## Success Criteria -- A read (a peek, and later an ad-hoc query dataflow) does not wait behind +- A read (a peek, and later a temporary dataflow) does not wait behind maintenance work on a worker thread. Read tail latency is decoupled from maintenance load. - Correctness is unchanged. A read observes exactly the collection at its @@ -64,12 +64,14 @@ maintenance threads. The design is a full arc, delivered in two stages. Stage 1 gets most of the latency win with a small, self-contained change and no second runtime. Stage 2 -is the full two-runtime architecture that also isolates ad-hoc query dataflows. +is the full two-runtime architecture that also isolates temporary dataflows. The staging matters because the two stages need very different amounts of the [shared-arrangements primitive](../20260719_shared_arrangements_across_runtimes/README.md). -Stage 1 needs only `Arc` batches and a `Send` snapshot. Stage 2 needs the -publish, import, and cross-thread hold machinery. +Stage 1 needs only `Arc` batches and a `Send` snapshot. Stage 2 adds publish and +import. Its cross-thread hold aggregation is available but not load-bearing, +because the interactive runtime reads read-only and never holds maintenance +compaction back (see "Compaction reconciliation"). ## Stage 1: off-worker peek reads @@ -114,10 +116,10 @@ import, or cross-thread hold registration. The worker snapshots its own trace on demand. There is no second runtime and no cross-runtime compaction to reconcile. **What stage 1 does not solve.** Slow-path peeks still ship a transient dataflow -onto the maintenance runtime, and ad-hoc query dataflows still run there. Their +onto the maintenance runtime, and temporary dataflows still run there. Their rendering and operator work still contends with maintenance. Stage 1 isolates the fast-path read, which is the common case and the sharpest latency problem, -but not query dataflow rendering. +but not temporary-dataflow rendering. ### Cancellation and lifecycle (stage 1) @@ -127,60 +129,82 @@ how `pending_peeks` is drained today (`compute_state.rs:701-705`). A snapshot held by a running task keeps its batches alive independently of the trace, so cancellation is about not sending a response, not about memory safety. -## Stage 2: the query runtime +## Stage 2: the interactive runtime **Idea.** Stand up a second compute timely runtime in the same process, the -*query runtime*, alongside the *maintenance runtime*. The maintenance runtime -owns indexes and MVs as today and publishes their arrangements. The query -runtime serves slow-path peeks and renders ad-hoc query dataflows by importing -those arrangements, so query rendering never competes with maintenance for -threads. - -This is where the -[shared-arrangements primitive](../20260719_shared_arrangements_across_runtimes/README.md) -is used in full: `Arranged::publish` on the maintenance side, `import` and -`snapshot` on the query side, with cross-thread hold aggregation driving -compaction. +*interactive runtime*, alongside the *maintenance runtime*. The maintenance +runtime owns the durable, maintained work (indexes, materialized views, +subscribes) as today and publishes its index arrangements. The interactive +runtime owns the ephemeral, read-oriented work (peeks and temporary dataflows), +importing maintenance arrangements so that serving reads never competes with +maintenance for threads. + +The split stays **internal to compute**. The controller sees one replica behind +one protocol endpoint. A routing layer in the replica process dispatches each +command to the runtime that owns its target, and merges both runtimes' responses +back into the single stream the controller expects. The controller protocol and +its compaction authority are unchanged. **Topology.** One replica process, two compute `serve` instances (two -`ClusterSpec` runtimes, `compute/src/server.rs:85-131`), equal worker counts. -Query worker `i` imports maintenance worker `i`'s publication point. Because both -runtimes route keys by the same `hash % peers` -(`compute/src/extensions/arrange.rs:133`), a key lives on the same worker ordinal -in both, so an imported arrangement is co-located with any query-side operator -on the same key, with no cross-worker exchange. This reuses the equal-worker -invariant `clusterd` already asserts between storage and compute -(`clusterd/src/lib.rs:426-429`). +`ClusterSpec` runtimes, `compute/src/server.rs:85-131`). The interactive runtime +runs with **exactly the same worker count** as the maintenance runtime. This is +a categorical requirement, not a tuning knob: equal worker counts plus the +shared `hash % peers` key routing (`compute/src/extensions/arrange.rs:133`) mean +a key lives on the same worker ordinal in both runtimes, so interactive worker +`i` imports maintenance worker `i`'s publication point and any interactive-side +operator on that key is co-located with no cross-worker exchange, ever. It +reuses the equal-worker invariant `clusterd` already asserts between storage and +compute (`clusterd/src/lib.rs:426-429`). + +**Routing by target id.** The routing layer dispatches on the id a command +targets: + +- `Peek`, `CancelPeek`, and `CreateDataflow` for a **temporary dataflow** go to + the interactive runtime. A temporary dataflow is identified by its export + being a `GlobalId::Transient` (`compute-types/src/dataflows.rs:380-383`), which + is exactly how a slow-path peek's one-off dataflow and other ad-hoc query + work are already tagged + are already tagged. +- `CreateDataflow` for maintained work (indexes, MVs, and **subscribes**), + `Schedule`, `AllowWrites`, and `AllowCompaction` for a maintained id go to the + maintenance runtime. +- `AllowCompaction` (including the empty-frontier drop) for a transient id go to + the interactive runtime that owns that temporary dataflow. +- Lifecycle commands (`CreateInstance`, `UpdateConfiguration`, + `InitializationComplete`) go to both. + +The one wrinkle is that subscribes also export transient ids yet must stay on +the maintenance runtime, because they are continuously maintained rather than +read once. So the routing predicate is not "transient id" alone. It is "a +one-shot temporary dataflow" versus "maintained work", and a subscribe is +distinguished by carrying a subscribe sink in its `DataflowDescription`. The +exact predicate is an implementation detail, but the principle is: maintained +work to maintenance, one-shot reads to interactive. + +Each runtime keeps its own worker-0 broadcast +(`compute/src/command_channel.rs`), so command ordering within a runtime is +preserved. **Publication.** When the maintenance runtime renders an index (`export_index`, `compute/src/render.rs:688`), it also publishes the arrangement and records the resulting `SharedTraceHandle`s in a per-process, per-worker -registry keyed by `GlobalId`, so the query runtime's worker `i` can find worker +registry keyed by `GlobalId`, so interactive worker `i` finds maintenance worker `i`'s handle. Publication cost is one `Arc` clone plus a mutex push per batch, so maintenance throughput is unaffected. -**Query-side reads.** +**Interactive-side reads.** -- A slow-path peek is served by the query runtime: instead of shipping a - transient index dataflow onto the maintenance runtime, the query runtime +- A slow-path peek is served by the interactive runtime: instead of shipping a + transient index dataflow onto the maintenance runtime, the interactive runtime renders it, importing the base arrangements it needs, and peeks its own result. -- An ad-hoc query dataflow (the general form) is rendered entirely on the query - runtime, importing maintenance arrangements as ordinary `Arranged` collections - through `SharedTraceHandle::import`. - -**Command routing.** The replica process demultiplexes the controller's command -stream between the two runtimes: `Peek`, `CancelPeek`, and transient query -dataflows go to the query runtime; `CreateDataflow` for maintained -indexes/MVs, `Schedule`, `AllowCompaction`, and `AllowWrites` go to the -maintenance runtime. Both runtimes see the lifecycle commands (`CreateInstance`, -`UpdateConfiguration`, `InitializationComplete`). Each runtime keeps its own -worker-0 broadcast (`compute/src/command_channel.rs`), so ordering within a -runtime is preserved. - -**Response merging.** The process merges responses from both runtimes into the -single stream the controller expects. `PeekResponse`s originate in the query -runtime, `Frontiers` for maintained collections in the maintenance runtime. This -extends the existing per-process `PartitionedComputeState` merge +- A temporary dataflow (the general form) is rendered entirely on the + interactive runtime, importing maintenance arrangements as ordinary `Arranged` + collections through `SharedTraceHandle::import`. + +**Response merging.** The routing layer merges responses from both runtimes into +the single stream the controller expects. `PeekResponse`s originate in the +interactive runtime, `Frontiers` for maintained collections in the maintenance +runtime. This extends the existing per-process `PartitionedComputeState` merge (`compute-client/src/service.rs:79`) to span two runtimes rather than only the workers of one. @@ -193,51 +217,56 @@ controller drives `since` through `AllowCompaction` exactly as today. The offloaded read is protected by the controller's per-peek read hold plus the snapshot's immutable batches. -**Stage 2.** Two things now hold a maintenance arrangement back from compaction: - -1. The controller, through `AllowCompaction` to the maintenance runtime, derived - from its read holds (`instance.rs:1922-1981`). -2. The query runtime, through the primitive's hold aggregation: an imported - arrangement registers logical and physical holds that the publisher forwards - to the maintenance `TraceAgent`. - -These must not fight. The controller remains the authority: it already holds a -read capability for the query being served (a peek hold, or the read hold behind -a slow-path dataflow's `as_of`). The query runtime's holds are a *finer-grained, -in-process realization* of that same requirement. The design keeps them -consistent by construction: - -- The query runtime never needs a hold *below* what the controller has already - guaranteed. It imports at an `as_of` the controller vouched for, so its holds - sit at or above the controller's `since`. -- The maintenance runtime's `TraceAgent` receives the *meet* of the controller's - `AllowCompaction` frontier and the query runtime's forwarded holds. Because - compaction only advances, the effective `since` is the lower of the two, which - is always `<=` any time either party still needs. - -The open risk is a query-runtime hold that lags the controller: if the query -runtime wedges while holding a hold, it pins maintenance compaction. That is the -lease-expiry question the primitive design already flags, surfaced here as a -concrete operational concern (see open questions). - -## Controller transparency - -Two options for how much the controller knows about the split. - -- **Transparent.** The controller still sees one replica and one response - stream. The process routes commands and merges responses internally. Minimal - protocol change. This is the recommended starting point, and it is enough for - both stages: the split is an internal replica detail, and the controller's - compaction authority is unchanged because `AllowCompaction` still targets the - maintenance runtime's traces. -- **Explicit.** The controller models two runtimes and routes to them directly, - enabling smarter placement and per-runtime introspection. This is a real - protocol and controller change (`ReplicaState` would grow a sub-runtime axis, - `instance.rs:3082-3097`). Deferred until transparency proves insufficient. - -The recommendation is transparent for both stages. The cost is that peek routing -and per-runtime memory attribution are hidden from the controller, which -introspection will eventually want (open question). +**Stage 2.** The controller stays the sole compaction authority, and the +interactive runtime **never pins the maintenance arrangement back**. It either +tracks the maintenance runtime's `since` (following it as it advances) or holds +nothing at all. It never forwards a hold that would keep the maintenance trace +below the maintenance runtime's own compaction frontier. + +This works because the controller already holds the maintenance arrangement back +for exactly as long as the 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` +(`instance.rs:1922-1981`). The maintenance runtime will not compact past that +hold, so an interactive-side read that imports at that `as_of` (or snapshots at +that timestamp) stays valid for its whole life without the interactive runtime +holding anything itself. When the read finishes and the controller releases, +the maintenance runtime compacts and the interactive side follows. + +The consequence is a deliberate non-goal: the interactive runtime does not keep +a maintenance arrangement readable at times the controller has allowed to +compact. There is no independent interactive lease, so a wedged interactive +runtime cannot pin maintenance compaction, and the lease-expiry question the +primitive raises does not arise for this integration. The cost is that an +interactive read must sit within a compaction window the controller is already +holding, which it always is, because the controller acquires that hold before +issuing the peek or creating the temporary dataflow. + +This also feeds back to the primitive: because the interactive runtime reads +read-only and never holds below maintenance's `since`, the sharing module's +cross-thread hold *aggregation* is not exercised by this integration. Imports +register holds that at most mirror maintenance, so the publisher's hold +forwarding is a no-op relative to maintenance's own compaction. The primitive's +`snapshot` and `import` are used. Its hold-aggregation machinery is available but +not load-bearing here, which removes the wedge and lease hazards from the +compute integration entirely. + +## Internal to compute, one protocol endpoint + +The split is invisible to the controller. It sees one replica behind one +protocol endpoint. The replica's routing layer dispatches commands to the owning +runtime (see "Routing by target id") and merges both runtimes' responses into +one stream. The controller protocol is unchanged, and its compaction authority +is unchanged because `AllowCompaction` for a maintained id still targets the +maintenance runtime's trace. + +This is a deliberate choice over teaching the controller about two runtimes. An +explicit model (the controller routing to two sub-runtimes, `ReplicaState` +growing a sub-runtime axis at `instance.rs:3082-3097`) would enable smarter +placement and per-runtime introspection, but it is a real protocol and +controller change and it 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. That is a +follow-up, not a blocker (see open questions). ## Resource sharing and the single-instance assumption @@ -252,39 +281,38 @@ replica process hosts a single instance" (`compute_state.rs:495-496`): - 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 query + pattern exists, but per-runtime labels are needed to tell maintenance and interactive metrics apart. ## Thread allocation Two runtimes of `N` workers each put `2N` worker threads on the same cores. The -query runtime is mostly blocked waiting on reads and imports, so it is not `N` +interactive runtime is mostly blocked waiting on reads and imports, so it is not `N` cores of steady load, but the design must not silently double the thread count -and oversubscribe. Options (thread split, smaller query pool, cooperative +and oversubscribe. Options (thread split, smaller interactive pool, cooperative yielding) are a scheduling decision left to the open questions, not fixed here. Stage 1 avoids this entirely (one runtime plus a bounded reader pool). ## Open questions - **Thread and core allocation between runtimes.** How many worker threads does - the query runtime get, and how are cores shared with maintenance? Needs + the interactive runtime get, and how are cores shared with maintenance? Needs measurement. -- **Lease expiry for query-runtime holds (stage 2).** A wedged query runtime - pins maintenance compaction through its forwarded holds. Do we want a - time-based lease with snapshot invalidation, or is same-process trust enough? - The primitive design assumes trust for now. - **Import queue backpressure (stage 2).** The primitive's replay queue is bounded today only because producer and consumer share a worker step. Across - runtimes they are independently scheduled, so a lagging query runtime can grow - the queue without bound. A bound plus publisher backpressure is likely needed - before long-lived query dataflows are safe. Captured in the primitive design. + runtimes they are independently scheduled, so a lagging interactive runtime can + grow the queue without bound. A bound plus publisher backpressure is likely + needed. This is a liveness and memory concern, not a compaction one: since the + interactive runtime holds nothing back (see "Compaction reconciliation"), a + slow reader cannot wedge maintenance compaction, only fall behind on its own + imports. Captured in the primitive design. - **Introspection and memory attribution.** Under transparency the controller cannot see which runtime holds what memory. Shared arrangements are counted by whoever drops last. Introspection needs a per-runtime view eventually. - **Slow-path versus stage 1.** Stage 1 offloads fast-path index peeks but not slow-path peeks. Is it worth teaching the coordinator to prefer the fast path more aggressively once stage 1 lands, deferring stage 2? -- **Failure and restart.** If the query runtime dies, peeks in flight fail and +- **Failure and restart.** If the interactive runtime dies, peeks in flight fail and should be retried on maintenance as a fallback. If maintenance dies, the whole replica restarts as today. The fallback path needs specifying. @@ -294,7 +322,7 @@ Stage 1 avoids this entirely (one runtime plus a bounded reader pool). 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 - query dataflow rendering. Stage 1's offload achieves read isolation without + temporary-dataflow rendering. Stage 1's offload achieves read isolation without this coupling. - **Serve all peeks from Persist.** Peeks can read persist shards directly (`PeekTarget::Persist`), bypassing arrangements. This already exists for some @@ -306,4 +334,4 @@ Stage 1 avoids this entirely (one runtime plus a bounded reader pool). point. - **Do only stage 1, never stage 2.** Viable if fast-path peek isolation turns out to cover the observed latency problem. Stage 2 is justified only if ad-hoc - query dataflow rendering is shown to contend materially with maintenance. + temporary-dataflow rendering is shown to contend materially with maintenance. From 188ef0fca42912f8333a3561c894d72507957de5 Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 15:06:34 +0000 Subject: [PATCH 4/6] design: settle worker counts (deferred core sharing) and shared fate Equal worker counts are settled, with core sharing between the 2N threads left to later measurement. The two runtimes share fate like the replica process does today: a fatal error in either restarts the whole replica, with no partial failure and no cross-runtime read fallback. Retire the corresponding open questions. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../20260720_two_runtime_compute/README.md | 35 ++++++++++++------- 1 file changed, 23 insertions(+), 12 deletions(-) diff --git a/doc/developer/design/20260720_two_runtime_compute/README.md b/doc/developer/design/20260720_two_runtime_compute/README.md index c262ed3ba01a3..f9acd8697343d 100644 --- a/doc/developer/design/20260720_two_runtime_compute/README.md +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -286,18 +286,32 @@ replica process hosts a single instance" (`compute_state.rs:495-496`): ## Thread allocation -Two runtimes of `N` workers each put `2N` worker threads on the same cores. The -interactive runtime is mostly blocked waiting on reads and imports, so it is not `N` -cores of steady load, but the design must not silently double the thread count -and oversubscribe. Options (thread split, smaller interactive pool, cooperative -yielding) are a scheduling decision left to the open questions, not fixed here. -Stage 1 avoids this entirely (one runtime plus a bounded reader pool). +The interactive runtime runs the same number of workers as the maintenance +runtime, which is a correctness requirement for pairwise import (see "Topology") +independent of scheduling. That puts `2N` worker threads on the process's cores. +How those threads share cores (dedicated splits, oversubscription relying on the +interactive runtime being mostly blocked on reads and imports, pinning) is a +tuning question deferred to later measurement, not fixed by this design. Stage 1 +sidesteps it entirely: one runtime plus a bounded reader pool. + +## 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 peek 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 deliberate. It keeps the failure model +identical to today and avoids a cross-runtime fallback path that would otherwise +need its own correctness argument. ## Open questions -- **Thread and core allocation between runtimes.** How many worker threads does - the interactive runtime get, and how are cores shared with maintenance? Needs - measurement. +- **Core sharing between runtimes.** Worker counts are settled (equal, see + "Thread allocation"). How the resulting `2N` threads share cores, and whether + oversubscription is acceptable given the interactive runtime is mostly blocked + on reads, is deferred to later measurement. - **Import queue backpressure (stage 2).** The primitive's replay queue is bounded today only because producer and consumer share a worker step. Across runtimes they are independently scheduled, so a lagging interactive runtime can @@ -312,9 +326,6 @@ Stage 1 avoids this entirely (one runtime plus a bounded reader pool). - **Slow-path versus stage 1.** Stage 1 offloads fast-path index peeks but not slow-path peeks. Is it worth teaching the coordinator to prefer the fast path more aggressively once stage 1 lands, deferring stage 2? -- **Failure and restart.** If the interactive runtime dies, peeks in flight fail and - should be retried on maintenance as a fallback. If maintenance dies, the whole - replica restarts as today. The fallback path needs specifying. ## Alternatives From 0bafdbc2091bf0c41fa10ed83d3c05de2fbce36c Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 16:22:46 +0000 Subject: [PATCH 5/6] design: rewrite two-runtime compute around one architecture Collapse the worker-mediated stage 1 into a single architecture: the interactive runtime serves all reads from threads that hold the published arrangement directly, so a read's latency floor is CPU contention with maintenance, not waiting for a maintenance worker to return to its loop. Correct the framing that adversarial review flagged: the walk itself is not made faster (a point-lookup stays microseconds), it is moved off the maintenance critical path; the snapshot memory cost is transient and bounded, not a new doubling, since fueled merges already hold input and output together. State the compaction and Send work as required primitive changes rather than done facts: the publisher must not pin compaction (today it freezes it at publish-time since), reads must register no independent hold, and snapshot_at must enforce the since gate. Call out the new components honestly: the process-level multiplexer (not a PartitionedComputeState reuse), the publication registry with a block-until-published handshake, the interactive read path, the equal-peers assertion, and the cancellation and batches_through hazards. Fix the background note that described the spines as Arc-backed on main. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../20260720_two_runtime_compute/README.md | 595 ++++++++++-------- .../background.md | 8 +- 2 files changed, 329 insertions(+), 274 deletions(-) diff --git a/doc/developer/design/20260720_two_runtime_compute/README.md b/doc/developer/design/20260720_two_runtime_compute/README.md index f9acd8697343d..013eaca0fe01e 100644 --- a/doc/developer/design/20260720_two_runtime_compute/README.md +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -7,7 +7,8 @@ the differential-level primitive this design consumes. - Implementation of the `Arc`-batches prerequisite: differential TimelyDataflow/differential-dataflow#807, materialize - MaterializeInc/materialize#37743. + MaterializeInc/materialize#37743. File and line references to the `Arc` + spines below are to the state on those branches, not to `main`. ## The Problem @@ -19,34 +20,43 @@ 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`), so it waits for the worker that owns the relevant -shard to come around to it. That worker may be deep in a join or a large merge. -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. +`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](../20260719_shared_arrangements_across_runtimes/README.md) -removes the second half of that restriction: arrangement batches are now `Arc`'d -and their contents are `Send + Sync` (`row-spine/src/lib.rs:175-177`), so a batch -can be read from another thread. This design uses that to move reads off the -maintenance threads. +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 (a peek, and later a temporary dataflow) does not wait behind - maintenance work on a worker thread. Read tail latency is decoupled from - maintenance load. +- 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. + compacts past a time a read still needs, and the read path never pins an + arrangement back beyond what the controller already holds. - Maintenance throughput is not regressed. Publishing an arrangement for reading - costs the maintenance path only constant per-batch overhead. -- The change is stageable: an early increment ships read isolation without the - full second-runtime architecture. + costs the maintenance path only constant per-batch overhead, and the read side + does not hold maintenance compaction back. + +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 @@ -55,244 +65,253 @@ maintenance threads. - 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. -- Scheduling policy between the two runtimes (thread counts, pinning, - priorities). Discussed under open questions, not decided here. +- 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. -## Solution overview: two stages - -The design is a full arc, delivered in two stages. Stage 1 gets most of the -latency win with a small, self-contained change and no second runtime. Stage 2 -is the full two-runtime architecture that also isolates temporary dataflows. - -The staging matters because the two stages need very different amounts of the -[shared-arrangements primitive](../20260719_shared_arrangements_across_runtimes/README.md). -Stage 1 needs only `Arc` batches and a `Send` snapshot. Stage 2 adds publish and -import. Its cross-thread hold aggregation is available but not load-bearing, -because the interactive runtime reads read-only and never holds maintenance -compaction back (see "Compaction reconciliation"). - -## Stage 1: off-worker peek reads - -**Idea.** Keep one timely runtime. On an index peek, the worker does only the -cheap, gated part on-thread, then hands the expensive cursor walk to a reader -thread pool. This mirrors what Persist peeks already do -(`compute_state.rs:1254-1325`). - -**Mechanism.** When a `Peek { target: Index { id }, timestamp, .. }` arrives, the -worker: - -1. Gates as today: it waits until the trace `upper` passes `timestamp` and - checks `since <= timestamp` (`compute_state.rs:1517-1544`). This stays - on-thread because it is a frontier comparison, not a data scan. -2. Takes a `Send` snapshot of its own trace. A snapshot is a clone of the - current batch chain (a handful of `Arc` clones) plus the `since` and `upper` - frontiers. This is cheap and does not block on maintenance. -3. Hands the snapshot, the `SafeMfpPlan`, the `RowSetFinishing`, and the peek - `uuid` to a reader pool. The pool runs the existing `PeekResultIterator` - cursor walk (`compute/src/compute_state/peek_result_iterator.rs`) against the - snapshot and produces the `PeekResponse`. -4. The pool delivers the response back through the existing response path, - re-activating the worker via a `SyncActivator` exactly as the Persist peek - task does. - -**Why this is correct and cheap.** - -- The snapshot's `Arc`'d batches are immutable. Once taken, the maintenance - worker may merge and compact freely. The reader pool reads the pinned - pre-merge batches. No torn read, no lock held across the walk. -- The controller's per-peek read hold already keeps the trace `since <= - timestamp` until the `PeekResponse` is received - (`compute-client/src/controller/instance.rs:1922-1981`), so the snapshot is - accurate at `timestamp`. The existing `since` gate stays as the safety net. -- `PeekResultIterator` already operates over a `CursorList` and a `Vec` of - batches, which become `Send` once batches are `Send + Sync`. The walk moves - off-thread almost unchanged. - -**What stage 1 needs from the primitive.** Only the `Arc` batches (done in #807 -and #37743) and a `Send` snapshot type. It does *not* need a publication point, -import, or cross-thread hold registration. The worker snapshots its own trace on -demand. There is no second runtime and no cross-runtime compaction to reconcile. - -**What stage 1 does not solve.** Slow-path peeks still ship a transient dataflow -onto the maintenance runtime, and temporary dataflows still run there. Their -rendering and operator work still contends with maintenance. Stage 1 isolates -the fast-path read, which is the common case and the sharpest latency problem, -but not temporary-dataflow rendering. - -### Cancellation and lifecycle (stage 1) - -`CancelPeek` must cancel an in-flight offloaded read. The reader pool task holds -a cancellation flag keyed by `uuid`, checked between output batches, mirroring -how `pending_peeks` is drained today (`compute_state.rs:701-705`). A snapshot -held by a running task keeps its batches alive independently of the trace, so -cancellation is about not sending a response, not about memory safety. - -## Stage 2: the interactive runtime - -**Idea.** Stand up a second compute timely runtime in the same process, the -*interactive runtime*, alongside the *maintenance runtime*. The maintenance -runtime owns the durable, maintained work (indexes, materialized views, -subscribes) as today and publishes its index arrangements. The interactive -runtime owns the ephemeral, read-oriented work (peeks and temporary dataflows), -importing maintenance arrangements so that serving reads never competes with -maintenance for threads. - -The split stays **internal to compute**. The controller sees one replica behind -one protocol endpoint. A routing layer in the replica process dispatches each -command to the runtime that owns its target, and merges both runtimes' responses -back into the single stream the controller expects. The controller protocol and -its compaction authority are unchanged. - -**Topology.** One replica process, two compute `serve` instances (two -`ClusterSpec` runtimes, `compute/src/server.rs:85-131`). The interactive runtime -runs with **exactly the same worker count** as the maintenance runtime. This is -a categorical requirement, not a tuning knob: equal worker counts plus the -shared `hash % peers` key routing (`compute/src/extensions/arrange.rs:133`) mean -a key lives on the same worker ordinal in both runtimes, so interactive worker -`i` imports maintenance worker `i`'s publication point and any interactive-side -operator on that key is co-located with no cross-worker exchange, ever. It -reuses the equal-worker invariant `clusterd` already asserts between storage and -compute (`clusterd/src/lib.rs:426-429`). - -**Routing by target id.** The routing layer dispatches on the id a command +## 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. + +### 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 `SharedTraceHandle::import`, 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. + +## 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 runtime. A temporary dataflow is identified by its export - being a `GlobalId::Transient` (`compute-types/src/dataflows.rs:380-383`), which - is exactly how a slow-path peek's one-off dataflow and other ad-hoc query - work are already tagged - are already tagged. -- `CreateDataflow` for maintained work (indexes, MVs, and **subscribes**), - `Schedule`, `AllowWrites`, and `AllowCompaction` for a maintained id go to the - maintenance runtime. -- `AllowCompaction` (including the empty-frontier drop) for a transient id go to - the interactive runtime that owns that temporary dataflow. +- `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. -The one wrinkle is that subscribes also export transient ids yet must stay on -the maintenance runtime, because they are continuously maintained rather than -read once. So the routing predicate is not "transient id" alone. It is "a -one-shot temporary dataflow" versus "maintained work", and a subscribe is -distinguished by carrying a subscribe sink in its `DataflowDescription`. The -exact predicate is an implementation detail, but the principle is: maintained -work to maintenance, one-shot reads to interactive. - -Each runtime keeps its own worker-0 broadcast -(`compute/src/command_channel.rs`), so command ordering within a runtime is -preserved. - -**Publication.** When the maintenance runtime renders an index -(`export_index`, `compute/src/render.rs:688`), it also publishes the arrangement -and records the resulting `SharedTraceHandle`s in a per-process, per-worker -registry keyed by `GlobalId`, so interactive worker `i` finds maintenance worker -`i`'s handle. Publication cost is one `Arc` clone plus a mutex push per batch, so -maintenance throughput is unaffected. - -**Interactive-side reads.** - -- A slow-path peek is served by the interactive runtime: instead of shipping a - transient index dataflow onto the maintenance runtime, the interactive runtime - renders it, importing the base arrangements it needs, and peeks its own result. -- A temporary dataflow (the general form) is rendered entirely on the - interactive runtime, importing maintenance arrangements as ordinary `Arranged` - collections through `SharedTraceHandle::import`. - -**Response merging.** The routing layer merges responses from both runtimes into -the single stream the controller expects. `PeekResponse`s originate in the -interactive runtime, `Frontiers` for maintained collections in the maintenance -runtime. This extends the existing per-process `PartitionedComputeState` merge -(`compute-client/src/service.rs:79`) to span two runtimes rather than only the -workers of one. - -## Compaction reconciliation - -This is the subtlest correctness point, and it differs by stage. - -**Stage 1.** No reconciliation. There is one runtime and one trace. The -controller drives `since` through `AllowCompaction` exactly as today. The -offloaded read is protected by the controller's per-peek read hold plus the -snapshot's immutable batches. - -**Stage 2.** The controller stays the sole compaction authority, and the -interactive runtime **never pins the maintenance arrangement back**. It either -tracks the maintenance runtime's `since` (following it as it advances) or holds -nothing at all. It never forwards a hold that would keep the maintenance trace -below the maintenance runtime's own compaction frontier. - -This works because the controller already holds the maintenance arrangement back -for exactly as long as the 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` -(`instance.rs:1922-1981`). The maintenance runtime will not compact past that -hold, so an interactive-side read that imports at that `as_of` (or snapshots at -that timestamp) stays valid for its whole life without the interactive runtime -holding anything itself. When the read finishes and the controller releases, -the maintenance runtime compacts and the interactive side follows. - -The consequence is a deliberate non-goal: the interactive runtime does not keep -a maintenance arrangement readable at times the controller has allowed to -compact. There is no independent interactive lease, so a wedged interactive -runtime cannot pin maintenance compaction, and the lease-expiry question the -primitive raises does not arise for this integration. The cost is that an -interactive read must sit within a compaction window the controller is already -holding, which it always is, because the controller acquires that hold before -issuing the peek or creating the temporary dataflow. - -This also feeds back to the primitive: because the interactive runtime reads -read-only and never holds below maintenance's `since`, the sharing module's -cross-thread hold *aggregation* is not exercised by this integration. Imports -register holds that at most mirror maintenance, so the publisher's hold -forwarding is a no-op relative to maintenance's own compaction. The primitive's -`snapshot` and `import` are used. Its hold-aggregation machinery is available but -not load-bearing here, which removes the wedge and lease hazards from the -compute integration entirely. +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`. + +## Publication and the publish-versus-read handshake + +When the maintenance runtime renders an index (`export_index`, +`compute/src/render.rs:688`), it publishes the arrangement and records the +resulting `SharedTraceHandle`s in a **per-process publication registry** keyed by +`GlobalId`, one entry per worker ordinal. This registry is a new first-class +per-process object, shared into both runtimes the same 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. 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 of a read 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 read's snapshot pins immutable batches, so even after the controller + releases and maintenance compacts, an in-progress walk is unaffected. +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. + +**Two changes to the shared-arrangements primitive are required for this to +hold, and the primitive does not do them yet.** Both were found by adversarial +review of `sharing.rs` and are the 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. The publisher's liveness hold must instead track the maintenance + runtime's controller-driven compaction frontier, so publishing does not hold + the trace back at all. +- **Reads must register no independent hold.** `import` and handle registration + install real holds that the publisher forwards to the maintenance trace + (`sharing.rs:169-170,229-230,435-463`), so a wedged interactive reader would + pin maintenance compaction. The integration needs a read mode that registers + no hold and relies solely on the controller's hold (criterion 1 above). Only + then is it true that a wedged interactive runtime cannot pin maintenance. + +- **`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. + +With those changes, the read side is purely a reader: it holds nothing back, the +controller remains the sole authority, and a stuck interactive thread can fall +behind on its own reads but cannot wedge maintenance compaction. Without them, +the "interactive never pins maintenance" property this design depends on is +false. ## Internal to compute, one protocol endpoint -The split is invisible to the controller. It sees one replica behind one -protocol endpoint. The replica's routing layer dispatches commands to the owning -runtime (see "Routing by target id") and merges both runtimes' responses into -one stream. The controller protocol is unchanged, and its compaction authority -is unchanged because `AllowCompaction` for a maintained id still targets the -maintenance runtime's trace. - -This is a deliberate choice over teaching the controller about two runtimes. An -explicit model (the controller routing to two sub-runtimes, `ReplicaState` -growing a sub-runtime axis at `instance.rs:3082-3097`) would enable smarter -placement and per-runtime introspection, but it is a real protocol and -controller change and it 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. That is a -follow-up, not a blocker (see open questions). +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`, `background.md`). Three -process-global assumptions must be reconciled, since they are written for "a -replica process hosts a single instance" (`compute_state.rs:495-496`): +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 must agree on it, or it must move to per-runtime state. + 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. - -## Thread allocation - -The interactive runtime runs the same number of workers as the maintenance -runtime, which is a correctness requirement for pairwise import (see "Topology") -independent of scheduling. That puts `2N` worker threads on the process's cores. -How those threads share cores (dedicated splits, oversubscription relying on the -interactive runtime being mostly blocked on reads and imports, pinning) is a -tuning question deferred to later measurement, not fixed by this design. Stage 1 -sidesteps it entirely: one runtime plus a bounded reader pool. + pattern exists, but per-runtime labels are needed to tell maintenance and + interactive metrics apart. ## Failure model @@ -300,49 +319,85 @@ 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 peek in flight when the process dies fails with the +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 deliberate. It keeps the failure model -identical to today and avoids a cross-runtime fallback path that would otherwise -need its own correctness argument. +the existing behavior. + +Shared fate is a requirement, not just a convenience: it is what lets the read +side hold nothing and rely on the controller. 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 reader's state and a half-served peek 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. +- An interactive read path that gates and walks a `SharedTraceHandle` snapshot + (a new `PendingPeek` variant or reader-pool equivalent), since today's + `handle_peek` reads only the local `TraceManager` (`compute_state.rs:673-699`). +- The three primitive changes in "Compaction": publisher-does-not-pin, no-hold + reads, and the `snapshot_at` since gate. +- 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 settled (equal, see - "Thread allocation"). How the resulting `2N` threads share cores, and whether - oversubscription is acceptable given the interactive runtime is mostly blocked - on reads, is deferred to later measurement. -- **Import queue backpressure (stage 2).** The primitive's replay queue is - bounded today only because producer and consumer share a worker step. Across - runtimes they are independently scheduled, so a lagging interactive runtime can - grow the queue without bound. A bound plus publisher backpressure is likely - needed. This is a liveness and memory concern, not a compaction one: since the - interactive runtime holds nothing back (see "Compaction reconciliation"), a - slow reader cannot wedge maintenance compaction, only fall behind on its own - imports. Captured in the primitive design. -- **Introspection and memory attribution.** Under transparency the controller - cannot see which runtime holds what memory. Shared arrangements are counted by - whoever drops last. Introspection needs a per-runtime view eventually. -- **Slow-path versus stage 1.** Stage 1 offloads fast-path index peeks but not - slow-path peeks. Is it worth teaching the coordinator to prefer the fast path - more aggressively once stage 1 lands, deferring stage 2? +- **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 primitive's replay queue is bounded today + only because producer and consumer share a worker step. Across runtimes they + are independently scheduled, so a lagging interactive runtime can grow the + queue without bound. A bound plus publisher backpressure is likely needed. + This is a liveness and memory concern, not a compaction one: with the no-hold + read mode, a slow reader falls behind on its own imports but cannot wedge + maintenance compaction. 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 can +- **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. Stage 1's offload achieves read isolation without - this coupling. + 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 already 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. + (`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 would copy data across the process - boundary, which is what persist already offers. In-process sharing is the whole + 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. -- **Do only stage 1, never stage 2.** Viable if fast-path peek isolation turns - out to cover the observed latency problem. Stage 2 is justified only if ad-hoc - temporary-dataflow rendering is shown to contend materially with maintenance. diff --git a/doc/developer/design/20260720_two_runtime_compute/background.md b/doc/developer/design/20260720_two_runtime_compute/background.md index bb2b9ccb037c4..9eab9757b0a2b 100644 --- a/doc/developer/design/20260720_two_runtime_compute/background.md +++ b/doc/developer/design/20260720_two_runtime_compute/background.md @@ -114,10 +114,10 @@ pub struct TraceBundle { 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`. Both spines are already `Arc`-backed -after the batch migration (`row-spine/src/lib.rs:51`, `typedefs.rs:107`), and -their batch contents are asserted `Send + Sync` -(`row-spine/src/lib.rs:175-177`). The remaining single-runtime coupling is +`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`. From 619251766619a267766312894e099a54423751fb Mon Sep 17 00:00:00 2001 From: Claude Date: Mon, 20 Jul 2026 19:07:20 +0000 Subject: [PATCH 6/6] design: keep change-stream import as the general path, snapshot as peek mode Revert the snapshot-only conclusion. A frozen snapshot cannot serve a subscribe, which needs the ongoing change stream, so a snapshot-only primitive would foreclose ever moving subscribes to the interactive runtime. Keep the general model: import an arrangement with an as_of and replay its change stream, with a real read hold released on drop. Peeks use a pin-free snapshot mode of the same shared trace, dataflows and future subscribes use the import mode. Reframe the compaction story accordingly. The two defects adversarial review found are (1) publishing must carry no independent compaction floor (today the publisher freezes it at publish-time since) and (2) snapshot_at must enforce the since gate. An importer holding its own as_of is intended, not a defect: it matches a stuck read today and is bounded by the controller's hold and shared fate. Drop the incorrect 'reads register no hold' requirement. Determine and document the insertion point: a publication point on the arrange stream at export_index, not an Arc-native TraceBox (which would lock the maintenance worker's hot path). Per-process registry keyed by GlobalId; two interactive read entry points (snapshot peek, import dataflow). Add a corrections note to the primitive design so the two docs agree. Co-Authored-By: Claude Fable 5 Claude-Session: https://claude.ai/code/session_01FwFQJa3HSy8oGJZ3X9uGZx --- .../README.md | 14 ++ .../20260720_two_runtime_compute/README.md | 185 +++++++++++++----- 2 files changed, 145 insertions(+), 54 deletions(-) diff --git a/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md index 6aa3cb62e9cdb..2eb44cb4f0f52 100644 --- a/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md +++ b/doc/developer/design/20260719_shared_arrangements_across_runtimes/README.md @@ -146,6 +146,20 @@ 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: diff --git a/doc/developer/design/20260720_two_runtime_compute/README.md b/doc/developer/design/20260720_two_runtime_compute/README.md index 013eaca0fe01e..7b1ffec9c569d 100644 --- a/doc/developer/design/20260720_two_runtime_compute/README.md +++ b/doc/developer/design/20260720_two_runtime_compute/README.md @@ -45,11 +45,17 @@ doing maintenance. - 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, and the read path never pins an - arrangement back beyond what the controller already holds. + 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 the read side - does not hold maintenance compaction back. + 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, @@ -90,6 +96,14 @@ 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 @@ -119,9 +133,20 @@ peek" model most directly. Either way the walk is off the maintenance threads. 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 `SharedTraceHandle::import`, 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. +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 @@ -199,16 +224,49 @@ 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`. -## Publication and the publish-versus-read handshake - -When the maintenance runtime renders an index (`export_index`, -`compute/src/render.rs:688`), it publishes the arrangement and records the -resulting `SharedTraceHandle`s in a **per-process publication registry** keyed by -`GlobalId`, one entry per worker ordinal. This registry is a new first-class -per-process object, shared into both runtimes the same way the persist client +## 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. Publication cost is one `Arc` clone plus a mutex push per batch. +`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 @@ -229,7 +287,7 @@ 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 of a read rests on three things, and only the first exists today: +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 @@ -238,40 +296,50 @@ Correctness of a read rests on three things, and only the first exists today: *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 read's snapshot pins immutable batches, so even after the controller - releases and maintenance compacts, an in-progress walk is unaffected. +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. -**Two changes to the shared-arrangements primitive are required for this to -hold, and the primitive does not do them yet.** Both were found by adversarial -review of `sharing.rs` and are the gating work for this design: +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. The publisher's liveness hold must instead track the maintenance - runtime's controller-driven compaction frontier, so publishing does not hold - the trace back at all. -- **Reads must register no independent hold.** `import` and handle registration - install real holds that the publisher forwards to the maintenance trace - (`sharing.rs:169-170,229-230,435-463`), so a wedged interactive reader would - pin maintenance compaction. The integration needs a read mode that registers - no hold and relies solely on the controller's hold (criterion 1 above). Only - then is it true that a wedged interactive runtime cannot pin maintenance. - + 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. -With those changes, the read side is purely a reader: it holds nothing back, the -controller remains the sole authority, and a stuck interactive thread can fall -behind on its own reads but cannot wedge maintenance compaction. Without them, -the "interactive never pins maintenance" property this design depends on is -false. +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 @@ -323,13 +391,15 @@ 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 lets the read -side hold nothing and rely on the controller. 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 reader's state and a half-served peek would leak, and that fallback path -would need its own correctness argument. +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 @@ -339,11 +409,14 @@ Called out explicitly, because the mechanism does not reduce to existing objects `PartitionedComputeState`). - The per-process publication registry keyed by `GlobalId`, plus the block-until-published handshake. -- An interactive read path that gates and walks a `SharedTraceHandle` snapshot - (a new `PendingPeek` variant or reader-pool equivalent), since today's - `handle_peek` reads only the local `TraceManager` (`compute_state.rs:673-699`). -- The three primitive changes in "Compaction": publisher-does-not-pin, no-hold - reads, and the `snapshot_at` since gate. +- 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 @@ -356,13 +429,17 @@ Called out explicitly, because the mechanism does not reduce to existing objects 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 primitive's replay queue is bounded today +- **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 runtime can grow the - queue without bound. A bound plus publisher backpressure is likely needed. - This is a liveness and memory concern, not a compaction one: with the no-hold - read mode, a slow reader falls behind on its own imports but cannot wedge - maintenance compaction. Captured in the primitive design. + 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