compute: make index peek processing cooperative

[aljoscha]

Motivation

Part 2 of 3 in a stack that makes index peeks stop monopolizing the compute
worker. Part 1 is #38039, a replica crash fix in the code this commit rewrites.

Serving a ready index peek walked the whole arrangement cursor in one call.
The compute worker is a single thread, so for a large arrangement that pins it
for the full duration of the scan: dataflows aren't scheduled, commands aren't
handled, and the peek can't even observe its own cancellation.

Part of CPU-195.

Description

A peek now scans in bounded slices. PeekScan owns the cursor, the rows
collected so far, and the size accounting, and its step spends one budget
before handing the worker back. The cursor owns the batches it reads rather
than borrowing them from the trace, so a scan is self-contained and parking one
between activations is safe. This is the same shape the peek stash path already
uses to pump rows across ticks.

Fuel is charged per cursor position, not per row returned. Counting rows would
let a selective map_filter_project over a large arrangement run arbitrarily
long without ever reaching a yield point, which is exactly the case we care
about.

Budgets nest. Each peek gets its own turn (peek_yielding), bounded by what
all peeks together may spend in one activation (peek_yielding_total). Peeks
that don't get a turn are served first on the next activation, so a long peek
can't starve the ones behind it.

A slice always advances the cursor at least once, even on a spent budget. A
yielded peek keeps the worker from parking, so a slice that did no work at all
would be a livelock rather than a slow peek. Making that a property of the loop
means a budget of work:0 degrades to a slow peek instead of hanging a worker,
rather than relying on config validation to rule the value out.

A yielded peek is work the worker owes itself, so run_client doesn't park
while any peek has work left. That is also why handle_peek no longer serves
the peek inline: process_peeks runs later in the same loop iteration, so
latency is unchanged, and routing everything through there means a burst of
peeks shares one budget instead of each getting a full one.

Cancellation of a long scan now actually works. Previously the worker could not
observe a CancelPeek while it was inside the scan.

YieldSpec moves out of the linear join into crate::yielding so both callers
share one policy type and config format. That move is pure, the linear join
behaves the same.

Metrics

Peek timings now accumulate across activations and are reported once the peek is
done. Reporting per activation would turn one slow peek into a string of fast
ones. The cost is that a peek cancelled mid-scan reports nothing, which the help
texts now say.

That accumulation removes any view of how long a peek holds the worker in one
go, which is the quantity the budgets actually bound, so
mz_peek_processing_seconds times one pass over the pending peeks.

Known limits, deliberately left

The error-trace scan stays unbudgeted. It walks every key of the errs trace,
which in practice holds a handful of rows. There is a NOTE: on it.

Cursor setup is unbudgeted, and a single unit of fuel is not a bounded amount of
work. Seeking across a run of literal constraints that match no key happens
inside one unit. So the budget bounds how often we get to yield, not the length
of any one slice. Both have NOTE:s.

Reading this diff before #38039 merges

This PR targets main, so until #38039 lands its commit shows up here too. The
extra commit is the replica crash fix, plus
test/sqllogictest/peek_result_thinning.slt. Review it there, not here. Once it
merges this branch gets rebased and the diff narrows to the two commits above.

The thinning guard's checked_mul does legitimately belong to this diff. It
comes from #38039 and is carried into the new PeekScan::absorb shape rather
than reintroduced, because the guard has to stay overflow-safe in its new home.
The rationale comment travels with it.

Verification

New unit tests in src/compute/src/yielding.rs cover the yield-spec parser and
the budget arithmetic: that a nested budget is bounded by both its own and the
shared allowance, that an unbounded budget never spends, and that a budget that
is not yet spent never hands out a zero allowance. That last one would spin the
scan loop, and writing it caught a real bug.

test/clusterd-test-driver/scripts/peek_yielding.spec is the targeted test. It
serves peeks over a 2000-row index under work:1, so the scan cannot finish in
one activation and has to resume on the order of 2000 times. It pins that no row
is lost or duplicated across a yield, that a finished scan leaves nothing behind
that corrupts the next peek, and that a peek spanning many activations still
produces exactly one response.

Two new driver capabilities were needed to write it. peek-count peeks an index
directly and reports the row count, because the existing count tallies through
an ephemeral reduce dataflow and then peeks that dataflow's single-row output,
so it walks one cursor position and cannot exercise a scan at all. And
peek-count takes optional literal constraints, which reach the other cursor
path, the one that seeks from one literal to the next rather than stepping. The
literal cases cover unsorted input, literals matching no key, ending by
exhausting the literal list rather than the cursor, and an empty list.

Broader coverage comes from running the suite with a budget below production, so
that any peek over ~1000 cursor positions resumes at least once.
peek_yielding and peek_yielding_total are wired into
get_variable_system_parameters for that, and into parallel-workload's flag
flipping.

That default is kept within an order of magnitude of production on purpose. It
reaches every mzcompose composition, and a very small value buys little extra
coverage while multiplying the timely steps a large peek needs. The randomized
runs go lower. Both parameters are also pinned to their production values in
ADDITIONAL_BENCHMARKING_SYSTEM_PARAMETERS, because the benchmarks measure
against an older image that does not know them, so a non-production value there
would only slow down one side of the comparison.

Not covered by a test. The round robin across concurrent peeks, and with it
peek_yielding_total, which only binds once more than one peek is pending. The
driver awaits each peek before sending the next, so expressing this needs a
concurrency primitive in the driver rather than another command.

[linear-code]

CPU-195

[aljoscha]

sound like we now want a metric/histogram that tracks how long an invocation of the outer process peeks is? Because before these metrics somewhat gave us that, no?

[aljoscha]

Yes, exactly right, and that was a real loss. index_peek_total_seconds was observed once per process_peek call, so its distribution effectively told you how long a single call held the worker. Accumulating it per peek gives you the peek's total cost but throws away the per-invocation view, which is the one that says whether yielding is doing its job.

Added mz_peek_processing_seconds: the duration of one process_peeks pass, i.e. all peeks served in one activation before the worker goes back to scheduling dataflows and handling commands. That is precisely the quantity peek_yielding_total bounds, so the metric and the knob now measure the same thing.

Only recorded for activations that actually had a pending peek. process_peeks runs on every worker iteration, so otherwise the histogram would be almost entirely zeroes and useless. That short-circuit also skips the Budget construction, which was doing an Instant::now() per iteration for nothing.

[aljoscha]

Recreating on the correct base so GitHub picks up the stack. Superseded, see the replacement linked below.