From 7d07537a87fe60ea8db4ad66df80d449afc190dc Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:13:02 +0000 Subject: [PATCH 1/2] fix(v1): recover env pool after worker exit --- verifiers/v1/serve/pool.py | 107 ++++++++++++++++++++++++++++++------- 1 file changed, 88 insertions(+), 19 deletions(-) diff --git a/verifiers/v1/serve/pool.py b/verifiers/v1/serve/pool.py index faa8b5f503..31a972078b 100644 --- a/verifiers/v1/serve/pool.py +++ b/verifiers/v1/serve/pool.py @@ -16,9 +16,8 @@ Scaling is elastic but upscale-only: a new worker is spawned when in-flight requests reach 90% of current capacity (`workers * multiplex`). Workers are spawned `spawn`-style (own env, own loop) and monitor a death pipe so an orphaned worker self-exits if the -broker dies. TODO: downscale idle workers, per-worker restart-on-death, stats/lag -monitors (v0 had them; omitted here — rollout errors are returned as data, not crashes, -so worker death is rare). +broker dies. The broker monitors each process sentinel, fails requests owned by a dead +worker, and replaces it. TODO: downscale idle workers and stats/lag monitors. """ import asyncio @@ -37,7 +36,7 @@ from verifiers.v1.env import EnvConfig from verifiers.v1.serve.server import EnvServer -from verifiers.v1.serve.types import HealthResponse, RunGroupRequest +from verifiers.v1.serve.types import BaseResponse, HealthResponse, RunGroupRequest logger = logging.getLogger(__name__) @@ -112,8 +111,8 @@ def __init__( def _worker_path(self, i: int) -> str: return f"/tmp/vf-pool-{self.session}-{i}" - def _spawn_worker(self) -> None: - i = len(self.workers) # upscale-only, so the next index is the current count + def _spawn_worker(self, index: int | None = None) -> None: + i = len(self.workers) if index is None else index address = f"ipc://{self._worker_path(i)}" parent_conn, child_conn = self._mpctx.Pipe() proc = self._mpctx.Process( @@ -136,6 +135,7 @@ def _spawn_worker(self) -> None: self.workers.append( { "process": proc, + "sentinel": proc.sentinel, "dealer": dealer, "pipe": parent_conn, "active": 0, @@ -144,6 +144,78 @@ def _spawn_worker(self) -> None: ) if self._poller is not None: self._poller.register(dealer, zmq.POLLIN) + self._poller.register(proc.sentinel, zmq.POLLIN) + + async def _drain_worker_replies( + self, worker: dict, pending: dict[bytes, dict] + ) -> int: + """Relay every queued reply and return the rollout slots it completed.""" + released = 0 + while True: + try: + request_id, data = await worker["dealer"].recv_multipart( + flags=zmq.NOBLOCK, copy=False + ) + except zmq.Again: + return released + # Copy only the routing key; relay the response Frames unchanged. + entry = pending.pop(request_id.bytes, None) + if entry is None: + continue + entry["worker"]["active"] -= entry["rollout_slots"] + released += entry["rollout_slots"] + with contextlib.suppress(zmq.ZMQError): + await self.frontend.send_multipart( + [entry["client_id"], request_id, data], copy=False + ) + + async def _worker_died(self, worker: dict, pending: dict[bytes, dict]) -> int: + """Remove a dead worker, fail its requests, and start its replacement. + + Returns the rollout slots released from the pool's in-flight count. + """ + process = worker["process"] + process.join() + logger.error( + "EnvServerPool worker %d (pid=%s) exited with code %s; restarting", + worker["index"], + process.pid, + process.exitcode, + ) + if self._poller is not None: + self._poller.unregister(worker["dealer"]) + self._poller.unregister(worker["sentinel"]) + self.workers.remove(worker) + + # The process sentinel and its final DEALER reply can become readable in + # adjacent loop iterations. Honor every reply already queued before failing + # requests that the dead worker can no longer complete. + released = await self._drain_worker_replies(worker, pending) + data = msgpack.packb( + BaseResponse( + success=False, + error=f"env server worker exited unexpectedly (exit code {process.exitcode})", + ).model_dump(mode="python"), + use_bin_type=True, + ) + for request_id, entry in list(pending.items()): + if entry["worker"] is not worker: + continue + pending.pop(request_id) + released += entry["rollout_slots"] + with contextlib.suppress(zmq.ZMQError): + await self.frontend.send_multipart( + [entry["client_id"], request_id, data] + ) + + with contextlib.suppress(Exception): + worker["pipe"].close() + worker["dealer"].close() + with contextlib.suppress(OSError): + os.unlink(self._worker_path(worker["index"])) + process.close() + self._spawn_worker(worker["index"]) + return released def _maybe_scale_up(self, in_flight: int) -> None: """Spawn one more worker when in-flight rollout slots reach 90% of capacity. @@ -187,6 +259,16 @@ async def run(self) -> None: in_flight = 0 while True: events = dict(await self._poller.poll()) + # Drain replies before handling a simultaneous process exit: data already + # handed to ZMQ is a completed request, not a worker failure. + for w in list(self.workers): + if w["dealer"] in events: + in_flight -= await self._drain_worker_replies(w, pending) + + for w in list(self.workers): + if w["sentinel"] in events: + in_flight -= await self._worker_died(w, pending) + if self.frontend in events: ( client_id, @@ -222,19 +304,6 @@ async def run(self) -> None: ) if self.elastic: self._maybe_scale_up(in_flight) - for w in self.workers: - if w["dealer"] in events: - request_id, data = await w["dealer"].recv_multipart(copy=False) - # Copy only the routing key; relay the response Frames unchanged. - entry = pending.pop(request_id.bytes, None) - if entry is None: - continue - entry["worker"]["active"] -= entry["rollout_slots"] - in_flight -= entry["rollout_slots"] - with contextlib.suppress(zmq.ZMQError): - await self.frontend.send_multipart( - [entry["client_id"], request_id, data], copy=False - ) except (asyncio.CancelledError, KeyboardInterrupt): pass finally: From 767add932250bee1548115b438ac0f17320c184f Mon Sep 17 00:00:00 2001 From: morluto <76467478+morluto@users.noreply.github.com> Date: Tue, 14 Jul 2026 03:42:40 +0000 Subject: [PATCH 2/2] fix(v1): retain dead worker through cleanup --- verifiers/v1/serve/pool.py | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/verifiers/v1/serve/pool.py b/verifiers/v1/serve/pool.py index 31a972078b..33f8b25573 100644 --- a/verifiers/v1/serve/pool.py +++ b/verifiers/v1/serve/pool.py @@ -185,7 +185,6 @@ async def _worker_died(self, worker: dict, pending: dict[bytes, dict]) -> int: if self._poller is not None: self._poller.unregister(worker["dealer"]) self._poller.unregister(worker["sentinel"]) - self.workers.remove(worker) # The process sentinel and its final DEALER reply can become readable in # adjacent loop iterations. Honor every reply already queued before failing @@ -198,23 +197,28 @@ async def _worker_died(self, worker: dict, pending: dict[bytes, dict]) -> int: ).model_dump(mode="python"), use_bin_type=True, ) + failed_responses = [] for request_id, entry in list(pending.items()): if entry["worker"] is not worker: continue pending.pop(request_id) released += entry["rollout_slots"] - with contextlib.suppress(zmq.ZMQError): - await self.frontend.send_multipart( - [entry["client_id"], request_id, data] - ) + failed_responses.append([entry["client_id"], request_id, data]) + # Keep the dead worker under shutdown's ownership until all of its resources + # are closed, then restore capacity before sending failures can suspend. with contextlib.suppress(Exception): worker["pipe"].close() worker["dealer"].close() with contextlib.suppress(OSError): os.unlink(self._worker_path(worker["index"])) process.close() + self.workers.remove(worker) self._spawn_worker(worker["index"]) + + for response in failed_responses: + with contextlib.suppress(zmq.ZMQError): + await self.frontend.send_multipart(response) return released def _maybe_scale_up(self, in_flight: int) -> None: