기다림을 끝낸 뒤 연결 자리도 반환되는가?

2. 검증 구조

실제 TCP를 사용하지만 모델은 없다. 테스트 서버는 요청마다 Connection: close로 응답한다. 따라서 다음 요청 성공은 같은 TCP 연결의 재사용이 아니라 같은 client 풀의 자리가 반환되어 새 연결을 열 수 있다는 증거다.

3. 입력 조건과 출력

세 조건 × 두 client 종류로 총 여섯 경우를 확인한다. deadline 조건에서는 생성·probe의 HTTP 요청 deadline을 둘 다 0.1초로, 취소·pool 조건에서는 둘 다 5초로 준다. 테스트용 값이므로 제품의 생성 300초·probe 요청당 5초 기본값을 바꾸지 않는다. 여기서 probe는 list_models 경로이며 전체 check_ready의 health→models 순서는 별도 readiness 테스트에서 확인한다.

4. 코드 해설

5. 상태 변화

7. 오류를 읽는 법

deadline 테스트에서 busy가 나오면 응답 대기 전에 풀에서 막힌 것이다. 취소 뒤 후속 요청이 막히면 연결 반환을 확인한다. closed Event가 끝나지 않으면 client가 socket을 닫았는지 점검한다. 외부의 2초 safety timeout 실패는 서비스 timeout 성공으로 해석하면 안 된다.

현재 전체 코드

"""Real socket tests: deadlines/cancellation must release the occupied pool slot."""

import asyncio
import json
from collections.abc import AsyncGenerator
from contextlib import asynccontextmanager
from dataclasses import dataclass, field

import httpx2
import pytest

from app.engines.base import GenerationRequest, InferenceError, Message
from app.engines.turbofieldfare import TurboFieldfareAdapter


MODEL = "gemma-4-26b-a4b-it"
REQUEST = GenerationRequest(messages=(Message(role="user", content="Synthetic"),))


@dataclass
class PendingRequest:
    path: str
    release: asyncio.Event = field(default_factory=asyncio.Event)
    closed: asyncio.Event = field(default_factory=asyncio.Event)


@asynccontextmanager
async def held_upstream() -> AsyncGenerator[
    tuple[str, asyncio.Queue[PendingRequest]], None,
]:
    pending: asyncio.Queue[PendingRequest] = asyncio.Queue()
    handlers: set[asyncio.Task] = set()

    async def handle(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        task = asyncio.current_task()
        assert task is not None
        handlers.add(task)
        item: PendingRequest | None = None
        waits: list[asyncio.Task] = []
        try:
            head = (await reader.readuntil(b"\r\n\r\n")).decode("ascii")
            first, *headers = head.split("\r\n")
            _, path, _ = first.split()
            size = next((int(header.split(":", 1)[1]) for header in headers
                         if header.lower().startswith("content-length:")), 0)
            await reader.readexactly(size)
            item = PendingRequest(path)
            # No response until released. A peer EOF proves client cancellation
            # closed the socket, without relying on a fabricated HTTP exception.
            waits = [asyncio.create_task(item.release.wait()),
                     asyncio.create_task(reader.read(1))]
            await pending.put(item)
            done, _ = await asyncio.wait(waits, return_when=asyncio.FIRST_COMPLETED)
            if waits[1] in done:
                assert waits[1].result() == b""
                return
            if path == "/health":
                body = {"status": "ok"}
            elif path == "/v1/models":
                body = {"object": "list", "data": [{"id": MODEL, "object": "model"}]}
            else:
                assert path == "/v1/chat/completions"
                body = {
                    "id": "chatcmpl-held", "object": "chat.completion", "model": MODEL,
                    "choices": [{"index": 0,
                                 "message": {"role": "assistant", "content": "OK"},
                                 "finish_reason": "stop"}],
                    "usage": {"prompt_tokens": 1, "completion_tokens": 1,
                              "total_tokens": 2,
                              "prompt_tokens_details": {"cached_tokens": 0}},
                }
            payload = json.dumps(body).encode()
            writer.write(
                b"HTTP/1.1 200 OK\r\nContent-Type: application/json\r\n"
                b"Connection: close\r\n"
                + f"Content-Length: {len(payload)}\r\n\r\n".encode() + payload
            )
            await writer.drain()
        except (asyncio.IncompleteReadError, ConnectionError):
            pass
        finally:
            for wait in waits:
                wait.cancel()
            await asyncio.gather(*waits, return_exceptions=True)
            writer.close()
            try:
                await writer.wait_closed()
            finally:
                if item is not None:
                    item.closed.set()
                handlers.discard(task)

    server = await asyncio.start_server(handle, "127.0.0.1", 0)
    async with server:
        try:
            yield f"http://127.0.0.1:{server.sockets[0].getsockname()[1]}", pending
        finally:
            tasks = tuple(handlers)
            for task in tasks:
                task.cancel()
            await asyncio.gather(*tasks, return_exceptions=True)


@asynccontextmanager
async def adapter_at(url: str, *, deadline: float = 5) -> AsyncGenerator[
    TurboFieldfareAdapter, None,
]:
    # Disable HTTP read timeout: only the adapter's real total deadline can end
    # a held response. One connection makes a leaked pool slot observable.
    options = {
        "base_url": url, "trust_env": False,
        "timeout": httpx2.Timeout(None, pool=0.05),
        "limits": httpx2.Limits(max_connections=1, max_keepalive_connections=1),
    }
    async with (
        httpx2.AsyncClient(**options) as generation_client,
        httpx2.AsyncClient(**options) as probe_client,
    ):
        yield TurboFieldfareAdapter(
            generation_client, probe_client=probe_client, model=MODEL,
            probe_timeout=deadline, generation_timeout=deadline,
        )


async def invoke(adapter: TurboFieldfareAdapter, kind: str):
    if kind == "generation":
        return await adapter.generate(REQUEST)
    return await adapter.list_models()


async def assert_next_request_works(adapter, pending, kind: str) -> None:
    task = asyncio.create_task(invoke(adapter, kind))
    try:
        item = await asyncio.wait_for(pending.get(), timeout=2)
        item.release.set()
        result = await asyncio.wait_for(task, timeout=2)
        if kind == "generation":
            assert result.text == "OK"
        else:
            assert result[0].id == MODEL
    finally:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)


@pytest.mark.parametrize("kind", ["generation", "probe"])
@pytest.mark.parametrize("termination", ["deadline", "cancel"])
def test_real_deadline_or_cancellation_closes_connection_and_releases_slot(
    kind: str, termination: str,
) -> None:
    async def scenario() -> None:
        async with held_upstream() as (url, pending):
            async with adapter_at(url, deadline=0.1 if termination == "deadline" else 5) as adapter:
                task = asyncio.create_task(invoke(adapter, kind))
                try:
                    item = await asyncio.wait_for(pending.get(), timeout=2)
                    assert item.path == ("/v1/chat/completions" if kind == "generation" else "/v1/models")
                    if termination == "cancel":
                        task.cancel()
                        with pytest.raises(asyncio.CancelledError):
                            await asyncio.wait_for(task, timeout=2)
                    else:
                        with pytest.raises(InferenceError) as caught:
                            await asyncio.wait_for(task, timeout=2)
                        assert caught.value.code == "timeout"
                        assert caught.value.upstream_status is None
                    await asyncio.wait_for(item.closed.wait(), timeout=2)
                    assert not item.release.is_set()
                    # Same client, same one-slot pool, a new TCP connection.
                    await assert_next_request_works(adapter, pending, kind)
                finally:
                    task.cancel()
                    await asyncio.gather(task, return_exceptions=True)

    asyncio.run(scenario())


@pytest.mark.parametrize("kind", ["generation", "probe"])
def test_real_pool_timeout_is_busy_and_does_not_cancel_active_request(kind: str) -> None:
    async def scenario() -> None:
        async with held_upstream() as (url, pending):
            async with adapter_at(url) as adapter:
                first = asyncio.create_task(invoke(adapter, kind))
                try:
                    item = await asyncio.wait_for(pending.get(), timeout=2)
                    with pytest.raises(InferenceError) as caught:
                        await asyncio.wait_for(invoke(adapter, kind), timeout=2)
                    assert caught.value.code == "busy"
                    assert caught.value.upstream_status is None
                    assert not first.done()
                    assert not item.closed.is_set()
                    assert pending.empty()  # The rejected request never reached upstream.
                    item.release.set()
                    await asyncio.wait_for(first, timeout=2)
                    await assert_next_request_works(adapter, pending, kind)
                finally:
                    first.cancel()
                    await asyncio.gather(first, return_exceptions=True)

    asyncio.run(scenario())

실제 테스트 소스와 동일한 코드다. 실행 명령: cd be 후 .venv/bin/python -m pytest -q tests/test_inference_timeouts.py.

8. 다음 파일과의 연결

lifespan 종료 검증에서 다른 경계의 보장을 확인한다. 전체 구조는 Backend 개요, client 조립은 main.py, 오류 번역은 Adapter 구현과 연결된다.