persist: fetch source parts concurrently

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

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

* Key the operator's capability bookkeeping by the time a part was minted
  at rather than by arrival (FIFO) order. For each time we retain a data
  and a completed-fetches capability and count 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.

* Run up to `persist_source_fetch_concurrency` (default 1, i.e. the
  previous serial behavior) fetches 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.

Recreates antiguru#181 on top of the async operators,
omitting that PR's deasync refactor of the shard_source operators (and
its dependent txn-wal and introspection-naming follow-ups).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01MR6SwYEgWVmMJUGn4ZmLMX

Author: claude

src/persist-client/src/cfg.rs

@@ -312,6 +312,7 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet {
         .add(&crate::cfg::USE_CRITICAL_SINCE_CATALOG)
         .add(&crate::cfg::USE_CRITICAL_SINCE_SOURCE)
         .add(&crate::cfg::USE_CRITICAL_SINCE_SNAPSHOT)
+        .add(&crate::cfg::SOURCE_FETCH_CONCURRENCY)
         .add(&BATCH_BUILDER_MAX_OUTSTANDING_PARTS)
         .add(&COMPACTION_HEURISTIC_MIN_INPUTS)
         .add(&COMPACTION_HEURISTIC_MIN_PARTS)
@@ -489,6 +490,17 @@ pub const USE_CRITICAL_SINCE_SOURCE: Config<bool> = Config::new(
     "Use the critical since (instead of the overall since) in the Persist source.",
 );
 
+/// Maximum number of part fetches the persist source issues concurrently, per
+/// worker. Concurrent fetches amortize the blob-store round-trip, which
+/// dominates when there are many small parts (e.g. a fine-grained hydration
+/// replay). `1` keeps the previous serial behavior.
+pub const SOURCE_FETCH_CONCURRENCY: Config<usize> = Config::new(
+    "persist_source_fetch_concurrency",
+    1,
+    "Maximum number of part fetches the persist source issues concurrently per worker \
+     (1 = serial).",
+);
+
 /// Migrate snapshots to use the critical since when opening a new read handle.
 pub const USE_CRITICAL_SINCE_SNAPSHOT: Config<bool> = Config::new(
     "persist_use_critical_since_snapshot",

src/persist-client/src/fetch.rs

@@ -155,6 +155,33 @@ where
     pub(crate) _phantom: PhantomData<fn() -> (K, V, T, D)>,
 }
 
+// Hand-written (rather than derived) so cloning does not require `K: Clone`
+// etc.: every field is an `Arc` or independently `Clone`. The `schema_cache`
+// clone shares the schema-lookup maps and applier, so clones reuse cached
+// schema fetches and only duplicate a small per-clone migration memo. Used to
+// run several `fetch_leased_part` calls concurrently, each on its own clone.
+impl<K, V, T, D> Clone for BatchFetcher<K, V, T, D>
+where
+    T: Timestamp + Lattice + Codec64,
+    K: Debug + Codec,
+    V: Debug + Codec,
+    D: Monoid + Codec64 + Send + Sync,
+{
+    fn clone(&self) -> Self {
+        Self {
+            cfg: self.cfg.clone(),
+            blob: Arc::clone(&self.blob),
+            metrics: Arc::clone(&self.metrics),
+            shard_metrics: Arc::clone(&self.shard_metrics),
+            shard_id: self.shard_id.clone(),
+            read_schemas: self.read_schemas.clone(),
+            schema_cache: self.schema_cache.clone(),
+            is_transient: self.is_transient,
+            _phantom: PhantomData,
+        }
+    }
+}
+
 impl<K, V, T, D> BatchFetcher<K, V, T, D>
 where
     K: Debug + Codec,