Field Log · Entry

FastAPI SSE 스트리밍: Backpressure·취소·재연결 설계 (8/10)

오늘의 결론

• 브라우저용 API는 POST /runs로 실행을 만들고 GET /runs/{id}/events로 SSE를 구독하는 두 단계가 가장 다루기 쉽습니다. Native EventSource는 GET만 지원하기 때문입니다.
• SSE payload는 문자열 조각이 아니라 event type + monotonic id + versioned data 계약입니다. done과 error는 정확히 하나의 terminal event여야 합니다.
• asyncio.Queue(maxsize=N)의 await put()으로 느린 소비자의 압력을 producer까지 전달합니다. 무제한 queue는 backpressure가 아니라 지연된 메모리 장애입니다.
• HTTP 200 header가 전송된 뒤에는 status code를 500으로 바꿀 수 없습니다. Stream 중 오류는 공개 가능한 error event로 끝내고 상세 원인은 server trace에 남깁니다.
• 재연결을 지원한다고 말하려면 event를 durable하게 저장해야 합니다. Process memory만 쓴다면 Last-Event-ID를 받아도 잃어버린 구간을 복구할 수 없습니다.

앞 글에서는 session event와 idempotency result를 PostgreSQL에 영속화했습니다. 이번에는 그 run을 FastAPI로 열고, 토큰·근거·완료를 순서 있고 취소 가능한 stream으로 전달합니다.

이 글의 대상과 학습 시간

• 대상: RAG Agent 응답을 HTTP streaming API로 제공하려는 Python backend 개발자
• 선수 지식: async generator, HTTP request/response, 87편의 run·event 개념
• 빠르게 읽기: 약 14분
• 코드와 장애 test까지 따라 하기: 약 65분
• 산출물: SSE event contract, bounded producer/consumer, disconnect·resume policy

1. Streaming은 출력 형식이 아니라 lifecycle 계약이다

다음 endpoint는 demo에서는 동작합니다.

@app.get("/answer")
async def answer():
    return StreamingResponse(llm_tokens(), media_type="text/event-stream")

그러나 production 질문에는 답하지 못합니다.

• Client가 끊기면 LLM·retrieval task도 멈추는가?
• Producer가 network보다 빠르면 token은 어디에 쌓이는가?
• Citation은 어떤 answer span과 연결되는가?
• 중간 오류를 client가 정상 종료와 어떻게 구분하는가?
• 재접속한 client는 어디부터 이어 받는가?
• 같은 POST가 재시도되면 새 run이 생기는가?
• Proxy buffering과 idle timeout은 어떻게 확인하는가?

Streaming API의 핵심은 yield가 아니라 순서·종료·취소·재생에 관한 protocol입니다.

2. SSE를 선택하는 조건

SSE(Server-Sent Events)는 server에서 client로 UTF-8 text event를 지속 전송합니다.

SSE가 잘 맞는 경우

• Client는 질문 한 번을 보내고 server event를 계속 받음
• Token delta, stage progress, citation처럼 server → client 방향이 중심
• HTTP infrastructure와 browser reconnect semantics를 활용하고 싶음
• Message가 text/JSON이며 binary frame이 필요 없음

WebSocket이 더 맞는 경우

• 한 연결에서 client와 server가 계속 양방향 message를 교환
• Human approval, voice frame, collaborative cursor처럼 low-latency upstream event가 많음
• 독립적인 subprotocol과 connection state를 운영할 준비가 됨

SSE가 WebSocket보다 항상 단순한 것은 아닙니다. 다만 RAG 답변 한 건의 진행 상황을 server에서 밀어주는 문제에는 HTTP semantics를 유지할 수 있다는 장점이 있습니다.

3. Browser API는 POST와 GET을 분리한다

FastAPI server 자체는 POST response에도 SSE를 보낼 수 있습니다. Python SDK나 fetch() 기반 parser라면 다음 형태도 가능합니다.

POST /v1/answers:stream
Content-Type: application/json
Accept: text/event-stream

하지만 browser의 native EventSource constructor는 URL에 GET을 보냅니다. Request body와 임의 authorization header도 직접 지정할 수 없습니다. 따라서 browser 제품에서는 다음 두 단계가 더 명확합니다.

POST /v1/runs
Idempotency-Key: 01J...

→ 202 Accepted
  {"run_id":"run_123", "events_url":"/v1/runs/run_123/events"}

GET /v1/runs/run_123/events
Accept: text/event-stream
Last-Event-ID: run_123:17   # reconnect일 때 browser가 전송

이 구조에는 세 가지 이점이 있습니다.

1. Run 생성의 validation·authorization·idempotency 오류를 stream 시작 전 HTTP 4xx로 반환
2. Event 구독이 끊겨도 durable run의 실행 정책을 독립적으로 결정
3. 한 run에 재연결하거나 승인된 여러 observer를 붙일 수 있음

Cookie 인증을 쓴다면 CSRF 정책을 함께 검토합니다. Query string access token은 access log·history·referrer에 남을 수 있어 기본 선택으로 두지 않습니다.

4. Event type을 먼저 고정한다

Token 문자열만 보내면 client는 검색 시작, citation 확정, 정상 종료를 추측해야 합니다. 최소 계약을 명시합니다.

Event	의미	반복 가능	Terminal
accepted	Run ID·protocol version 확정	1	아니요
retrieval	검색 단계의 공개 가능한 progress	여러 번	아니요
delta	Answer text의 append-only 조각	여러 번	아니요
citation	Evidence와 answer span의 연결	여러 번	아니요
done	최종 usage·termination·answer hash	1	예
error	공개 error code·retryability	1	예

불변식은 다음과 같습니다.

• Event ID는 run 안에서 단조 증가합니다.
• done 또는 error 뒤에는 event가 없습니다.
• delta는 기본적으로 append-only이며 중간 조각을 삭제하거나 재배열하지 않습니다.
• Client는 모르는 event type을 무시할 수 있고 protocol major version 불일치는 거부합니다.
• Citation은 display URL만 믿지 않고 85편의 evidence ID·source span을 참조합니다.

Pydantic payload

from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field


class DeltaData(BaseModel):
    model_config = ConfigDict(extra="forbid")
    text: str = Field(min_length=1, max_length=8_192)


class CitationData(BaseModel):
    model_config = ConfigDict(extra="forbid")
    evidence_id: str
    answer_start: int = Field(ge=0)
    answer_end: int = Field(gt=0)


class DoneData(BaseModel):
    model_config = ConfigDict(extra="forbid")
    termination: Literal["completed", "max_steps", "policy_stop"]
    answer_sha256: str
    input_tokens: int = Field(ge=0)
    output_tokens: int = Field(ge=0)


class ErrorData(BaseModel):
    model_config = ConfigDict(extra="forbid")
    code: Literal[
        "upstream_timeout", "rate_limited", "protocol_error",
        "policy_denied", "internal_error",
    ]
    retryable: bool
    message: str = Field(max_length=300)


class StreamEvent(BaseModel):
    model_config = ConfigDict(extra="forbid")
    run_id: str
    sequence: int = Field(ge=1)
    type: Literal["accepted", "retrieval", "delta", "citation", "done", "error"]
    schema_version: Literal[1] = 1
    data: dict

    @property
    def event_id(self) -> str:
        return f"{self.run_id}:{self.sequence}"

    @property
    def terminal(self) -> bool:
        return self.type in {"done", "error"}

실제 구현에서는 data도 discriminated union으로 검증합니다. 여기서는 SSE envelope에 집중하려고 축약했습니다.

5. SSE wire format을 이해한다

SSE event는 빈 줄로 구분된 field 묶음입니다.

id: run_123:18
event: delta
data: {"run_id":"run_123","sequence":18,"text":"검색 결과를"}

id: run_123:19
event: citation
data: {"evidence_id":"ev_07","answer_start":0,"answer_end":6}

id는 resume cursor이고 event는 browser listener 이름입니다. 여러 줄 payload를 손으로 이어 붙이기보다 framework encoder를 사용해 newline·JSON escaping 실수를 줄입니다.

FastAPI 0.135.0부터 native SSE 지원이 추가되었습니다. 해당 API를 쓸 때는 fastapi.sse.EventSourceResponse와 ServerSentEvent를 사용할 수 있습니다. 프로젝트가 이전 version이라면 dependency를 올리고 회귀 test를 하거나, 검증된 별도 SSE response 구현을 선택해야 합니다.

6. Run 생성은 stream 밖에서 끝낸다

from typing import Annotated

from fastapi import Depends, FastAPI, Header, status
from pydantic import BaseModel, Field

app = FastAPI()


class CreateRunRequest(BaseModel):
    session_id: str
    query: str = Field(min_length=1, max_length=8_000)


class CreateRunResponse(BaseModel):
    run_id: str
    events_url: str
    replayed: bool


@app.post("/v1/runs", status_code=status.HTTP_202_ACCEPTED)
async def create_run(
    body: CreateRunRequest,
    idempotency_key: Annotated[str, Header(alias="Idempotency-Key")],
    principal: Principal = Depends(require_principal),
) -> CreateRunResponse:
    # Authorization, quota, request hash validation은 HTTP header 전송 전에 끝낸다.
    result = await run_service.create_or_replay(
        tenant_id=principal.tenant_id,
        principal_id=principal.id,
        body=body,
        idempotency_key=idempotency_key,
    )
    return CreateRunResponse(
        run_id=result.run_id,
        events_url=f"/v1/runs/{result.run_id}/events",
        replayed=result.replayed,
    )

87편의 canonical request hash를 재사용합니다. 같은 key·같은 request는 같은 run_id를, 같은 key·다른 request는 409를 반환해야 합니다.

7. Bounded queue가 backpressure 경계다

Application producer와 network consumer의 속도는 다릅니다. asyncio.Queue()의 기본 maxsize=0은 무제한입니다. 느린 client가 여러 개면 token·citation object가 heap에 계속 쌓일 수 있습니다.

import asyncio
from collections.abc import AsyncIterator
from contextlib import suppress


class StreamProtocolError(RuntimeError):
    pass


async def produce_events(
    queue: asyncio.Queue[StreamEvent],
    *,
    run_id: str,
) -> None:
    saw_terminal = False
    try:
        async for event in agent_runtime.stream(run_id=run_id):
            if saw_terminal:
                raise StreamProtocolError("event emitted after terminal")

            # Queue가 가득 차면 producer가 여기서 기다린다.
            await queue.put(event)
            saw_terminal = event.terminal

        if not saw_terminal:
            raise StreamProtocolError("stream ended without terminal event")
    except asyncio.CancelledError:
        # Upstream HTTP client와 tool task가 취소를 관찰하게 반드시 다시 raise한다.
        raise
    except Exception as exc:
        if not saw_terminal:
            await queue.put(public_error_event(run_id=run_id, exc=exc))


async def consume_queue(
    *,
    run_id: str,
    request: Request,
) -> AsyncIterator[StreamEvent]:
    queue: asyncio.Queue[StreamEvent] = asyncio.Queue(maxsize=64)  # [예시값]
    producer = asyncio.create_task(
        produce_events(queue, run_id=run_id),
        name=f"agent-producer:{run_id}",
    )
    try:
        while True:
            if await request.is_disconnected():
                break

            try:
                event = await asyncio.wait_for(queue.get(), timeout=0.5)
            except TimeoutError:
                if producer.done():
                    # producer exception이나 terminal 누락을 관찰한다.
                    await producer
                    raise StreamProtocolError("producer ended without queued terminal")
                continue

            yield event
            if event.terminal:
                break
    finally:
        producer.cancel()
        with suppress(asyncio.CancelledError):
            await producer

maxsize=64는 정답이 아니라 설명용 시작점입니다. Event 크기, 동시 stream 수, client 처리 속도, memory budget을 측정해 정합니다.

왜 token을 drop하면 안 되는가

Metric sample은 손실 허용 정책을 둘 수 있지만 answer delta를 하나 버리면 최종 문장이 달라집니다. Citation을 버리면 근거 표시도 깨집니다.

• delta, citation, done, error: drop 금지
• 중복 progress heartbeat: coalesce 가능
• Queue가 계속 막힘: overall deadline이나 slow-consumer policy로 명시적으로 종료

Silent drop 대신 관찰 가능한 실패를 선택합니다.

8. FastAPI native SSE endpoint

from collections.abc import AsyncIterator
from typing import Annotated

from fastapi import Depends, Header, HTTPException, Request
from fastapi.sse import EventSourceResponse, ServerSentEvent


@app.get(
    "/v1/runs/{run_id}/events",
    response_class=EventSourceResponse,
)
async def stream_run_events(
    run_id: str,
    request: Request,
    last_event_id: Annotated[
        str | None,
        Header(alias="Last-Event-ID"),
    ] = None,
    principal: Principal = Depends(require_principal),
) -> AsyncIterator[ServerSentEvent]:
    # 이 검사는 generator 첫 yield 전에 끝나야 HTTP 404/403으로 응답할 수 있다.
    run = await run_service.get_authorized(
        tenant_id=principal.tenant_id,
        principal_id=principal.id,
        run_id=run_id,
    )
    if run is None:
        raise HTTPException(status_code=404, detail="run_not_found")

    cursor = parse_and_validate_cursor(last_event_id, expected_run_id=run_id)

    async for event in event_service.replay_then_tail(
        tenant_id=principal.tenant_id,
        run_id=run_id,
        after_sequence=cursor.sequence if cursor else 0,
        request=request,
    ):
        yield ServerSentEvent(
            event=event.type,
            id=event.event_id,
            data=event.model_dump(mode="json"),
            retry=3_000 if event.type == "accepted" else None,  # [예시값]
        )

FastAPI native response는 SSE framing, JSON data encoding, keep-alive ping과 관련 response header를 처리합니다. 그래도 reverse proxy와 CDN이 buffering하지 않는지는 실제 배포 경로에서 확인해야 합니다.

Cursor parsing은 엄격하게 한다

from dataclasses import dataclass


@dataclass(frozen=True)
class EventCursor:
    run_id: str
    sequence: int


def parse_and_validate_cursor(raw: str | None, *, expected_run_id: str) -> EventCursor | None:
    if raw is None:
        return None
    try:
        run_id, sequence_text = raw.rsplit(":", maxsplit=1)
        sequence = int(sequence_text)
    except (ValueError, TypeError) as exc:
        raise HTTPException(status_code=400, detail="invalid_last_event_id") from exc
    if run_id != expected_run_id or sequence < 0:
        raise HTTPException(status_code=400, detail="invalid_last_event_id")
    return EventCursor(run_id=run_id, sequence=sequence)

다른 run의 cursor를 섞거나 음수·과도하게 긴 값을 허용하지 않습니다. Tenant authorization을 통과한 뒤 cursor가 해당 run에 속하는지도 검증합니다.

9. Disconnect 정책은 run 종류에 따라 다르다

Client disconnect를 발견했을 때 무조건 Agent를 취소하는 것도, 무조건 계속 실행하는 것도 정답이 아닙니다.

Ephemeral interactive run

• 구독자가 사라지면 가치가 없음
• SSE consumer의 finally에서 producer task cancel
• Cancellation을 retrieval·LLM HTTP request·tool child task까지 전파
• Session에 cancelled_by_disconnect termination 기록

Durable background run

• Client가 다시 접속해 결과를 받을 수 있음
• Disconnect는 subscriber만 제거하고 Agent worker는 계속 실행
• Event는 DB에 먼저 append한 뒤 live subscriber에 알림
• 별도 POST /runs/{id}/cancel과 authorization으로 run 취소

POST create → GET subscribe 구조는 durable policy와 잘 맞습니다. 앞의 in-process consume_queue 예시는 ephemeral stream의 취소 전파를 보여준 것이며, durable run에서는 producer task의 소유자가 HTTP handler가 아니라 worker입니다.

Structured concurrency 규칙

• Child task를 fire-and-forget하지 않음
• CancelledError를 일반 exception으로 바꾸지 않음
• Tool마다 timeout이 있어도 run 전체 deadline을 공유
• Cleanup에도 짧은 timeout을 두고 connection pool을 반환
• Cancellation reason을 metric과 event에 기록

10. HTTP 200 이후 오류는 event로 끝낸다

첫 byte가 전송되면 response status와 header는 사실상 확정됩니다. 그 뒤 provider timeout이 발생해도 HTTP 500으로 되돌릴 수 없습니다.

event: error
id: run_123:24
data: {
  "code":"upstream_timeout",
  "retryable":true,
  "message":"답변 생성 시간이 제한을 초과했습니다."
}

Client는 EOF만 보고 성공으로 처리하면 안 됩니다.

let terminalSeen = false;

source.addEventListener("done", (event) => {
  terminalSeen = true;
  source.close();
});

source.addEventListener("error", (event) => {
  // Native EventSource의 network error와 named SSE error event를 구분해 감싼다.
  const payload = JSON.parse((event as MessageEvent).data);
  terminalSeen = true;
  source.close();
  showPublicError(payload);
});

Named event: error와 browser EventSource.onerror가 이름상 헷갈릴 수 있습니다. 제품 contract에서 run_error처럼 별도 event name을 쓰는 것도 좋습니다.

Server log에는 exception type, provider request ID, trace ID를 남기되 API에는 stack trace·prompt·credential·raw provider body를 보내지 않습니다.

11. 진짜 resume에는 durable event log가 필요하다

Last-Event-ID header를 읽는 것만으로 resume가 완성되지는 않습니다.

Process-memory stream

• Worker 재시작 시 과거 event 소실
• Load balancer가 다른 worker로 보내면 cursor를 알 수 없음
• 지원 가능한 계약: “새 run 시작” 또는 명시적 resume_not_supported

Durable replay stream

87편의 agent_events에 SSE 공개 event를 sequence 순서로 저장합니다.

SELECT sequence, event_type, payload
FROM agent_events
WHERE tenant_id = $1
  AND session_id = $2
  AND sequence > $3
ORDER BY sequence ASC
LIMIT 500;

안전한 replay_then_tail은 database를 source of truth로 사용합니다.

1. Cursor와 tenant/run ownership 검증
2. Cursor 이후 persisted event를 순서대로 replay
3. Live notification을 받으면 DB를 다시 query
4. 이미 보낸 sequence 이하는 deduplicate
5. Terminal event를 보내면 connection 종료

PostgreSQL LISTEN/NOTIFY나 Redis Pub/Sub은 “새 event가 생겼다”는 wake-up hint로 쓸 수 있습니다. Message 자체의 durability는 event table이 담당합니다. Notification과 query 사이 race가 있어도 monotonic sequence 재조회로 gap을 채웁니다.

Retention이 cursor보다 앞서면

Event retention으로 cursor 이후 일부가 삭제됐다면 조용히 최신 event부터 보내면 안 됩니다. 410 Gone 또는 versioned history_expired 오류로 client가 snapshot을 다시 받도록 합니다.

12. Network backpressure와 application backpressure는 다르다

Uvicorn은 transport write buffer가 high-water mark에 도달하면 response writer를 기다리게 하는 flow control을 제공합니다. 하지만 이것만으로 application 내부의 다음 항목이 자동 제한되지는 않습니다.

• LLM SDK callback이 별도 task에 쌓는 token
• Broker subscriber buffer
• 여러 stage가 동시에 emit하는 progress event
• DB write 앞 in-memory batch

따라서 각 비동기 경계에 capacity와 overflow policy가 있어야 합니다.

LLM socket → parser → agent event queue → durable append → subscriber queue → ASGI send
             bounded      bounded/batched                 bounded

관찰할 metric은 queue의 평균 길이만이 아닙니다.

• stream_queue_depth{stage}
• stream_queue_put_wait_seconds
• sse_first_event_seconds
• sse_last_event_seconds
• sse_disconnect_total{policy}
• sse_slow_consumer_total
• sse_replay_events_total
• run_cancel_propagation_seconds

Queue put wait가 길어지는데 CPU는 낮다면 slow consumer나 downstream network가 병목일 수 있습니다.

13. Heartbeat와 proxy를 실제 경로에서 검증한다

FastAPI native SSE는 idle connection에 주기적 ping을 보낼 수 있고 기본 동작은 문서 기준 15초입니다. Ping은 application progress가 아닙니다. Client UI에서 “검색 중”으로 해석하지 않습니다.

확인할 항목:

• Response Content-Type: text/event-stream
• Cache 방지 header
• Proxy buffering 비활성화 여부
• Load balancer·CDN idle timeout보다 짧은 heartbeat
• Compression이 작은 event를 과도하게 buffering하지 않는지
• HTTP/2·HTTP/1.1 환경의 connection limit
• Deploy drain 시 새 run 거부와 기존 stream 종료 정책

Framework가 X-Accel-Buffering: no를 설정해도 모든 proxy가 이를 존중하는 것은 아닙니다. Localhost test만으로 production streaming을 검증했다고 보지 않습니다.

14. Test는 느림·취소·terminal을 강제로 만든다

Event contract test

import pytest


@pytest.mark.asyncio
async def test_stream_has_monotonic_ids_and_one_terminal():
    events = [event async for event in fake_runtime.stream("run_test")]

    assert [event.sequence for event in events] == list(range(1, len(events) + 1))
    terminals = [event for event in events if event.terminal]
    assert len(terminals) == 1
    assert events[-1] == terminals[0]

Bounded queue test

@pytest.mark.asyncio
async def test_slow_consumer_blocks_producer():
    queue: asyncio.Queue[int] = asyncio.Queue(maxsize=1)
    await queue.put(1)

    blocked_put = asyncio.create_task(queue.put(2))
    await asyncio.sleep(0)
    assert not blocked_put.done()

    assert await queue.get() == 1
    await asyncio.wait_for(blocked_put, timeout=0.1)
    assert await queue.get() == 2

반드시 포함할 장애 case

• Client가 첫 delta 직후 disconnect
• Queue가 가득 찬 상태에서 run deadline 도달
• Retrieval 성공 후 LLM timeout
• 일부 delta 뒤 provider protocol 오류
• Terminal 없이 upstream generator 종료
• done 뒤 잘못된 추가 event
• 다른 run ID의 Last-Event-ID
• 이미 retention된 cursor
• Replay 중 live event가 도착하는 race
• Duplicate POST가 같은 run_id를 반환

ASGI in-process test client가 response를 buffering할 수 있으므로 실제 socket을 쓰는 integration test도 한 개 둡니다. Uvicorn을 ephemeral port로 띄우고 첫 event latency, disconnect, proxy와 같은 wire behavior를 확인합니다.

15. 자주 실패하는 구현

1. asyncio.Queue()를 maxsize 없이 만든다

느린 client가 memory 사용량을 결정합니다. Capacity와 종료 정책을 둡니다.

2. 모든 exception을 잡고 EOF로 끝낸다

Client는 성공과 실패를 구분하지 못합니다. 공개 error terminal을 보냅니다.

3. Disconnect 뒤 Agent task를 orphan으로 남긴다

Ephemeral run이면 child task를 cancel·await합니다. Durable run이면 ownership을 worker로 명확히 옮깁니다.

4. Last-Event-ID만 읽고 resume를 지원한다고 말한다

Event log가 process와 함께 사라지면 replay가 아닙니다.

5. Native EventSource로 POST body를 보낼 수 있다고 가정한다

Browser에서는 POST 생성과 GET 구독을 분리하거나 fetch() streaming parser를 사용합니다.

6. HTTP 200 이후 status를 바꾸려 한다

Stream 중 오류는 terminal event입니다. 인증·입력 validation은 첫 yield 전에 끝냅니다.

7. Token delta를 lossy queue로 처리한다

한 조각 손실이 answer corruption입니다. Drop 가능한 progress와 drop 불가 content를 구분합니다.

직접 적용 체크리스트

• Browser용 run 생성 POST와 SSE 구독 GET의 역할이 분리돼 있다.
• Event type·schema version·monotonic ID·terminal 규칙이 문서화돼 있다.
• 모든 application queue에 maxsize와 overflow policy가 있다.
• delta·citation·terminal event를 조용히 drop하지 않는다.
• Stream 전 오류는 HTTP status, stream 후 오류는 terminal event로 보낸다.
• Ephemeral·durable run의 disconnect 정책이 다르다.
• Cancellation을 child task와 upstream request까지 전파하고 await한다.
• Resume를 제공하면 durable event log와 retention 오류가 있다.
• Cursor에 tenant/run ownership과 sequence validation이 있다.
• Slow client·disconnect·replay race를 실제 socket으로 test한다.
• Proxy buffering·idle timeout·deploy drain을 배포 경로에서 확인한다.

스스로 확인하기

1. StreamingResponse만 썼을 때 terminal protocol이 자동으로 생기지 않는 이유는 무엇인가?
2. Native browser EventSource와 POST request body가 충돌하는 지점은 무엇인가?
3. Uvicorn flow control이 있어도 bounded application queue가 필요한 이유는 무엇인가?
4. Durable run에서 disconnect가 곧 run cancellation이 아닌 이유는 무엇인가?
5. Last-Event-ID 이후 event를 정확히 replay하려면 어떤 저장소 불변식이 필요한가?

자주 묻는 질문

Q1. Token마다 DB에 INSERT해야 하나요?

반드시 그렇지는 않습니다. 짧은 interval이나 byte 기준으로 delta를 batch하되 event ID와 answer hash가 재현되게 설계할 수 있습니다. Citation·tool receipt·terminal처럼 의미 있는 boundary는 개별 event로 두는 편이 audit에 유리합니다.

Q2. SSE event를 gzip으로 압축하면 안 되나요?

가능 여부보다 buffering 영향을 먼저 봅니다. Proxy나 compressor가 작은 chunk를 모으면 first-token latency가 나빠질 수 있으므로 실제 경로에서 측정합니다.

Q3. Client가 매우 느리면 영원히 기다려야 하나요?

아닙니다. Queue put wait·run deadline·subscriber lag 기준으로 명시적 slow-consumer 종료 정책을 둡니다. Durable run이면 client는 나중에 replay할 수 있습니다.

Q4. 여러 subscriber가 한 run을 볼 수 있나요?

가능하지만 authorization과 fan-out memory budget이 필요합니다. 각 subscriber queue를 분리하고 한 느린 observer가 다른 observer나 producer를 막지 않게 합니다.

Q5. SSE retry 값이 application retry도 수행하나요?

아닙니다. Browser의 reconnect delay hint입니다. Run 생성 idempotency, failed run 재실행, provider retry는 별도 정책입니다.

마무리

Production streaming 경계는 다음 다섯 계약으로 요약됩니다.

1. POST idempotency로 하나의 durable run 생성
2. Versioned event와 정확히 하나의 terminal
3. Bounded queue로 producer와 consumer 속도 연결
4. Run 종류에 맞는 disconnect·cancellation ownership
5. Durable sequence log를 기반으로 한 replay

다음 글에서는 provider와 vector DB 없이도 빠르게 실패를 재현하도록 fake·contract test·record/replay·fault injection으로 RAG Agent test pyramid를 만듭니다.

검증 정보

• 글의 성격: FastAPI·Starlette·Uvicorn 공식 문서 해설 + 작성자 streaming protocol 설계 제안 + 가상 Python 예시
• 마지막 검증일: 2026-07-16
• 검증 기준: FastAPI 0.135.0+ native SSE API, Starlette Request.is_disconnected(), Uvicorn write flow control
• 실행 주의: 실제 dependency version을 pin하고 ServerSentEvent 인자·proxy header 동작을 선택한 FastAPI version에서 회귀 test
• 수치 표기: queue 64, polling 0.5초, retry 3초는 설명용이며 event 크기·동시성·SLO로 재조정
• 경험 표기: run ID·endpoint·metric은 실제 운영 정보가 아닌 가상 설계 예시

참고자료

• FastAPI Server-Sent Events
• FastAPI StreamingResponse
• Starlette Request: is_disconnected
• Uvicorn Server Behavior: Flow Control
• MDN Using server-sent events
• HTML Living Standard: Server-sent events
• Python asyncio Queue