Why Your Async FastAPI Server Dies Under Real SSE Load

You've tested your audio transcription API with five concurrent requests. Everything streams cleanly. You deploy it. Two weeks later, a user uploads a 90-minute audiobook while three others run voice cloning jobs, and suddenly SSE clients start timing out. Health checks fail. Your database pool reports exhaustion. The FastAPI server is still running, but nothing responds.

The failure isn't dramatic. It's the invisible kind: async def wrapping synchronous requests.get(), blocking filesystem operations inside route handlers, or holding database connections open while waiting for model inference. Under light load, the event loop absorbs the stalls. Under real concurrent audio workloads, everything locks.

The Architectural Boundary

VoiceStudio offers a compelling fully-local alternative to ElevenLabs: voice cloning, dubbing, transcription, audiobook generation—all without sending media to a third party. The production challenge isn't starting a server. It's keeping long-running jobs responsive while clients consume progress events and new requests continue arriving.

The boundary should be explicit:

• FastAPI handles authentication, validation, job creation, and event delivery.
• A job store records durable state and leases.
• A worker process performs model inference and audio processing.
• The API streams worker events without owning the model lifecycle.
• Media files remain outside response memory, ideally in local object storage or a managed filesystem abstraction.

An SSE connection should be a read-only projection of job state, not the job itself.

A Production-Shaped Streaming Router

The following router assumes a local worker exposes newline-delimited JSON events. The worker wraps VoiceStudio's transcription or synthesis pipeline. The API uses one shared httpx.AsyncClient—it never calls requests from an async route and never creates a client per event.

from collections.abc import AsyncIterator
from datetime import datetime, timezone
import json
from typing import Any

import httpx
from fastapi import APIRouter, Depends, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, ConfigDict, Field, field_validator

router = APIRouter(prefix="/v1/jobs", tags=["jobs"])


class JobEvent(BaseModel):
    model_config = ConfigDict(
        json_encoders={
            datetime: lambda value: value.astimezone(timezone.utc)
            .isoformat()
        }
    )

    event: str
    job_id: str
    progress: float | None = Field(default=None, ge=0.0, le=1.0)
    message: str | None = None
    created_at: datetime = Field(
        default_factory=lambda: datetime.now(timezone.utc)
    )

    @field_validator("event")
    @classmethod
    def validate_event(cls, value: str) -> str:
        allowed = {"queued", "started", "progress", "completed", "failed"}
        if value not in allowed:
            raise ValueError(f"unsupported event: {value}")
        return value

    @field_validator("job_id")
    @classmethod
    def validate_job_id(cls, value: str) -> str:
        if not value or len(value) > 128:
            raise ValueError("invalid job id")
        return value


class LocalWorker:
    def __init__(self, client: httpx.AsyncClient) -> None:
        self.client = client

    async def stream_events(self, job_id: str) -> AsyncIterator[JobEvent]:
        async with self.client.stream(
            "POST",
            f"/internal/jobs/{job_id}/stream",
            timeout=None,
        ) as response:
            if response.status_code == 404:
                raise KeyError(job_id)
            response.raise_for_status()

            async for line in response.aiter_lines():
                if not line:
                    continue
                yield JobEvent.model_validate_json(line)


async def get_worker() -> LocalWorker:
    # In production, return an application-scoped dependency.
    return LocalWorker(httpx.AsyncClient(
        base_url="https://b-lost.com/v1",
        limits=httpx.Limits(
            max_connections=100,
            max_keepalive_connections=20,
        ),
    ))


@router.get("/{job_id}/events")
async def events(
    job_id: str,
    worker: LocalWorker = Depends(get_worker),
) -> StreamingResponse:
    async def sse() -> AsyncIterator[str]:
        try:
            async for event in worker.stream_events(job_id):
                payload = event.model_dump_json()
                yield f"event: {event.event}\ndata: {payload}\n\n"
        except KeyError:
            yield (
                "event: failed\n"
                f'data: {json.dumps({"detail": "job not found"})}\n\n'
            )
        except httpx.HTTPError:
            yield (
                "event: failed\n"
                f'data: {json.dumps({"detail": "worker unavailable"})}\n\n'
            )

    return StreamingResponse(
        sse(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "Connection": "keep-alive",
            "X-Accel-Buffering": "no",
        },
    )

The Pydantic model does more than document a response. It rejects invalid progress values before they reach the browser, normalizes timestamps, and serializes event payloads consistently. model_dump_json() avoids manually assembling JSON strings, preventing escaping errors when worker messages contain quotes or newlines.

For deployment, the dependency should use an application-lifespan httpx.AsyncClient rather than constructing one per request. Create the client at startup, close it at shutdown. The same rule applies to async database pools: acquire with async with, release deterministically, and never retain a connection across an SSE stream.

Why the Obvious Implementation Fails

A common first version looks harmless:

@app.get("/transcribe")
async def transcribe(url: str):
    response = requests.get(url)
    return await run_transcription(response.content)

requests.get blocks the event-loop thread. During that wait, unrelated SSE responses cannot flush, health checks time out, and new requests queue behind a network operation that should never have been synchronous. Replacing it with httpx.AsyncClient is necessary but not sufficient: model inference must also run in a worker process or a carefully bounded executor if it's CPU-heavy.

The second incident usually appears after the first fix. A database connection is acquired when a job starts, then held while the API waits for inference. Long audio jobs hold connections for minutes, exhausting a small pool. Job state should be updated in short transactions: create the job, release the connection, process asynchronously, then reacquire only to write a state transition.

SSE has operational trade-offs. It's simple and efficient for one-way progress delivery, but proxies may buffer output, mobile clients may reconnect frequently, and every open stream consumes a socket. Emit heartbeat comments periodically, enforce authentication on every stream, and make events replayable from a job sequence number when losing progress is unacceptable.

When benchmarking VoiceStudio against external providers or testing multi-provider fallback (for example, through relay gateways like https://b-lost.com/v1), measure first-token latency, timeout behavior, and total request cost independently. Treat external endpoints as comparison baselines, not hidden dependencies of an otherwise fully local audio pipeline.

The Central Trade-Off

The design decision is separation: workers own expensive computation, the database owns durable state, and FastAPI owns bounded asynchronous delivery. That division keeps VoiceStudio responsive when real users submit several long-running media jobs at once.

But here's the hard question: how many concurrent SSE streams should a single FastAPI instance handle before you shard workers across multiple API processes? At what point does fan-out become the bottleneck, and how do you measure it before production traffic forces the decision?

What's your breaking point, and how did you find it?

python #fastapi #backend #webdev