Replace ReadaheadPipeline with anyio producer + sync iterator

The ThreadPoolExecutor-based ReadaheadPipeline and the
_PrefetchIterator wrapper around _variant_chunks_gen are gone. In
their place:

- _produce_variant_chunks: async producer running on the reader's
  BlockingPortal. An outer anyio task group manages variant-chunk
  fetches; each fetch task uses an inner task group to fan out the
  field reads concurrently. The byte-budget refill semantics
  (bootstrap chunk runs solo, subsequent chunks scheduled until the
  in-flight count exceeds readahead_bytes / per-chunk-bytes) are
  preserved. After fetching, the producer applies the variant filter,
  materialises the output dict, and sends it through a 1-buffer
  MemoryObjectStream. Telemetry — max_in_flight, last_chunk_bytes,
  the final iteration log line — is reported via a shared dict.
- _AsyncBackedIterator: sync iterator wrapping the channel via
  portal.call. close() cancels the producer task and shuts the
  channel; __del__ closes defensively. BaseExceptionGroup is unwrapped
  to a single exception so handle_exception in cli.py still surfaces
  the original ValueError.
- weakref.finalize arms close on garbage collection. Without it the
  portal's daemon thread joins on the asyncio default executor's
  non-daemon decode workers and wedges process exit when the user
  forgets to use the reader as a context manager.

CLI: --readahead-workers is removed; --io-concurrency caps concurrent
store.get calls (default 32) and --decode-threads sizes the decode
pool (default os.cpu_count()), separating IO from CPU concurrency.

Tests: TestReadaheadPipeline, TestPrefetchIteratorDirect, and the
_DepthTrackingPipeline / _make_pipeline / _shared_test_portal
helpers are deleted. A new TestVariantChunksIterator covers eager
validation, empty-fields short-circuit, exception propagation, close
cancellation, and max_in_flight semantics through the public
variant_chunks() API. The static-field-not-in-pipeline check now
monkeypatches _read_block_async.

Performance benchmarks against the four backends are deferred to
before merging.

Author: jeromekelleher

tests/test_cli.py

@@ -951,9 +951,13 @@ def test_size_param_rejects_invalid(self):
         with pytest.raises(click.UsageError):
             cli.SIZE.convert("not-a-size", None, None)
 
-    def test_make_reader_forwards_workers(self, fx_vcz_path):
-        with cli.make_reader(fx_vcz_path, readahead_workers=4) as reader:
-            assert reader._readahead_workers == 4
+    def test_make_reader_forwards_io_concurrency(self, fx_vcz_path):
+        with cli.make_reader(fx_vcz_path, io_concurrency=4) as reader:
+            assert reader._io_concurrency == 4
+
+    def test_make_reader_forwards_decode_threads(self, fx_vcz_path):
+        with cli.make_reader(fx_vcz_path, decode_threads=2) as reader:
+            assert reader._decode_threads == 2
 
     def test_make_reader_forwards_bytes(self, fx_vcz_path):
         with cli.make_reader(fx_vcz_path, readahead_bytes=1024) as reader:
@@ -983,14 +987,15 @@ def test_view_forwards_flags(self, monkeypatch, tmp_path, fx_vcz_path):
         runner = ct.CliRunner()
         result = runner.invoke(
             cli.vcztools_main,
-            f"view --no-version --readahead-workers 4 "
+            f"view --no-version --io-concurrency 4 --decode-threads 3 "
             f"--readahead-buffer-size 100M {fx_vcz_path} "
             f"-o {output_path.as_posix()}",
             catch_exceptions=False,
         )
         assert result.exit_code == 0
         assert captured == {
-            "readahead_workers": 4,
+            "io_concurrency": 4,
+            "decode_threads": 3,
             "readahead_bytes": 100 * 1024 * 1024,
         }
 
@@ -1000,13 +1005,17 @@ def test_query_forwards_flags(self, monkeypatch, tmp_path, fx_vcz_path):
         runner = ct.CliRunner()
         result = runner.invoke(
             cli.vcztools_main,
-            f"query -f '%POS\n' --readahead-workers 2 "
+            f"query -f '%POS\n' --io-concurrency 2 --decode-threads 1 "
             f"--readahead-buffer-size 1024 {fx_vcz_path} "
             f"-o {output_path.as_posix()}",
             catch_exceptions=False,
         )
         assert result.exit_code == 0
-        assert captured == {"readahead_workers": 2, "readahead_bytes": 1024}
+        assert captured == {
+            "io_concurrency": 2,
+            "decode_threads": 1,
+            "readahead_bytes": 1024,
+        }
 
     def test_view_plink_forwards_flags(self, monkeypatch, tmp_path, fx_vcz_path):
         captured = self._spy_vcz_reader_init(monkeypatch)
@@ -1015,13 +1024,14 @@ def test_view_plink_forwards_flags(self, monkeypatch, tmp_path, fx_vcz_path):
         result = runner.invoke(
             cli.vcztools_main,
             f"view-plink --max-alleles 2 -e 'CHROM==\"X\"' "
-            f"--readahead-workers 8 --readahead-buffer-size 2M "
+            f"--io-concurrency 8 --decode-threads 5 --readahead-buffer-size 2M "
             f"{fx_vcz_path} --out {out.as_posix()}",
             catch_exceptions=False,
         )
         assert result.exit_code == 0
         assert captured == {
-            "readahead_workers": 8,
+            "io_concurrency": 8,
+            "decode_threads": 5,
             "readahead_bytes": 2 * 1024 * 1024,
         }