[Bugfix][Router] Fix leaked in-flight counters in RequestStatsMonitor

[lfsun02]

Issues

• In-flight counters leak on failure. process_request only called on_request_complete on the success path, so a backend error left the request permanently counted in in_prefill_requests / in_decoding_requests. loadaware and priority routing consume these and then treat the engine as permanently loaded.
• Per-request bookkeeping grows unbounded. request_start_time and first_token_time (one entry per request) were never removed, which causes steady memory growth for the life of the process.
• on_request_complete always decremented in_decoding_requests, even for a request that failed during prefill.

What is changed

• moved on_request_complete from the success path into process_request's finally, so it runs on every exit.
• decrement the phase the request was actually in (in_decoding_requests if it reached first token, else in_prefill_requests).
• pop request_start_time / first_token_time on completion.
• on_request_complete is now idempotent, so failover retries and the finally path can't double-count.

Tests

• pre-commit check
• unit tests

---

• Make sure the code changes pass the pre-commit checks.
• Sign-off your commit by using -s when doing git commit
• Try to classify PRs for easy understanding of the type of changes, such as [Bugfix], [Feat], and [CI].

Detailed Checklist (Click to Expand)

Thank you for your contribution to production-stack! Before submitting the pull request, please ensure the PR meets the following criteria. This helps us maintain the code quality and improve the efficiency of the review process.

PR Title and Classification

Please try to classify PRs for easy understanding of the type of changes. The PR title is prefixed appropriately to indicate the type of change. Please use one of the following:

• [Bugfix] for bug fixes.
• [CI/Build] for build or continuous integration improvements.
• [Doc] for documentation fixes and improvements.
• [Feat] for new features in the cluster (e.g., autoscaling, disaggregated prefill, etc.).
• [Router] for changes to the vllm_router (e.g., routing algorithm, router observability, etc.).
• [Misc] for PRs that do not fit the above categories. Please use this sparingly.

Note: If the PR spans more than one category, please include all relevant prefixes.

Code Quality

The PR need to meet the following code quality standards:

• Pass all linter checks. Please use pre-commit to format your code. See README.md for installation.
• The code need to be well-documented to ensure future contributors can easily understand the code.
• Please include sufficient tests to ensure the change is stay correct and robust. This includes both unit tests and integration tests.

DCO and Signed-off-by

When contributing changes to this project, you must agree to the DCO. Commits must include a Signed-off-by: header which certifies agreement with the terms of the DCO.

Using -s with git commit will automatically add this header.

What to Expect for the Reviews

We aim to address all PRs in a timely manner. If no one reviews your PR within 5 days, please @-mention one of YuhanLiu11
, Shaoting-Feng or ApostaC.

[gemini-code-assist]

Code Review

This pull request refactors process_request to ensure on_request_complete is executed within a finally block, preventing in-flight request leaks during backend errors or client disconnects. It also updates on_request_complete to properly decrement prefill counters for early failures and clean up tracking state, backed by new unit tests. The review feedback identifies a critical bug where exceptions raised before the try block but after on_new_request can still leak counters, and suggests using the passed timestamp instead of time.time() for consistent latency calculations.

[LOGO127]

AI-assisted independent validation of head 17febbb339cb725b46e2536f7d5307f0ac6c3334: the six tests in src/tests/test_request_stats.py pass locally. I found a remaining streaming-lifecycle case worth covering before relying on the client-disconnect guarantee.

The new process_request finally only runs when that inner generator is exhausted or closed. route_general_request pre-starts it with anext(stream_generator), then returns a StreamingResponse(traced_stream()). If sending response headers or a yielded body chunk is interrupted, the response can return/raise without explicitly closing that already-started inner generator.

Using the actual router, actual RequestStatsMonitor and a real loopback aiohttp backend, with discovery/selection fixtures and injected ASGI disconnect events:

• Normal completion: prefill=0, decode=0, request_start_time entries=0 (both ASGI versions).
• Interrupted header send: prefill=1, decode=0, entries=1.
• Interrupted body send: prefill=0, decode=1, entries=1.
• Result: 4 failed, 2 passed, across ASGI 2.3 (http.disconnect cancellation) and 2.4 (send raises OSError).

This measures state immediately after the ASGI response call exits and one event-loop turn, while the response remains alive. It does not establish permanent leakage after garbage collection, a real client TCP disconnect, GPU inference, or full deployed-router behavior. The backend is a finite SSE response. Python 3.12.3, FastAPI 0.128.0, Starlette 0.50.0, aiohttp 3.14.3; not a fully lockfile-synchronized environment.

Real TCP control added: With the project's declared Uvicorn 0.34.0 (advertising ASGI 2.3), I also ran the actual route through a loopback Uvicorn server and closed an actual TCP client after receiving headers, or after receiving the first SSE body chunk, while the backend awaited permission to send more. Both cases passed: prefill=0, decode=0, bookkeeping=0 after the response exited. These controls do not reproduce the send-interruption case above: cancellation reaches the inner generator while it awaits backend data and its finally runs. This is therefore a narrower send-interruption/ownership boundary, not a claim that ordinary Uvicorn client disconnects generally leak. No real socket backpressure/send-interruption reproduction yet.

Backpressure control: A third real-TCP case stopped reading after the first chunk, used a bounded backend payload and a small per-listener send buffer, observed Uvicorn's actual flow.write_paused == True, then closed the client. It also passed, with all three counters already zero at response exit (not merely after an extra event-loop turn). Thus all three real-TCP controls pass on Uvicorn 0.34.0; I have not reproduced a deployed-path leak. Please treat the injected send-interruption cases as optional lifecycle-hardening coverage, not a demonstrated production blocker.

If you choose to add explicit response-lifecycle cleanup/coverage, it should account for the header-send case too, where traced_stream has not started although process_request has. I have not opened a competing PR or changed your patch.

Standalone pytest reproducer (run with the router dependencies installed)

Save as test_disconnect_lifecycle.py, then run from the repository root:

PYTHONPATH=src pytest test_disconnect_lifecycle.py -q -s

"""Actual router/ASGI and loopback HTTP; only discovery/router selection are fixtures."""

import asyncio
import json
import time
from types import SimpleNamespace

import aiohttp
from aiohttp import web
import pytest
from starlette.requests import Request, ClientDisconnect

from vllm_router.services.request_service import request as routing
from vllm_router.stats.request_stats import RequestStatsMonitor, SingletonMeta


@pytest.mark.asyncio
@pytest.mark.parametrize("spec_version", ["2.3", "2.4"])
@pytest.mark.parametrize("disconnect", [None, "headers", "body"])
async def test_asgi_send_failure_releases_request(monkeypatch, disconnect, spec_version):
    async def backend(request):
        return web.Response(body=b'data: {"text":"hello"}\n\n', content_type="text/event-stream")

    backend_app = web.Application()
    backend_app.router.add_post("/v1/chat/completions", backend)
    server = web.AppRunner(backend_app)
    await server.setup()
    site = web.TCPSite(server, "127.0.0.1", 0)
    await site.start()
    port = site._server.sockets[0].getsockname()[1]
    url = f"http://127.0.0.1:{port}"
    SingletonMeta._instances.pop(RequestStatsMonitor, None)
    monitor = RequestStatsMonitor(sliding_window_size=10)
    endpoint = SimpleNamespace(url=url, model_names=["test"], sleep=False)
    discovery = SimpleNamespace(get_endpoint_info=lambda: [endpoint], aliases={})
    monkeypatch.setattr(routing, "get_service_discovery", lambda: discovery)
    response = None
    try:
        async with aiohttp.ClientSession() as client:
            state = SimpleNamespace(
                otel_enabled=False, semantic_cache_available=False,
                request_stats_monitor=monitor, aiohttp_client_wrapper=lambda: client,
                engine_stats_scraper=SimpleNamespace(get_engine_stats=lambda: {}),
                router=SimpleNamespace(max_instance_failover_reroute_attempts=0,
                    route_request=lambda *args: url, extract_session_id=lambda *args: None),
            )
            scope = {"type": "http", "asgi": {"version": "3.0", "spec_version": spec_version},
                     "method": "POST", "path": "/v1/chat/completions", "query_string": b"",
                     "headers": [(b"content-type", b"application/json"), (b"x-request-id", b"disconnect-probe")],
                     "app": SimpleNamespace(state=state)}
            request = Request(scope)
            request._body = json.dumps({"model": "test", "stream": True}).encode()
            response = await routing.route_general_request(request, scope["path"], None)

            disconnected = asyncio.Event()

            async def receive():
                await disconnected.wait()
                return {"type": "http.disconnect"}

            async def send(message):
                if (disconnect == "headers" and message["type"] == "http.response.start") or (
                    disconnect == "body" and message["type"] == "http.response.body"
                ):
                    if spec_version == "2.3":
                        disconnected.set()
                        await asyncio.Future()
                    raise OSError("client socket closed")

            if disconnect and spec_version == "2.4":
                with pytest.raises(ClientDisconnect):
                    await response(scope, receive, send)
            else:
                await response(scope, receive, send)
            await asyncio.sleep(0)
            stats = monitor.get_request_stats(time.time())[url]
            observed = {"spec_version": spec_version, "disconnect": disconnect, "prefill": stats.in_prefill_requests,
                        "decode": stats.in_decoding_requests,
                        "bookkeeping": len(monitor.request_start_time)}
            print(observed)
            assert observed["prefill"] == observed["decode"] == observed["bookkeeping"] == 0
    finally:
        if response is not None:
            await response.body_iterator.aclose()
        await server.cleanup()
        SingletonMeta._instances.pop(RequestStatsMonitor, None)