persist: fetch source parts concurrently

[antiguru]

Motivation

Hydrating a shard with many small parts is bound by the blob-store round-trip rather than bandwidth. This lets shard_source_fetch issue several part fetches at once so the round-trips overlap.

What this does

The fetch operator stays an async timely operator. Two changes make it safe to complete fetches out of order:

• Time-keyed capability bookkeeping (not FIFO). For each time a part was minted at, the operator retains a data capability and a completed-fetches capability and counts the outstanding fetches at that time. When the count reaches zero both capabilities drop, advancing the data frontier and releasing that time's leases. Results can now complete in any order without advancing the frontier past unproduced data.
• Concurrent fetching. Up to persist_source_fetch_concurrency (default 1, i.e. the previous serial behavior) fetches run at once via a FuturesUnordered, each on a cheap per-call BatchFetcher clone that shares the schema cache. In-flight bytes stay bounded by the existing fetch semaphore inside fetch_leased_part.

On a missing blob the operator reports the error and freezes (retains all capabilities, stops draining results) exactly as before, so the frontier never advances past a part we failed to emit.

New tests cover data fetching, mid-stream shutdown, the listing- and fetch-path freezes, and the concurrent fetch and concurrent-freeze paths.

Relationship to the original PR

Recreates antiguru/materialize#181 on top of the existing async operators, intentionally omitting that PR's deasync refactor of the shard_source operators. The deasync-only follow-ups are therefore also not needed and not included:

• the txn-wal read-worker park bound (only required because deasync moved listen polling into a raw tokio task that no longer wakes the timely worker — the async listen here still activates normally), and
• the introspection-naming change (shard_source_descs_return → shard_source_descs) — the separate lease-return operator is retained, so the dataflow graph is unchanged.

---

Generated by Claude Code

[petrosagg]

Looks good overall, with one simplification in capability tracking

[antiguru]

Thanks for the reviews!