wehub-resource-sync
8f10353f0c
CI / lint (push) Has been cancelled
CI / js-syntax (push) Successful in 10m24s
CI / deps-audit (push) Has been cancelled
CI / test (push) Failing after 24m55s
CI / sast-bandit (push) Failing after 11m13s
CI / trivy (push) Failing after 9m32s
Docker Publish / build-and-push (push) Failing after 34m3s
69 lines
2.4 KiB
Python
69 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from collections.abc import AsyncIterator
|
|
|
|
from fastapi import APIRouter, HTTPException
|
|
from fastapi.responses import StreamingResponse
|
|
|
|
from app.core.config import JOB_ID_RE
|
|
from app.core.registry import get as registry_get
|
|
|
|
router = APIRouter(tags=["events"])
|
|
|
|
# Close SSE connections that outlive this threshold to prevent zombie
|
|
# connections from accumulating when clients disconnect without a TCP RST.
|
|
_MAX_SSE_SECONDS = 4 * 3600 # 4 hours
|
|
# Hard cap on concurrent SSE connections to prevent resource exhaustion
|
|
# from tab leaks or aggressive reconnect loops.
|
|
_MAX_SSE_CONNECTIONS = 200
|
|
# Counter is only mutated from the async event loop (no awaits between
|
|
# check+increment), so no lock is needed.
|
|
_sse_active = 0
|
|
|
|
|
|
@router.get("/jobs/{job_id}/events")
|
|
async def job_events(job_id: str) -> StreamingResponse:
|
|
"""Server-Sent Events stream of job state updates. Closes when the job
|
|
reaches a terminal status (done, error, cancelled) or after 4 hours."""
|
|
global _sse_active
|
|
if not JOB_ID_RE.match(job_id):
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
job = registry_get(job_id)
|
|
if job is None:
|
|
raise HTTPException(status_code=404, detail="job not found")
|
|
if _sse_active >= _MAX_SSE_CONNECTIONS:
|
|
raise HTTPException(status_code=503, detail="too many concurrent streams")
|
|
_sse_active += 1
|
|
|
|
async def stream() -> AsyncIterator[str]:
|
|
global _sse_active
|
|
try:
|
|
last = None
|
|
keepalive_at = 0
|
|
loop = asyncio.get_running_loop()
|
|
deadline = loop.time() + _MAX_SSE_SECONDS
|
|
while loop.time() < deadline:
|
|
snapshot = job.to_state()
|
|
serialized = json.dumps(snapshot)
|
|
if serialized != last:
|
|
yield f"data: {serialized}\n\n"
|
|
last = serialized
|
|
keepalive_at = 0
|
|
if snapshot["status"] in ("done", "error", "cancelled"):
|
|
return
|
|
keepalive_at += 1
|
|
if keepalive_at >= 75: # ~15s
|
|
yield ": keepalive\n\n"
|
|
keepalive_at = 0
|
|
await asyncio.sleep(0.2)
|
|
finally:
|
|
_sse_active -= 1
|
|
|
|
return StreamingResponse(
|
|
stream(),
|
|
media_type="text/event-stream",
|
|
headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
|
|
)
|