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

* issues/952: fix streaming connection closed before body consumed in orchestrated disaggregated routing

Signed-off-by: shoreline2 <shoreline2@users.noreply.github.com>

* Ensure response is released in failure scenarios

Signed-off-by: shoreline2 <shoreline2@users.noreply.github.com>

* Address pre-commit ruff check

Signed-off-by: shoreline2 <shoreline2@users.noreply.github.com>

---------

Signed-off-by: shoreline2 <shoreline2@users.noreply.github.com>
Co-authored-by: shoreline2 <shoreline2@users.noreply.github.com>
Co-authored-by: Rui Zhang <51696593+ruizhang0101@users.noreply.github.com>

Author: shoreline2

src/vllm_router/services/request_service/request.py

@@ -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()
 
         # Send to Prefill
         async with client.post(
@@ -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:
+        )
+        try:
             if decode_resp.status != 200:
                 error_text = await decode_resp.text()
                 logger.error(
@@ -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"
@@ -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(