Share fixture workers through the launcher in tests; add launch: worker strings

Client(server_path) now accepts the DuckDB extension's launch:<argv> scheme,
selecting transport="launch" with the argv split shlex-style, so one worker
string names the same shared worker in both clients.

The test suite uses it to reach its fixture workers by default: every
fixture-worker Client goes through vgi_rpc.launcher, one warm worker per
command shared by every xdist process, instead of each process growing its own
pool (and subprocess-direct spawning a worker per test). At -n 32 on a 48-core
box that was hundreds of live vgi-fixture-worker processes.

- VGI_TEST_CLIENT_TRANSPORT=subprocess restores the per-process pools (the
  default on Windows, which has no AF_UNIX); the client_transport matrix runs
  launch + http, or the subprocess legs under that setting.
- A per-session VGI_RPC_TEST_SESSION lands in the launcher's hash domain, so a
  session never reuses a warm worker from an earlier run of older code; the
  controller stops the session's workers at unconfigure.
- The shared worker runs with --max-connections 0: at 64, scans holding
  connections while their next one waits in the backlog can deadlock.
- Tests of the subprocess transport itself (stderr capture, force-stop, worker
  debug) and the catalog DDL suite (fresh InMemoryCatalog per test) stay on
  subprocess; test_subprocess_transport.py keeps pooled and direct covered.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>

Author: rustyconover

tests/catalog/test_client_catalog.py

@@ -10,13 +10,9 @@
 see tests/catalog/test_integration.py which exercises all catalog operations.
 """
 
-import sys
-
+from tests.conftest import CATALOG_WORKER
 from vgi.client import Client
 
-# Worker command for catalog tests
-CATALOG_WORKER = f"{sys.executable} -m vgi._test_fixtures.catalog"
-
 
 class TestClientCatalogStatelessOperations:
     """Test catalog operations that work independently."""

tests/catalog/test_example_worker_catalog.py

@@ -8,6 +8,7 @@
 
 import pyarrow as pa
 
+from tests.conftest import FIXTURE_WORKER
 from vgi._test_fixtures.worker import ExampleWorker
 from vgi.catalog import (
     AttachOpaqueData,
@@ -23,7 +24,7 @@
 from vgi.client import Client
 
 # Worker command for catalog tests
-EXAMPLE_WORKER = "vgi-fixture-worker"
+EXAMPLE_WORKER = FIXTURE_WORKER
 
 
 def _get_expected_function_names() -> set[str]:

tests/conformance/test_function_inventory.py

@@ -19,6 +19,7 @@
 
 import pytest
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.catalog.catalog_interface import AttachOpaqueData, SchemaObjectType
 from vgi.client.client import Client
 
@@ -79,7 +80,7 @@ class TypeCoverage:
 @pytest.fixture(scope="module")
 def attached_example() -> tuple[str, AttachOpaqueData]:
     """Attach to the ``example`` catalog once per module and yield ``(worker, attach_opaque_data)``."""
-    worker = "vgi-fixture-worker"
+    worker = FIXTURE_WORKER
     client = Client(worker)
     result = client.catalog_attach(
         name=EXAMPLE_CATALOG_NAME,
@@ -92,7 +93,7 @@ def attached_example() -> tuple[str, AttachOpaqueData]:
 
 def test_example_catalog_is_attachable() -> None:
     """Sanity: the example worker advertises a catalog named ``example``."""
-    client = Client("vgi-fixture-worker")
+    client = Client(FIXTURE_WORKER)
     names = [c.name for c in client.catalogs()]
     assert EXAMPLE_CATALOG_NAME in names, (
         f"Example worker should advertise the {EXAMPLE_CATALOG_NAME!r} catalog; got {names!r}"
@@ -149,7 +150,7 @@ def test_fixture_worker_registers_at_least_one_per_category(
     broken — both are drift we want to catch.
     """
     _worker, attach_opaque_data = attached_example
-    client = Client("vgi-fixture-worker")
+    client = Client(FIXTURE_WORKER)
     infos = client.schema_contents(
         attach_opaque_data=attach_opaque_data,
         path=[EXAMPLE_SCHEMA_NAME],

tests/conftest.py

@@ -3,13 +3,82 @@
 """Shared fixtures for VGI tests."""
 
 import logging
+import os
+import shlex
+import shutil
+import subprocess
+import sys
+import uuid
 from typing import Any
 
 import pyarrow as pa
 import pytest
 
 from vgi import schema
 
+# =============================================================================
+# Fixture-worker transport
+# =============================================================================
+
+# How tests reach the fixture workers. ``launch`` (the default wherever AF_UNIX
+# exists) routes every fixture-worker ``Client`` through ``vgi_rpc.launcher``:
+# one warm worker per command, shared by every test process, so ``pytest -n 48``
+# talks to one worker instead of each xdist process growing its own pool (and
+# the matrix's ``subprocess-direct`` leg spawning a fresh worker per test) —
+# which put hundreds of ``vgi-fixture-worker`` processes on a 48-core box.
+# ``VGI_TEST_CLIENT_TRANSPORT=subprocess`` restores the per-process pools.
+TEST_TRANSPORT = os.environ.get("VGI_TEST_CLIENT_TRANSPORT") or ("subprocess" if sys.platform == "win32" else "launch")
+if TEST_TRANSPORT not in ("launch", "subprocess"):
+    raise pytest.UsageError(f"VGI_TEST_CLIENT_TRANSPORT must be 'launch' or 'subprocess', got {TEST_TRANSPORT!r}")
+
+# Part of the launcher's hash domain (it hashes every ``VGI_RPC_*`` variable), so
+# each session gets its own workers: never a warm one left by an earlier run of
+# older code, and never one another checkout on the machine is using. Set here,
+# before xdist spawns its workers, so every process of one session agrees on it.
+_SESSION_ENV = "VGI_RPC_TEST_SESSION"
+os.environ.setdefault(_SESSION_ENV, uuid.uuid4().hex)
+
+_LAUNCHED: list[list[str]] = []
+
+
+def _fixture_command(*argv: str) -> str:
+    """Return the ``Client`` worker string for a fixture command under ``TEST_TRANSPORT``."""
+    if TEST_TRANSPORT == "subprocess":
+        return shlex.join(argv)
+    # Every test process shares the one worker, so it must not cap connections:
+    # at the default 64 a scan holding its connections while its next one waits
+    # in the backlog can wait on the others doing the same, forever.
+    launch_argv = [shutil.which(argv[0]) or argv[0], *argv[1:], "--max-connections", "0"]
+    _LAUNCHED.append(launch_argv)
+    return "launch:" + shlex.join(launch_argv)
+
+
+#: The protocol fixture worker, and the catalog fixture worker.
+FIXTURE_WORKER = _fixture_command("vgi-fixture-worker")
+CATALOG_WORKER = _fixture_command(sys.executable, "-m", "vgi._test_fixtures.catalog")
+
+#: The protocol fixture worker as a plain subprocess, for tests of the
+#: subprocess transport itself (stderr capture, pooling).
+SUBPROCESS_FIXTURE_WORKER = "vgi-fixture-worker"
+
+
+def pytest_unconfigure(config: pytest.Config) -> None:
+    """Stop this session's launched workers rather than leaving them to idle out.
+
+    Only the controlling process does it — an xdist worker finishing early must
+    not pull the shared worker out from under the others.
+    """
+    if hasattr(config, "workerinput") or not _LAUNCHED or sys.platform == "win32":
+        return
+    from vgi_rpc.launcher import compute_hash, default_state_dir, gc_state_dir
+
+    for argv in _LAUNCHED:
+        # The worker's argv carries ``--unix <state dir>/<hash>.sock``, and the
+        # session variable makes that hash this session's alone.
+        subprocess.run(["pkill", "-TERM", "-f", f"{compute_hash(argv)}.sock"], check=False)
+    gc_state_dir(default_state_dir())
+
+
 # =============================================================================
 # Utility Functions (not fixtures, can be imported directly)
 # =============================================================================
@@ -72,8 +141,8 @@ def test_logger() -> logging.Logger:
 
 @pytest.fixture
 def fixture_worker() -> str:
-    """Return the path to the example worker."""
-    return "vgi-fixture-worker"
+    """Return the example worker, as a ``Client`` worker string for ``TEST_TRANSPORT``."""
+    return FIXTURE_WORKER
 
 
 @pytest.fixture(scope="session")
@@ -118,8 +187,11 @@ def _start(*, extra_args: list[str] | None = None, env: dict[str, str] | None =
 # Keys used to identify transport modes in conformance tests. Keeping the
 # literal values here (rather than inside the fixture) makes it easy for
 # individual tests to opt out with ``@pytest.mark.parametrize("client_transport",
-# ["subprocess-pooled"], indirect=True)``.
-_CLIENT_TRANSPORT_MODES = ["subprocess-pooled", "subprocess-direct", "http"]
+# ["subprocess-pooled"], indirect=True)``. The subprocess legs join the default
+# matrix only under ``VGI_TEST_CLIENT_TRANSPORT=subprocess``.
+_CLIENT_TRANSPORT_MODES = (
+    ["launch", "http"] if TEST_TRANSPORT == "launch" else ["subprocess-pooled", "subprocess-direct", "http"]
+)
 
 
 @pytest.fixture(scope="session")
@@ -160,7 +232,8 @@ def client_transport(
     returned client as a context manager.
 
     Modes:
-        subprocess-pooled: Pool-backed subprocess (the default path).
+        launch: The launcher-shared worker (``launch:`` worker string).
+        subprocess-pooled: Pool-backed subprocess.
         subprocess-direct: ``pool=None`` — direct Popen management.
         http: ``Client.from_http(base_url)`` backed by a per-test
             ``vgi-fixture-http`` subprocess. Skips if the HTTP transport
@@ -171,10 +244,14 @@ def client_transport(
     mode = request.param
 
     def _make() -> Client:
+        if mode == "launch":
+            if TEST_TRANSPORT != "launch":
+                pytest.skip("VGI_TEST_CLIENT_TRANSPORT=subprocess")
+            return Client(fixture_worker)
         if mode == "subprocess-pooled":
-            return Client(fixture_worker, pool=_default_pool)
+            return Client(SUBPROCESS_FIXTURE_WORKER, pool=_default_pool)
         if mode == "subprocess-direct":
-            return Client(fixture_worker, pool=None)
+            return Client(SUBPROCESS_FIXTURE_WORKER, pool=None)
         if mode == "http":
             if not _HTTP_TRANSPORT_READY:
                 pytest.skip("Client HTTP transport arrives in Phase 2 of whimsical-mccarthy plan")

tests/table/generator/test_exception_function.py

@@ -5,6 +5,7 @@
 import pyarrow as pa
 import pytest
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import Arguments
 from vgi.client import Client, ClientError
 
@@ -15,7 +16,7 @@ class TestGeneratorExceptionFunctionViaClient:
     def test_raises_exception_after_batches(self) -> None:
         """Function should raise exception after specified batches via Client."""
         with (
-            Client("vgi-fixture-worker") as client,
+            Client(FIXTURE_WORKER) as client,
             pytest.raises(ClientError) as exc_info,
         ):
             list(
@@ -31,7 +32,7 @@ def test_raises_exception_after_batches(self) -> None:
     def test_raises_exception_immediately(self) -> None:
         """Function with fail_after=0 should raise immediately via Client."""
         with (
-            Client("vgi-fixture-worker") as client,
+            Client(FIXTURE_WORKER) as client,
             pytest.raises(ClientError) as exc_info,
         ):
             list(
@@ -46,7 +47,7 @@ def test_raises_exception_immediately(self) -> None:
 
     def test_outputs_batches_before_failure(self) -> None:
         """Function should output batches before failing via Client."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             outputs: list[pa.RecordBatch] = []
             try:
                 for batch in client.table_function(

tests/table/generator/test_filter_echo_function.py

@@ -4,6 +4,7 @@
 
 import pyarrow as pa
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import Arguments
 from vgi.client import Client
 
@@ -13,7 +14,7 @@ class TestFilterEchoFunction:
 
     def test_no_filter_output(self) -> None:
         """Without filters, pushed_filters should be '(none)' for all rows."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             batches = list(
                 client.table_function(
                     function_name="filter_echo",
@@ -36,7 +37,7 @@ def test_no_filter_output(self) -> None:
 
     def test_count_argument(self) -> None:
         """Verify row count matches the count argument."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             batches = list(
                 client.table_function(
                     function_name="filter_echo",
@@ -50,7 +51,7 @@ def test_count_argument(self) -> None:
 
     def test_s_column_values(self) -> None:
         """The s column should match 'row_{n}' pattern."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             batches = list(
                 client.table_function(
                     function_name="filter_echo",

tests/table/generator/test_logging_function.py

@@ -4,6 +4,7 @@
 
 import pyarrow as pa
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import Arguments
 from vgi.client import Client
 
@@ -13,7 +14,7 @@ class TestLoggingGeneratorFunctionViaClient:
 
     def test_generates_correct_output(self) -> None:
         """Function should generate correct output via Client."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             outputs = list(
                 client.table_function(
                     function_name="logging_generator",
@@ -31,7 +32,7 @@ def test_generates_correct_output(self) -> None:
 
     def test_zero_count_produces_no_output(self) -> None:
         """Function with count=0 should produce no output via Client."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             outputs = list(
                 client.table_function(
                     function_name="logging_generator",
@@ -44,7 +45,7 @@ def test_zero_count_produces_no_output(self) -> None:
 
     def test_large_output(self) -> None:
         """Function should handle larger outputs via Client."""
-        with Client("vgi-fixture-worker") as client:
+        with Client(FIXTURE_WORKER) as client:
             outputs = list(
                 client.table_function(
                     function_name="logging_generator",

tests/table/generator/test_partitioned_function.py

@@ -8,6 +8,7 @@
 
 import pyarrow as pa
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import Arguments
 from vgi.client import Client
 
@@ -22,7 +23,7 @@ class TestPartitionedSequenceFunctionMultiWorker:
 
     def test_two_workers_produce_complete_sequence(self) -> None:
         """Two workers should together produce the complete sequence."""
-        with Client("vgi-fixture-worker", worker_limit=2) as client:
+        with Client(FIXTURE_WORKER, worker_limit=2) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",
@@ -39,7 +40,7 @@ def test_two_workers_produce_complete_sequence(self) -> None:
 
     def test_three_workers_produce_complete_sequence(self) -> None:
         """Three workers should together produce the complete sequence."""
-        with Client("vgi-fixture-worker", worker_limit=3) as client:
+        with Client(FIXTURE_WORKER, worker_limit=3) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",
@@ -56,7 +57,7 @@ def test_three_workers_produce_complete_sequence(self) -> None:
 
     def test_workers_produce_large_sequence(self) -> None:
         """Multiple workers should handle large sequences."""
-        with Client("vgi-fixture-worker", worker_limit=4) as client:
+        with Client(FIXTURE_WORKER, worker_limit=4) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",
@@ -73,7 +74,7 @@ def test_workers_produce_large_sequence(self) -> None:
 
     def test_uneven_distribution(self) -> None:
         """Workers should handle sequences that don't divide evenly."""
-        with Client("vgi-fixture-worker", worker_limit=3) as client:
+        with Client(FIXTURE_WORKER, worker_limit=3) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",
@@ -90,7 +91,7 @@ def test_uneven_distribution(self) -> None:
 
     def test_single_worker_fallback(self) -> None:
         """worker_limit=1 should work like single worker mode."""
-        with Client("vgi-fixture-worker", worker_limit=1) as client:
+        with Client(FIXTURE_WORKER, worker_limit=1) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",
@@ -107,7 +108,7 @@ def test_single_worker_fallback(self) -> None:
 
     def test_increment_with_multi_workers(self) -> None:
         """Multiple workers should handle increment parameter."""
-        with Client("vgi-fixture-worker", worker_limit=2) as client:
+        with Client(FIXTURE_WORKER, worker_limit=2) as client:
             outputs = list(
                 client.table_function(
                     function_name="partitioned_sequence",

tests/table/generator/test_projected_data_function.py

@@ -4,6 +4,7 @@
 
 import pyarrow as pa
 
+from tests.conftest import FIXTURE_WORKER
 from vgi.arguments import Arguments
 
 
@@ -14,7 +15,7 @@ def test_projection_via_client(self) -> None:
         """Projection should work correctly via Client subprocess."""
         from vgi.client import Client
 
-        with Client("vgi-fixture-worker", worker_limit=1) as client:
+        with Client(FIXTURE_WORKER, worker_limit=1) as client:
             outputs = list(
                 client.table_function(
                     function_name="projected_data",
@@ -36,7 +37,7 @@ def test_all_columns_via_client(self) -> None:
         """All columns should be returned when no projection specified."""
         from vgi.client import Client
 
-        with Client("vgi-fixture-worker", worker_limit=1) as client:
+        with Client(FIXTURE_WORKER, worker_limit=1) as client:
             outputs = list(
                 client.table_function(
                     function_name="projected_data",