compute: make correction buffer reads slice-proportional

[antiguru]

Motivation

MVs using the v2 correction buffer can stall during hydration when the input has to catch up through many distinct timestamps.
CorrectionV2::updates_before restructures the entire buffer on every call — Rc-wrapping every chunk, merging to the upper, rebuilding remainders, and re-merging chains for the chain invariant — so each read costs O(total buffered chunks) instead of O(drained slice), and a catch-up through T timestamps pays O(T²/chunk_capacity).
A micro-benchmark reproduces this: at 65536 timestamps (one day of 1s ticks at 16 updates/tick) a stepwise drain takes 13.5s per worker against CorrectionV1's 0.47s, with the ratio growing in T (~T^1.8).

Description

Restructures CorrectionV2 so reads only do work proportional to the drained slice, keeping the public interface, metrics, introspection logging, and dyncfgs unchanged.
CorrectionV1 remains the fallback via enable_correction_v2. Rebased on top of #36577, so chunks use the columnar chunk region storage.

• Times at or beyond the largest read upper live in a BucketChain of exponentially growing time-range buckets, each holding (time, data)-sorted chains under the chain invariant. Reads peel only the buckets below their upper; far-future updates (e.g. temporal-filter retractions) are never touched.
• Chains split at the upper at chunk granularity (Chain::split_at_time), reusing whole chunks and copying at most one straddling chunk per chain.
• Emitted-but-not-yet-cancelled updates live in a designated emitted chain outside any invariant, so reads never re-merge future chains. (We also evaluated a destructive take_before/insert_batch interface; it measured CPU-neutral because feedback cancellation forces the same merge either way, so the interface stays as-is.)
• Times below the since are advanced with a hybrid strategy, decided by a bounded count of distinct stale times: few stale times — the steady state, since the previously emitted chain was written just before the since advanced past it — use the existing Cursor::advance_by run-splitting; many stale times (a since jump, e.g. a sink restarting with an old as-of) are collapsed in one sort-and-consolidate pass. The always-bulk variant regressed the ReplicaExpiration feature benchmark by 31%; with the hybrid it is 3.3% faster than the merge base.
• BucketChain exposes bucket iterators for metrics reporting, and restore is invoked with bounded fuel per buffer operation — incomplete restoration is picked up by the next operation, so reads never stall the operator. Builds on the restore fast path from timely-util: avoid rebuilding well-formed bucket chains in restore #36897.

Benchmark results (stepwise drain with persist feedback, 16 updates per timestamp, medians, on top of #36577's columnar chunks):

T	v1	v2 before	v2 after
1024	5.2 ms	32 ms	~16 ms
4096	21 ms	104 ms	~64 ms
16384	85 ms	858 ms	270 ms

Scaling is now linear in T (was ~T^1.8); the since-jump scenario improves ~3× and inserts are unchanged. ("v2 before" measured pre-#36577; the columnar chunk storage adds a constant on this microbenchmark but does not change the asymptotics.)

Verification

New micro-benchmark cargo bench -p mz-compute --features bench --bench correction covering insert, stepwise drain with feedback, since jumps, and a temporal-filter pattern; the bench feature only gates pub visibility.
New tests assert emission equivalence with CorrectionV1 under a stepwise-drain-with-feedback workload, since-jump collapse onto the since, and that reads never observe times at or beyond their upper.
The ReplicaExpiration feature benchmark scenario was verified locally against the merge base (3.3% faster, less memory).

🤖 Generated with Claude Code