compute: walk fast-path index peeks off the serving worker

doc/developer/design/20260720_two_runtime_compute/peek-offload-stash-plan.md

@@ -0,0 +1,113 @@
+# Bringing the persist stash to the offloaded walk
+
+**Parked with the rest of the peek-placement work, and implemented.** This describes the gate
+that made the offload unreachable in a production configuration and how it was removed. It
+is retained because the gate is the reason two rounds of staging measurement showed nothing,
+which is worth not rediscovering. Whether the offload survives at all is
+[peek-placement.md](peek-placement.md)'s question.
+
+## Why
+
+E1 measured the offload cutting point-lookup tail latency 32-fold behind a 2.2 second scan.
+That measurement required `enable_compute_peek_response_stash = false` for the environment.
+With the stash on, which is how production runs, `should_offload_peek` declines every peek the stash could take, and the win is unreachable.
+
+The gate is one branch.
+
+```rust
+fn should_offload_peek(&self, peek_stash_usable: bool) -> bool {
+    if peek_stash_usable {
+        return false;
+    }
+```
+
+`peek_stash_usable` is a capability rather than a prediction.
+It is `finishing.is_streamable(arity) && ENABLE_PEEK_RESPONSE_STASH && location_available`, with no reference to result size, so a peek returning three rows declines the offload over a divergence that will never happen.
+The reachable domain today is peeks with an `ORDER BY` or a non-identity projection, which is not the traffic the offload was built for.
+
+## What the two mechanisms actually do
+
+They are usually described as alternatives.
+They are not, and seeing that is what makes the fix small.
+
+| | Walk runs on | Response path | Fixes |
+|---|---|---|---|
+| Peek response stash | the timely worker | persist, uploaded by an async task fed over a channel | response size |
+| Index peek offload | a blocking task | inline, returned whole | head-of-line blocking |
+
+The stash never moved the walk.
+`pump_rows` keeps stepping the row iterator on the worker thread and pushes batches into `rows_tx`, and the comment on that field says exactly why: a trace cursor is not `Send`, so the iterator cannot be handed to the upload task.
+
+That constraint is the one the offload has already solved.
+
+## The two facts that make this cheap
+
+Both are already in the tree, neither needs changing.
+
+* `StashingPeek::start_upload` takes `Box<dyn Iterator<Item = Result<(Row, NonZeroI64), String>>>`.
+  It is already independent of which trace produced the rows, because the maintenance path walks a `TraceBundle` and the interactive path a registry handle.
+* `spawn_offloaded_walk` is bounded on `TraceCursor<OksTr>: Send` and `TraceStorage<OksTr>: Send`, and `OffloadSnapshot::Ready` carries owned cursors and storage.
+  The offloaded walk's row source crosses threads by construction.
+
+So on the offload path the reason `pump_rows` exists does not apply.
+An offloaded walk can own the persist writer directly and stream into it as it walks, with no channel hop and no worker involvement at all.
+Unifying these makes the stash simpler on the path that matters, not more complex.
+
+## Plan
+
+### Phase 1: let the offloaded walk stash
+
+Pass the stash configuration into `spawn_offloaded_walk`: eligibility, threshold, persist location, and `batch_max_runs`.
+The task walks its snapshot, accumulates as it does today, and on crossing the threshold opens the upload and streams the remainder into it.
+`IndexOffloadPeek` then resolves to either a rows response or a stashed response, and `PendingPeek::IndexOffload` retires both the same way it retires a rows response now.
+
+The mid-walk divergence that `PeekStatus::UsePeekStash` expresses as a return value becomes an ordinary branch inside the task.
+Nothing has to be communicated back to the worker to make the decision, which is what forced the gate in the first place.
+
+### Phase 2: delete the gate
+
+`should_offload_peek` stops consulting `peek_stash_usable` and keeps only the in-flight cap.
+The offload becomes reachable for ordinary streamable peeks, which is where the measured win lives.
+
+Phases 1 and 2 land together behind `enable_index_peek_offload`, which is already off by default in production.
+
+### Phase 3: demote the worker-pumped stash to a fallback
+
+The inline stash still has to exist, because the offload declines in cases the stash does not.
+Those are the in-flight cap being reached, `OffloadSnapshot::NotReady`, and any trace whose cursor cannot produce a `Send` snapshot.
+This is a demotion rather than a deletion, and it should wait until E4 has been re-run, because the cap is what decides how often the fallback is taken.
+
+## What to be careful about
+
+**Compaction holds get longer.**
+An offloaded walk pins its batches for the duration of the walk.
+A stashing walk pins them for the walk plus the upload, which is seconds rather than milliseconds.
+`index_peek_offload_max_inflight` now bounds a much longer-lived hold, and its interaction with the controller read holds and the capping in the multiplexer needs checking before the cap default is trusted.
+
+**The finishing rule must not drift.**
+`is_streamable` decides whether rows can be shipped as they are produced.
+A non-streamable finishing has to accumulate and sort before anything is emitted, and the offloaded path must apply the same rule rather than assuming it can always stream.
+
+**The size limit must stay identical.**
+`max_result_size` is checked during the inline walk.
+If the offloaded walk checks it differently, a peek that errors on one substrate succeeds on the other, which is a correctness difference visible to users.
+The existing test asserting the offloaded walk returns exactly what the inline walk returns is the right place to extend.
+
+**Backpressure gets simpler, not harder.**
+Today the bounded `rows_tx` channel throttles the worker against the upload.
+With walk and upload on one task it becomes an ordinary await, and the channel and its capacity constant can go.
+
+## Alternative considered
+
+Let the offloaded walk abandon on divergence and re-run inline.
+Rejected.
+It walks the data twice, and the restart lands on the timely worker exactly when the result is large, which is precisely the case where blocking the worker hurts most.
+It would reintroduce the tail the offload exists to remove, at the worst moment.
+
+## Validation
+
+* Re-run E1 with the stash left on.
+  The flat tail has to survive, since surviving a production configuration is the entire point of the work.
+* Confirm engagement through `mz_index_peek_walks_total{substrate}` rather than inferring it from latency.
+* Extend the offloaded-equals-inline walk test to cover a result large enough to cross the stash threshold, on both cursor sources.
+* Re-run E4, because the cap now governs upload concurrency and compaction hold duration as well as walk concurrency.

doc/developer/design/20260720_two_runtime_compute/peek-placement.md

@@ -0,0 +1,101 @@
+# Where a peek's walk runs
+
+**Status: parked, deliberately.** The problem is real and M1's evidence is the strongest in
+this directory. What is not settled is which remedy to keep, and that is answerable by
+experiment on fixtures that already exist. Implementing every candidate, merging them, and
+then learning which one was needed is the expensive order.
+
+Nothing here is gated on the interactive runtime, and the interactive runtime does not need
+any of it. See [design.md](design.md) for the mechanism decomposition these solutions are
+scored against, and CPU-217 for this work's tracking issue.
+
+## Two axes, not one
+
+An earlier draft of this proposed a single `compute_peek_substrate` parameter enumerating
+placements, on the argument that a walk runs in exactly one place so the choice is
+mutually exclusive. **That argument is wrong.** Where the walk runs and whether it yields
+are independent:
+
+|  | run to completion | yields |
+|---|---|---|
+| **on the serving worker** | today's default | S1, cooperative slicing, PR #38040 |
+| **off the worker** | S5, a blocking task | coherent, unmeasured, unimplemented |
+
+A walk on a blocking task can perfectly well have yield points. It would gain a place to
+observe a cancellation request, which is S2's prerequisite, and a bound on how long one
+walk pins the batches its cursor covers. So the fourth cell is not a nonsense state that a
+config should forbid, it is a candidate nobody has tried.
+
+The single-selector shape would have made that cell unreachable, which is the failure mode
+of collapsing two questions into one name. design.md already had this right, listing S1 and
+S5 as separate solutions. Any future parameter follows the axes: placement and preemption
+are separate settings, or one setting over the cross, but not one setting over placement
+alone.
+
+## What is measured, per cell
+
+| Cell | M1, peek behind peek | M2, peek behind an activation | M8, freshness |
+|---|---|---|---|
+| on-worker, run to completion | baseline: E1 max 5783.7 ms, E11 58 of 261 slow | baseline: E12 p90 129.5 ms | baseline: E13 2340 ms peak |
+| on-worker, yielding (S1) | **predicted only** | not measured | E13 365 ms peak, light write load |
+| off-worker (S5) | E1 max 180.4 ms flat, E11 0 of 261, E8b 29151.7 to 152.4 ms | **worse**: E12 p90 148.2 ms | E13 about 101 ms |
+| off-worker, yielding | nothing | nothing | nothing |
+
+Two entries in that table decide the question and neither is filled in.
+
+**S1 on M1 is predicted, not measured.** It is the presumed default and the row with the
+most cells empty. Everything the decision rests on is an argument about its quantum.
+
+**Yielding does not rescue S5 on M2.** Worth stating because the new cell invites the
+hope. E12's regression is dispatch timing: the snapshot is taken in `process_peeks`, which
+runs only after `step_or_park` returns, and retirement costs another step. That cost is
+paid before the walk starts, so giving the walk yield points cannot recover it. S9,
+size- or residency-aware routing, is the candidate that addresses it.
+
+## The experiment that decides
+
+The cheap decisive question is whether S1 alone matches S5 on the three fixtures that
+carry S5's case. All three exist and are described in the project document "Interactive read isolation:
+experimental evaluation":
+
+* **E1**, a point lookup behind three concurrent scans at walk costs of 23, 190 and 2170 ms.
+* **E11**, the skewed point lookup, a hot key holding millions of values under open-loop
+  arrivals.
+* **E8b**, a lookup on a resident index behind swap-resident walks at matched swap depth.
+
+Decision rule, registered before running: **if S1 matches S5 within noise on all three,
+S5 has no remaining justification and should be deleted rather than merged.** If S5 wins on
+E8b alone, its case is the unattributed swap-walk duration (M11) and nothing else, which is
+a much narrower claim than the one it was built on.
+
+This needs PR #38040 and the existing fixtures. It does not need the interactive runtime,
+which is why parking costs nothing.
+
+## Why this is parked rather than dropped
+
+The evidence that *something* is needed is not in doubt. E11 is a field-reported shape, and
+E8b turned a 29.2 second victim latency into 152 ms. What is in doubt is which of four cells
+to keep, and there are two reasons not to answer that by building:
+
+The candidates are substitutes, not complements. S1 and S5 both create a preemption point
+for the same walk, so shipping both means carrying two mechanisms where measurement is
+expected to retire one. S5 already has no axis on which it is uniquely best: E2 showed it
+is not needed for the peek-tail win once the walk moves, E12 measured it behind doing
+nothing, and S1 reaches M8 as well.
+
+And a merged mechanism is harder to delete than an unmerged one. S5 is about 700 lines
+including `local_snapshot.rs`, `PendingPeek::IndexOffload` and its metric. Landing that to
+discover S1 subsumes it means removing it afterwards, from a tree where something may
+already depend on it.
+
+## Consequence for the interactive-runtime work
+
+S5's code currently rides on the interactive-runtime branch, where it is a conditional on
+top of an inline walk rather than load-bearing: interactive peeks resolve through
+`shared_index_peek_response`, and the offload branch is entered only when the parameter
+selects it. So it can come out without touching how the interactive runtime serves reads.
+
+Removing it costs the branch E1 and E11 as supporting evidence, which is correct: E2 showed
+those results belong to the walk substrate rather than to the second runtime, and the
+interactive runtime's own case is E7 and E9. A branch that keeps S5 is a branch arguing for
+two things at once, and one of them is the one we just agreed to decide by experiment.

doc/user/data/metrics.yml

@@ -1095,6 +1095,13 @@ metrics:
   help: Total time processing index peeks, from process_peek entry to response. Excluding peeks that use the peek response stash.
   source: src/compute/src/metrics.rs
   visibility: internal
+- name: mz_index_peek_walks_total
+  help: The total number of fast-path index peek walks, by the substrate that ran them.
+  labels:
+  - substrate
+  - worker_id
+  source: src/compute/src/metrics.rs
+  visibility: internal
 - name: mz_kafka_partition_offset_max
   help: High watermark offset on broker for partition
   labels:

misc/python/materialize/mzcompose/__init__.py

@@ -297,6 +297,16 @@ def get_variable_system_parameters(
             "true",
             ["true", "false"],
         ),
+        VariableSystemParameter(
+            "enable_index_peek_offload",
+            "true",
+            ["true", "false"],
+        ),
+        VariableSystemParameter(
+            "enable_compute_interactive_runtime",
+            "true",
+            ["true", "false"],
+        ),
         VariableSystemParameter(
             "enable_upsert_v2",
             "false",
@@ -584,6 +594,7 @@ def get_default_system_parameters(
 # all. Only add it in UNINTERESTING_SYSTEM_PARAMETERS if none of the above
 # apply.
 UNINTERESTING_SYSTEM_PARAMETERS = [
+    "index_peek_offload_max_inflight",
     "enable_compute_half_join2",
     "enable_mz_join_core",
     "linear_join_yielding",

misc/python/materialize/mzcompose/services/clusterd.py

@@ -39,6 +39,7 @@ def __init__(
         workers: int = 1,
         process_names: list[str] = [],
         mz_service: str = "materialized",
+        interactive_compute: bool = False,
     ) -> None:
         environment = [
             "CLUSTERD_LOG_FILTER",
@@ -78,6 +79,21 @@ def __init__(
             f"CLUSTERD_STORAGE_TIMELY_CONFIG={storage_timely_config}",
         ]
 
+        # When set, clusterd runs a second, interactive compute runtime alongside the
+        # maintenance one (see `--interactive-compute-timely-config` in
+        # `src/clusterd/src/lib.rs`). It must span the same number of Timely peers as
+        # the maintenance compute config, so it reuses `process_names`/`workers`; its
+        # addresses use a distinct port (2104) so the two runtimes don't collide.
+        ports = [2100, 2101, 6878]
+        if interactive_compute:
+            interactive_compute_timely_config = timely_config(
+                process_names, 2104, workers, DEFAULT_COMPUTE_EXERT_PROPORTIONALITY
+            )
+            environment += [
+                f"CLUSTERD_INTERACTIVE_COMPUTE_TIMELY_CONFIG={interactive_compute_timely_config}"
+            ]
+            ports += [2104]
+
         options = ["clusterd", f"--scratch-directory={scratch_directory}", *options]
 
         config: ServiceConfig = {}
@@ -106,7 +122,7 @@ def __init__(
         config.update(
             {
                 "command": options,
-                "ports": [2100, 2101, 6878],
+                "ports": ports,
                 "environment": environment,
                 "volumes": volumes or DEFAULT_MZ_VOLUMES,
                 "restart": restart,