chore: peel off indexer launcher follow-up

Signed-off-by: PeaBrane <yanrpei@gmail.com>

Author: PeaBrane

.github/filters.yaml

@@ -88,7 +88,6 @@ core:
   - 'lib/**'
   - 'tests/**'
   - 'components/src/dynamo/router/**'
-  - 'components/src/dynamo/indexer/**'
   - 'components/src/dynamo/mocker/**'
   - 'components/src/dynamo/frontend/**'
   - 'components/src/dynamo/common/**'
@@ -158,7 +157,6 @@ frontend:
   - 'container/deps/*'
   - 'container/compliance/**'
   - 'components/src/dynamo/router/**'
-  - 'components/src/dynamo/indexer/**'
   - 'components/src/dynamo/mocker/**'
   - 'components/src/dynamo/frontend/**'
   - 'components/src/dynamo/common/**'

components/src/dynamo/indexer/__init__.py

@@ -21,7 +21,6 @@
     send_request_with_retry,
     verify_response_timing,
     wait_for_frontend_ready,
-    wait_for_indexer_workers_active,
     wait_for_workers_ready,
 )
 from tests.router.router_process import FrontendRouterProcess, KVRouterProcess
@@ -981,9 +980,6 @@ async def send_requests_to_router(router, num_requests, router_name, endpoint):
         # so ZMQ sockets connect before recovery, avoiding the slow-joiner problem.
         if standalone_indexer_b_url:
             engine_workers.launch_indexer()
-            await wait_for_indexer_workers_active(
-                standalone_indexer_b_url, engine_workers.worker_id_to_zmq_ports
-            )
             logger.info(
                 f"Launched Indexer B at {standalone_indexer_b_url} "
                 f"(P2P recovery from Indexer A)"

components/src/dynamo/indexer/__main__.py

@@ -7,7 +7,6 @@
 import os
 import random
 import string
-import sys
 from typing import Any, Optional
 
 import aiohttp
@@ -32,11 +31,6 @@ def generate_random_suffix() -> str:
     return "".join(random.choices(string.ascii_lowercase, k=10))  # noqa: S311
 
 
-def get_kv_indexer_command() -> list[str]:
-    """Return the preferred standalone indexer command for the current Python env."""
-    return [sys.executable, "-m", "dynamo.indexer"]
-
-
 def assert_event_dumps_equal(
     expected: list[dict],
     actual: list[dict],
@@ -309,70 +303,6 @@ async def wait_for_workers_ready(
     return sorted(instance_ids)
 
 
-async def wait_for_indexer_workers_active(
-    indexer_url: str,
-    expected_workers: dict[int, dict[int, str]],
-    timeout_s: float = 30.0,
-) -> None:
-    """Wait until the standalone indexer reports all ZMQ listeners as active."""
-    if not expected_workers:
-        return
-
-    loop = asyncio.get_running_loop()
-    deadline = loop.time() + timeout_s
-    workers_url = f"{indexer_url}/workers"
-
-    async with aiohttp.ClientSession() as session:
-        while loop.time() < deadline:
-            try:
-                async with session.get(workers_url) as resp:
-                    if resp.status != 200:
-                        await asyncio.sleep(0.5)
-                        continue
-                    workers = await resp.json()
-            except aiohttp.ClientError:
-                await asyncio.sleep(0.5)
-                continue
-
-            workers_by_id = {
-                worker["instance_id"]: worker
-                for worker in workers
-                if worker.get("source") == "zmq"
-            }
-
-            all_active = True
-            for worker_id, endpoints in expected_workers.items():
-                worker = workers_by_id.get(worker_id)
-                if worker is None:
-                    all_active = False
-                    break
-
-                listeners = worker.get("listeners", {})
-                for dp_rank, endpoint in endpoints.items():
-                    listener = listeners.get(str(dp_rank))
-                    if listener is None:
-                        all_active = False
-                        break
-                    if listener.get("endpoint") != endpoint:
-                        all_active = False
-                        break
-                    if listener.get("status") != "active":
-                        all_active = False
-                        break
-
-                if not all_active:
-                    break
-
-            if all_active:
-                return
-
-            await asyncio.sleep(0.5)
-
-    raise RuntimeError(
-        f"Timed out waiting for indexer listeners to become active at {workers_url}"
-    )
-
-
 async def send_request_with_retry(url: str, payload: dict, max_retries: int = 8):
     """Send a single request with exponential backoff retry"""
     wait_time = 1  # Start with 1 second

components/src/dynamo/indexer/main.py

@@ -32,12 +32,7 @@
     _test_router_query_instance_id,
     _test_router_two_routers,
 )
-from tests.router.helper import (
-    generate_random_suffix,
-    get_kv_indexer_command,
-    get_runtime,
-    wait_for_indexer_workers_active,
-)
+from tests.router.helper import generate_random_suffix, get_runtime
 from tests.utils.constants import ROUTER_MODEL_NAME
 from tests.utils.managed_process import ManagedProcess
 from tests.utils.port_utils import allocate_ports, deallocate_ports
@@ -310,7 +305,15 @@ def __enter__(self):
             # Launch the standalone indexer binary
             block_size = self._mocker_args_orig.get("block_size", BLOCK_SIZE)
             indexer_cmd = [
-                *get_kv_indexer_command(),
+                "cargo",
+                "run",
+                "-p",
+                "dynamo-kv-router",
+                "--features",
+                "indexer-bin,test-endpoints",
+                "--bin",
+                "dynamo-kv-indexer",
+                "--",
                 "--block-size",
                 str(block_size),
                 "--port",
@@ -342,7 +345,7 @@ async def launch_mockers_with_indexer(self, endpoint):
         For each mocker:
         1. Launch a mocker process with --num-workers 1
         2. Poll endpoint.client().instance_ids() until a new worker_id appears
-        3. POST /register to the indexer with the worker_id and its ZMQ addresses
+        3. POST /workers to the indexer with the worker_id and its ZMQ addresses
 
         Args:
             endpoint: The dynamo endpoint object to discover worker IDs.
@@ -408,7 +411,7 @@ async def launch_mockers_with_indexer(self, endpoint):
                 for dp_rank in range(dp_size):
                     port = self._zmq_kv_events_ports[i * dp_size + dp_rank]
                     endpoint = f"tcp://127.0.0.1:{port}"
-                    zmq_addresses[dp_rank] = endpoint
+                    zmq_addresses[str(dp_rank)] = endpoint
 
                     payload = {
                         "instance_id": new_worker_id,
@@ -437,9 +440,6 @@ async def launch_mockers_with_indexer(self, endpoint):
                 f"zmq_addresses={zmq_addresses}"
             )
 
-        await wait_for_indexer_workers_active(
-            self.standalone_indexer_url, self.worker_id_to_zmq_ports
-        )
         logger.info(
             f"All {self.num_workers} mockers launched and registered with indexer"
         )
@@ -461,12 +461,20 @@ def launch_indexer(self):
         # Build --workers arg: "worker_id:dp_rank=zmq_addr,..."
         worker_entries = []
         for worker_id, zmq_addresses in self.worker_id_to_zmq_ports.items():
-            for dp_rank, zmq_endpoint in zmq_addresses.items():
-                worker_entries.append(f"{worker_id}:{dp_rank}={zmq_endpoint}")
+            for dp_rank_str, zmq_endpoint in zmq_addresses.items():
+                worker_entries.append(f"{worker_id}:{dp_rank_str}={zmq_endpoint}")
         workers_arg = ",".join(worker_entries)
 
         indexer_b_cmd = [
-            *get_kv_indexer_command(),
+            "cargo",
+            "run",
+            "-p",
+            "dynamo-kv-router",
+            "--features",
+            "indexer-bin,test-endpoints",
+            "--bin",
+            "dynamo-kv-indexer",
+            "--",
             "--block-size",
             str(block_size),
             "--port",

tests/router/common.py

@@ -68,15 +68,7 @@ def terminate_process_tree(
         return
 
     # 1. Snapshot children before signaling parent
-    try:
-        children = parent.children(recursive=True)
-    except (PermissionError, psutil.AccessDenied) as exc:
-        logger.warning(
-            "Unable to enumerate child processes for PID %s; falling back to parent-only termination: %s",
-            pid,
-            exc,
-        )
-        children = []
+    children = parent.children(recursive=True)
 
     # 2. Terminate parent first (graceful)
     terminate_process(parent, logger, immediate_kill=immediate_kill)
@@ -540,12 +532,6 @@ def _terminate_process_group(self, timeout: float = 8.0):
                         all_pgids.add(os.getpgid(child.pid))
                     except (ProcessLookupError, OSError):
                         pass
-            except (PermissionError, psutil.AccessDenied) as exc:
-                self._logger.warning(
-                    "Unable to enumerate child processes for PID %s; falling back to the main process group only: %s",
-                    self.proc.pid,
-                    exc,
-                )
             except psutil.NoSuchProcess:
                 pass
 
@@ -890,13 +876,6 @@ def subprocesses(self) -> list[psutil.Process]:
         try:
             parent = psutil.Process(self.proc.pid)
             return parent.children(recursive=True)
-        except (PermissionError, psutil.AccessDenied) as exc:
-            self._logger.warning(
-                "Unable to enumerate child processes for PID %s: %s",
-                self.proc.pid,
-                exc,
-            )
-            return []
         except psutil.NoSuchProcess:
             return []