mgf-livepush 0.1.2

Realtime push pipeline for mgf-common consumers — SSE streaming, a pluggable pub/sub broker seam (Redis + mock), and per-key connection-slot caps. Sibling of mgf-common under the mgf.* namespace.

mgf-livepush

Realtime push pipeline for mgf-common consumers: an SSE streaming primitive, a pluggable pub/sub broker seam (Redis adapter + an in-process mock), and a per-key connection-slot cap. A sibling of mgf-common under the mgf.* namespace.

Extracted from PlasmaMapper's Phase-4 push surface after the pre-release core re-design hardened it (heartbeat cadence, slot lifecycle, disconnect cleanup, and the dev/prod-proxy buffering recipe below). The broker seam means a consumer can swap Redis pub/sub for Redis Streams / NATS without touching the SSE machinery.

Install

pip install "mgf-livepush[redis,fastapi]"   # adapters are opt-in extras

Quickstart (FastAPI)

from redis.asyncio import Redis
from mgf.livepush import SSEStreamService, RedisBroker, SlotCap
from mgf.livepush.fastapi import sse_streaming_response

redis = Redis.from_url("redis://localhost:6379/0")
sse = SSEStreamService(RedisBroker(redis), SlotCap(redis, cap=20))

@router.get("/events/stream")
async def stream(request: Request):
    # 429 (Retry-After: 5) automatically when the per-key cap is full.
    return await sse_streaming_response(
        sse, request, channel=f"events:{tenant_id}", slot_key=str(tenant_id)
    )

Publisher side, anywhere:

await RedisBroker(redis).publish(f"events:{tenant_id}", json.dumps(event))

Framework-agnostic core: open_stream(...) returns an SSEStream (.headers + .body async byte iterator) — wrap .body in whatever streaming response your framework uses. Tests use MockBroker (no Redis).

The dev/prod-proxy SSE recipe (read this — it's the tarpit)

A FastAPI text/event-stream endpoint behind a proxy (SvelteKit vite, a +server.ts [...path] forward, nginx, AWS ALB) will silently buffer the body and break your latency budget even though the API side is correct. To get live push through a proxy:

1. Response headers (set by this lib): Cache-Control: no-cache, no-transform + X-Accel-Buffering: no. If your proxy gzips, also force Content-Encoding: identity on the proxied response.
2. Node fetch forwarding of a streamed body needs duplex: 'half'.
3. Playwright: waitForLoadState('networkidle') never fires with an open EventSource (the long-lived connection counts as in-flight) — wait on a concrete readiness signal instead.
4. Playwright: webServer spawns before globalSetup, so any env the proxy needs must be declared statically in webServer.env, not threaded from globalSetup.

Public surface

Name	What
SSEStreamService	acquire slot → subscribe → stream (open_stream)
SSEStream / sse_headers	the (headers, body) result + the proxy-safe headers
Broker / Subscription	the pub/sub port (Protocols)
RedisBroker / MockBroker	adapters ([redis] extra / in-process)
SlotCap	per-key INCR/DECR connection cap, fail-open, TTL
CapExceededError / LivePushError	typed errors (subclass AppError)
mgf.livepush.fastapi.sse_streaming_response	one-line FastAPI wrapper ([fastapi] extra)