Streaming & Protections

Version: 2.0.0 Status: SOURCE OF TRUTH Last Updated: 2026-06-28

---

Overview

Streaming endpoints have different characteristics than RPC/REST:

• High throughput: File uploads/downloads, telemetry

• Long-lived connections: WebSocket streams

• Minimal processing: Direct I/O, no heavy middleware

Implementation status. Most of the protections described in this document (rate limiting, bandwidth throttling, connection limits, upload size/timeout limits) and the streaming HTTP response classes (StreamingResponse, FileResponse) are NOT implemented in the current codebase. They are kept here as a roadmap and clearly flagged. The sections that are backed by code today are: app mounting / workload separation, the single-body Response, and the WebSocket API. Each section below states which case it falls under.

---

Workload Separation (implemented)

Type	Characteristics	Paths
Business Logic	Complex, auth, DB, validation	/api/*, /ws/rpc/*
Streaming	High throughput, minimal processing	/stream/*, /ws/raw/*

Apps with different workloads are mounted as separate instances. Each mounted app owns its own middleware chain, so a streaming app can be configured with a minimal chain while a business app uses the full one.

Mounting: use bare segment names, not URL prefixes

AsgiServer.mount(name, app) stores the literal name as the key in server.apps — it does not strip slashes (server.py:284-288):

def mount(self, name: str, app: Any) -> None:
    if name in self.apps:
        raise ValueError(f"mount {name!r} is already taken")
    app.mount_name = name
    self.attach_instance(app)
    self.apps[name] = app

The dispatcher resolves an incoming request by taking the slash-stripped first path segment and looking it up in server.apps (dispatcher.py:61-66, dispatcher.py:82-83):

def _resolve_mount(self, path: str) -> str:
    stripped = path.strip("/")
    if "/" in stripped:
        return stripped.split("/", 1)[0]
    return stripped
# ...
mount = self._resolve_mount(path)          # "/api/foo" -> "api"
app = self.server.apps.get(mount) or self.server.apps.get("")

Consequence: a leading slash in the mount name can never match. Mounting "/api" stores the key "/api", but a request to /api/foo resolves to the segment "api", which is not in apps — so it falls back to the empty-mount app ("") or raises HTTPNotFound.

Correct usage — bare segment names:

server = AsgiServer()
server.mount("api", BusinessApp())      # matches /api/...
server.mount("stream", StreamingApp())  # matches /stream/...
server.mount("", RootApp())             # fallback for unclaimed paths

Wrong usage — leading slash never matches:

server.mount("/api", BusinessApp())     # key "/api"; request /api/foo
                                        # resolves "api" -> no match

---

HTTP responses today: single-body Response (implemented)

The only HTTP response type that ships today is Response. Its ASGI interface sends exactly one http.response.start followed by one http.response.body message — the full body in a single chunk, with no more_body flag (response.py:202-225):

async def __call__(self, scope, receive, send):
    await send({
        "type": "http.response.start",
        "status": self.status_code,
        "headers": self._build_headers(),
    })
    await send({
        "type": "http.response.body",
        "body": self.body,
    })

There is no chunked/streamed HTTP body generation in the current code. Large payloads are materialised fully in memory.

---

Streaming Protections (roadmap — NOT implemented)

The protection matrix, the per-app limit configuration, and all the limiter helper classes in this section are roadmap only. No rate-limiting, bandwidth-throttling, connection-limiting, or upload size/timeout middleware exists in src/genro_asgi/. The middleware that ships today is: auth, session, cors, cache, compression, logging, errors (see middleware/). The code blocks below are illustrative designs, not callable API.

Protection Matrix (roadmap)

Protection	HTTP Upload	HTTP Download	WebSocket
Max body size	Configurable	N/A	Per-message limit
Timeout	Read timeout	Send timeout	Idle timeout
Rate limit	Requests/sec	Bandwidth	Messages/sec
Connection limit	Per-IP	Per-IP	Per-user

Configuration Example (roadmap — illustrative, no such class)

# ROADMAP: StreamingApp and these keyword arguments do not exist yet.
streaming_app = StreamingApp(
    max_upload_size=100 * 1024 * 1024,   # 100MB
    upload_timeout=300,                   # 5 min
    download_timeout=600,                 # 10 min
    chunk_size=64 * 1024,                 # 64KB chunks
    ws_max_message_size=1 * 1024 * 1024,  # 1MB per message
    ws_idle_timeout=60,                   # 60s idle disconnect
    ws_max_connections_per_user=10,
    rate_limit_requests=100,              # per minute
    rate_limit_bandwidth=10 * 1024 * 1024,  # 10MB/s
)

---

DoS Protections (roadmap — NOT implemented)

None of the helpers below ship today. The request body is read in full and exposed as request.body (request.py:508-510); there is no built-in incremental read with per-chunk timeout or a max-size guard. These are reference designs for the protections we intend to add.

Slowloris Attack (roadmap)

Attack: Client sends data very slowly to exhaust connections.

Protection (planned): Read timeout kills slow-sending clients.

# ROADMAP: not wired into the request pipeline today.
async def receive_body(receive, timeout=30):
    body = []
    while True:
        try:
            message = await asyncio.wait_for(receive(), timeout=timeout)
        except asyncio.TimeoutError:
            raise HTTPException(408, "Request Timeout")

        body.append(message.get("body", b""))
        if not message.get("more_body", False):
            break
    return b"".join(body)

Large Payload Attack (roadmap)

Attack: Client sends huge payload to exhaust memory.

Protection (planned): Max body size enforced before processing.

# ROADMAP: no max-body enforcement exists; request.body buffers the whole body.
async def receive_body_limited(receive, max_size=10 * 1024 * 1024):
    body = []
    total = 0
    while True:
        message = await receive()
        chunk = message.get("body", b"")
        total += len(chunk)
        if total > max_size:
            raise HTTPException(413, "Payload Too Large")
        body.append(chunk)
        if not message.get("more_body", False):
            break
    return b"".join(body)

Connection Exhaustion (roadmap)

Attack: Client opens many connections to exhaust server resources.

Protection (planned): Per-IP/per-user connection limits.

# ROADMAP: no ConnectionLimiter is present in the codebase.
class ConnectionLimiter:
    def __init__(self, max_per_ip=100):
        self.max_per_ip = max_per_ip
        self._connections: dict[str, int] = {}

    def acquire(self, client_ip: str) -> bool:
        count = self._connections.get(client_ip, 0)
        if count >= self.max_per_ip:
            return False
        self._connections[client_ip] = count + 1
        return True

    def release(self, client_ip: str) -> None:
        if client_ip in self._connections:
            self._connections[client_ip] -= 1
            if self._connections[client_ip] <= 0:
                del self._connections[client_ip]

Bandwidth Abuse (roadmap)

Attack: Client downloads at maximum speed to exhaust bandwidth.

Protection (planned): Rate limiting on throughput.

# ROADMAP: no BandwidthLimiter is present in the codebase.
class BandwidthLimiter:
    def __init__(self, bytes_per_second=10 * 1024 * 1024):
        self.bps = bytes_per_second
        self._last_check = time.monotonic()
        self._bytes_sent = 0

    async def throttle(self, chunk_size: int) -> None:
        self._bytes_sent += chunk_size
        elapsed = time.monotonic() - self._last_check

        if elapsed >= 1.0:
            self._bytes_sent = chunk_size
            self._last_check = time.monotonic()
        elif self._bytes_sent > self.bps:
            sleep_time = 1.0 - elapsed
            await asyncio.sleep(sleep_time)
            self._bytes_sent = 0
            self._last_check = time.monotonic()

---

StreamingResponse (roadmap — NOT implemented)

There is no StreamingResponse class in the codebase. The only HTTP response type is the single-body Response (see above). The snippet below describes the intended future API.

# ROADMAP: StreamingResponse does not exist yet.
async def generate_report():
    for i in range(1000):
        yield f"Line {i}\n".encode()
        await asyncio.sleep(0.01)

response = StreamingResponse(
    generate_report(),
    media_type="text/plain",
)

---

FileResponse (roadmap — NOT implemented)

There is no FileResponse class in the codebase. File downloads must be served by reading the file and returning a Response. The snippet below describes the intended future API.

# ROADMAP: FileResponse does not exist yet.
response = FileResponse(
    path="/data/large_file.zip",
    filename="download.zip",
    media_type="application/zip",
)

Planned features:

• Async file reading

• Chunked transfer

• Content-Length header

• Content-Disposition header

---

WebSocket Streaming (implemented)

The WebSocket class (websocket.py) ships today. A handler is constructed with the ASGI triple and driven through accept(), the receive_* / send_* methods, the iter_text() / iter_bytes() async iterators, and close(). __aiter__ is an alias for iter_text(), so iterating a WebSocket yields text messages (websocket.py:476-515).

Fire-and-Forget (Telemetry In)

async for over a WebSocket yields decoded text frames:

async def telemetry_handler(scope, receive, send):
    websocket = WebSocket(scope, receive, send)
    await websocket.accept()
    async for text in websocket:          # iter_text(): yields str
        await process_telemetry(text)

For binary telemetry use iter_bytes():

async def telemetry_handler(scope, receive, send):
    websocket = WebSocket(scope, receive, send)
    await websocket.accept()
    async for chunk in websocket.iter_bytes():
        await process_telemetry(chunk)

Notifications (Fire-and-Forget Out)

async def notification_handler(scope, receive, send):
    websocket = WebSocket(scope, receive, send)
    await websocket.accept()
    async for event in event_stream:
        await websocket.send_json(event)

Idle Timeout (pattern using the implemented API)

ws_idle_timeout is not a built-in config knob, but the timeout itself is trivial to express with asyncio.wait_for around the implemented receive_text() / close():

async def ws_handler(scope, receive, send, idle_timeout=60):
    websocket = WebSocket(scope, receive, send)
    await websocket.accept()
    while True:
        try:
            message = await asyncio.wait_for(
                websocket.receive_text(),
                timeout=idle_timeout,
            )
            await process(message)
        except asyncio.TimeoutError:
            await websocket.close(code=1000, reason="Idle timeout")
            break

For WSX (WebSocket eXtended) message handling, the framework already builds the WebSocket, calls accept(), and runs the message loop for you in wsx/handler.py (handler.py:86-87); the manual construction above applies to raw WebSocket handlers.

---

Architecture Diagram

Today, workload separation is achieved purely by mounting separate app instances, each with its own middleware chain. The limiter boxes are roadmap.

┌─────────────────────────────────────────────────────────────────┐
│  AsgiServer                                                       │
│                                                                  │
│  ┌───────────────────────────────────────────────┐              │
│  │ mount "api"  (Business App)                    │              │
│  │ full chain: auth → session → ... (per-app)     │              │
│  └───────────────────────────────────────────────┘              │
│                                                                  │
│  ┌───────────────────────────────────────────────┐              │
│  │ mount "stream"  (Streaming App)                │              │
│  │ minimal chain (per-app)                        │              │
│  │ [ROADMAP] ConnectionLimiter → BandwidthLimiter │              │
│  └───────────────────────────────────────────────┘              │
│                                                                  │
│  Dispatcher: first path segment -> apps[segment] or apps[""]     │
└─────────────────────────────────────────────────────────────────┘

---

Benefits

1. Performance: Heavy logic doesn’t slow streaming (separate per-app chains)

2. Isolation: Streaming issues don’t impact business logic (isolated instances)

3. Flexibility: Different middleware chains per workload

4. Security (roadmap): Appropriate protections per endpoint type once the limiter middleware lands

---

Copyright: Softwell S.r.l. (2025) License: Apache License 2.0