From 868124c01305828667c7f00371fcb5d89277c397 Mon Sep 17 00:00:00 2001 From: hubert-marek Date: Wed, 10 Jun 2026 21:39:01 +0000 Subject: [PATCH] fix(v1,serve): discard delivered intercepts; trim worker arenas at heartbeat cadence MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two env-worker memory fixes, measured together at ~74% peak / ~85% resting RSS reduction in a churn simulation of perplexity-shaped browser rollouts (4 concurrent, 45 turns): 1. InterceptionServer.intercepts retained every request's raw body — the full message history INCLUDING in-sandbox base64 screenshots — until rollout unregister, so a long rollout held every turn's request simultaneously (the dominant worker-memory term; the renderer's image offload rewrites a normalized COPY, so the intercept's base64 never left). forward_request now discards the intercept after delivery; the HTTP handler holds its own local reference, so delivery is unaffected. Discard is idempotent and rollout-unregister still sweeps undelivered entries. 2. Env workers had no malloc_trim anywhere (the per-batch trim is orchestrator-only), so freed arena pages ratcheted RSS to ~3x the live set. Trim at the existing 10s stats cadence via ctypes (GIL released; deliberately NOT gc.collect — full collections on fat heaps are what caused the worker heartbeat timeouts). Co-Authored-By: Claude Fable 5 (cherry picked from commit e522d36eedef03a3b92ef88f1580dea0475c43b8) (cherry picked from commit 65161288d84c7632527ebb883b45546dc6f0fd7c) --- verifiers/serve/server/env_worker.py | 17 +++++++++++++++ verifiers/v1/utils/endpoint_utils.py | 31 +++++++++++++++++++++------- 2 files changed, 40 insertions(+), 8 deletions(-) diff --git a/verifiers/serve/server/env_worker.py b/verifiers/serve/server/env_worker.py index 137714861f..0e76d6a2c0 100644 --- a/verifiers/serve/server/env_worker.py +++ b/verifiers/serve/server/env_worker.py @@ -257,9 +257,26 @@ async def send_error_response(error: str) -> None: async def stats_loop(self, interval: float = 10.0) -> None: """Loop to push worker stats to the router.""" + libc = None + try: + import ctypes + + libc = ctypes.CDLL("libc.so.6") + except OSError: + pass while True: await asyncio.sleep(interval) + # Return freed arena pages to the OS at the heartbeat cadence. + # Long multimodal rollouts churn hundreds of MB of boxed objects + # per rollout; without trim the worker's RSS ratchets to its + # high-water mark (~3x the live set, measured). ctypes calls + # release the GIL, so this never stalls the loop — unlike + # gc.collect(), which must NOT be called here (full collections + # on fat heaps are what caused worker heartbeat timeouts). + if libc is not None: + libc.malloc_trim(0) + stats = EnvWorkerStats( worker_id=self.worker_id, timestamp=time.time(), diff --git a/verifiers/v1/utils/endpoint_utils.py b/verifiers/v1/utils/endpoint_utils.py index 161d3c9a5b..09583a650a 100644 --- a/verifiers/v1/utils/endpoint_utils.py +++ b/verifiers/v1/utils/endpoint_utils.py @@ -224,6 +224,18 @@ def rollout_queue(self, rollout_key: str) -> asyncio.Queue[str]: def get_request(self, request_id: str) -> ConfigData: return cast(ConfigData, self.server.intercepts[request_id]) + def discard_request(self, request_id: str) -> None: + """Drop a delivered intercept from the server's per-request store. + + Each intercept retains the raw request body — the full message + history including in-sandbox base64 screenshots — and the server + only sweeps them at rollout unregister, so without per-delivery + discard a long browser rollout holds every turn's request body + simultaneously (~74% of env-worker memory measured). The HTTP + handler keeps its own local reference, so delivery is unaffected. + """ + self.server.intercepts.pop(request_id, None) + def request_context( self, request_id: str, request: ConfigData ) -> ModelRequestContext: @@ -450,14 +462,17 @@ async def forward_request( state._set_error(error_info(e)) raise finally: - if bool(request.get("stream")): - if request.get("protocol") != "openai_chat_completions": - raise NotImplementedError( - "Streaming interception is currently supported for OpenAI Chat Completions." - ) - await synthesize_stream(request, response, error) - else: - deliver_response(request, response, error) + try: + if bool(request.get("stream")): + if request.get("protocol") != "openai_chat_completions": + raise NotImplementedError( + "Streaming interception is currently supported for OpenAI Chat Completions." + ) + await synthesize_stream(request, response, error) + else: + deliver_response(request, response, error) + finally: + endpoint.discard_request(request_id) def normalize_endpoint_prompt(request: ConfigData) -> Messages: