Skip to content

[Bugfix]: Fix streaming "connection closed" before body consumed in orchestrated disaggregated routing #960

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ruizhang0101 merged 4 commits into from
Jun 23, 2026
+8 −3
Merged

[Bugfix]: Fix streaming "connection closed" before body consumed in orchestrated disaggregated routing #960

Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
5a84c2a
issues/952: fix streaming connection closed before body consumed in o…
shoreline2 May 24, 2026
03a7aff
Ensure response is released in failure scenarios
shoreline2 May 26, 2026
b2fb1ae
Address pre-commit ruff check
shoreline2 May 31, 2026
97c8a74
Merge branch 'main' into issues/952
ruizhang0101 Jun 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 8 additions & 3 deletions src/vllm_router/services/request_service/request.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -793,7 +793,7 @@ async def route_orchestrated_disaggregated_request(

try:
# Use the shared aiohttp client from app state
client = request.app.state.aiohttp_client_wrapper()
client: aiohttp.ClientSession = request.app.state.aiohttp_client_wrapper()

@shoreline2 shoreline2 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The reason behind adding the typing for this variable is for IDE auto-complete capabilities, where a programmer can see available methods like release().


# Send to Prefill
async with client.post(
Expand Down Expand Up @@ -842,15 +842,16 @@ async def route_orchestrated_disaggregated_request(
decode_api_url = f"{decode_url}{endpoint}"
logger.info(f"[{request_id}] Sending decode request to {decode_api_url}")

async with client.post(
decode_resp = await client.post(
decode_api_url,
json=decode_request,
headers={
"Content-Type": "application/json",
"X-Request-Id": request_id,
},
timeout=aiohttp.ClientTimeout(total=600),
) as decode_resp:
)
Comment on lines +845 to +853

@gemini-code-assist gemini-code-assist Bot May 25, 2026

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

Removing the context manager is necessary to fix the premature connection closure for streaming, but it introduces a risk of leaking the aiohttp.ClientResponse object if an exception occurs before the explicit release() calls. For instance, if decode_resp.text() (line 848) or decode_resp.read() (line 880) fails, the response will not be released. Since the router uses a shared aiohttp.ClientSession, these leaks can eventually exhaust the connection pool.

Consider wrapping the response handling in a try...except block to ensure decode_resp.release() is called on failure, or use a try...finally pattern with a flag to track if ownership has been transferred to the StreamingResponse.

@shoreline2 shoreline2 May 26, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I added a try...except block to ensure release() is called in the failure case. I removed explicit calls to release() in non-failure scenarios based on the AIOHTTP documentation that suggests it's unnecessary to do so:

release() It is not required to call release on the response object. When the client fully receives the payload, the underlying connection automatically returns back to pool. If the payload is not fully read, the connection is closed

As a result of adding that try/except block, a positive outcome is the diff for this MR becomes much smaller.

try:
if decode_resp.status != 200:
error_text = await decode_resp.text()
logger.error(
Expand All @@ -870,6 +871,7 @@ async def generate_stream():
if chunk:
yield chunk
finally:
decode_resp.release()
curr_time = time.time()
logger.info(
f"[{request_id}] Orchestrated streaming request completed, total time = {curr_time - in_router_time:.4f}s"
Expand All @@ -893,6 +895,9 @@ async def generate_stream():
content=json.loads(response_data),
headers={"X-Request-Id": request_id},
)
except Exception:
decode_resp.release()
raise

except aiohttp.ClientError as e:
logger.error(
Expand Down
Loading