perf(streaming): sub-range parallel fetch + non-blocking pool replacement

The 64-vs-128-worker benchmark pair proved a per-host throughput ceiling
(~270 MiB/s) on the HF bucket path: doubling connections exactly halved
per-connection speed (4.8 -> 2.2 MiB/s) and left the aggregate flat,
while per-episode latency doubled (5.7s -> 12s) and keep-up worsened.
Steady-state demand (148 MiB/s) is well below the ceiling; the keep-up
misses come entirely from consumer stalls (refill_wait 14-19s of ~84s):
the sim blocks the training hot path on ensure_ready() for the FIFO-head
replacement while episodes take 5.7-12s to arrive.

Two fixes:

- Non-blocking replacements: EpisodeByteCache.is_ready() (all cameras
  cached or futures done, no blocking) and the stream sim now swaps a
  replacement only when it is already resident, deferring otherwise;
  fetch capacity (~2x demand) repays the debt on later batches. A
  deferred_swaps metric is reported.
- Sub-range parallel fetch (native-http): --range-subranges N splits one
  camera GET into N concurrent sub-range GETs. Under a per-host ceiling
  this adds no bandwidth but divides per-episode latency by ~N. Keep
  workers x subranges near the ~64-connection saturation point (e.g.
  --workers 16 --range-subranges 4).

Verified: sub-range span math + order-preserving concat and is_ready
semantics (unit-level, network stubbed).

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

scripts/bench_episode_byte_cache.py

@@ -70,7 +70,20 @@ def parse_args() -> argparse.Namespace:
         help="Limit manifest construction to the first N episodes for local smoke tests.",
     )
     parser.add_argument("--pool-size", type=int, default=16)
-    parser.add_argument("--workers", type=int, default=8)
+    parser.add_argument(
+        "--workers",
+        type=int,
+        default=8,
+        help="Concurrent camera-fetch jobs. Total connections ~= workers x range-subranges; "
+        "the HF bucket path saturates around 64 connections per host, so keep the product near 64.",
+    )
+    parser.add_argument(
+        "--range-subranges",
+        type=int,
+        default=1,
+        help="Split each camera byte-range GET into N concurrent sub-range GETs (native-http only). "
+        "Divides per-episode latency by ~N under the per-host throughput ceiling.",
+    )
     parser.add_argument(
         "--native-http-connections",
         type=int,
@@ -392,10 +405,19 @@ def run_pool_stream_simulation(
     decoded_samples: list[tuple[int, float]] = []
     start = time.perf_counter()
 
+    deferred_swaps = 0
+
     def consume_ready_replacement() -> bool:
-        nonlocal refill_wait_s, replacement_count
+        nonlocal refill_wait_s, replacement_count, deferred_swaps
         if not pending:
             return False
+        # Non-blocking: only swap when the head replacement is fully resident. Blocking here
+        # stalls the training hot path on remote fetch latency (head-of-line); deferring lets
+        # the fetch pipeline (capacity ~2x demand) catch up while training continues on the
+        # current pool. The replacement debt is repaid on subsequent batches.
+        if not cache.is_ready(pending[0]):
+            deferred_swaps += 1
+            return False
         new_ep = pending.pop(0)
         wait_start = time.perf_counter()
         cache.ensure_ready(new_ep)
@@ -463,6 +485,7 @@ def run_pool_stream_simulation(
         "deadline_miss_s": deadline_miss_s,
         "replacements": float(replacement_count),
         "replacement_episodes_s": replacement_count / elapsed if elapsed > 0 else 0.0,
+        "deferred_swaps": float(deferred_swaps),
         "samples_per_episode": float(samples_per_episode),
         "prefetch_episodes": float(prefetch_episodes),
         "batch_size": float(batch_size),
@@ -722,6 +745,7 @@ def run_fetch_pool(
         native_http_connections=args.native_http_connections,
         native_http_timeout=args.native_http_timeout,
         native_http_retries=args.native_http_retries,
+        native_http_subranges=args.range_subranges,
         open_decoders=False,
     ) as cache:
         elapsed = _fill_cache(cache, episodes, progress_interval=args.progress_interval)