фикс sse
This commit is contained in:
parent
08d766a8b7
commit
4c422741c5
4 changed files with 33 additions and 3 deletions
|
|
@ -256,6 +256,11 @@ async def sync_devices(
|
||||||
|
|
||||||
async def _run_discovery_background(obj_id: int, task_id: str) -> None:
|
async def _run_discovery_background(obj_id: int, task_id: str) -> None:
|
||||||
"""Run SSH discovery in background, emitting SSE progress events."""
|
"""Run SSH discovery in background, emitting SSE progress events."""
|
||||||
|
# Clear stale replay buffer from any previous run of this task_id,
|
||||||
|
# then yield the event loop so the frontend SSE connection can register
|
||||||
|
# before we start publishing events.
|
||||||
|
sse_manager.clear_replay(task_id)
|
||||||
|
await asyncio.sleep(0)
|
||||||
|
|
||||||
async def emit(event_type: str, data: dict) -> None:
|
async def emit(event_type: str, data: dict) -> None:
|
||||||
await sse_manager.publish(SSEEvent(type=event_type, task_id=task_id, data=data))
|
await sse_manager.publish(SSEEvent(type=event_type, task_id=task_id, data=data))
|
||||||
|
|
|
||||||
|
|
@ -147,7 +147,7 @@ async def discover_via_ssh(
|
||||||
try:
|
try:
|
||||||
await db.flush()
|
await db.flush()
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
result.errors.append(f"Flush error for offline rack {rack_host}: {exc}")
|
raise RuntimeError(f"DB flush failed for offline rack {rack_host}: {exc}") from exc
|
||||||
await _emit("disc_rack_done", {
|
await _emit("disc_rack_done", {
|
||||||
"rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False
|
"rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False
|
||||||
})
|
})
|
||||||
|
|
@ -171,7 +171,10 @@ async def discover_via_ssh(
|
||||||
try:
|
try:
|
||||||
await db.flush()
|
await db.flush()
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
result.errors.append(f"Flush error after rack {rack_host}: {exc}")
|
# Session is now invalid — re-raise so the outer handler emits disc_done with error.
|
||||||
|
# Continuing would cause every subsequent DB call to fail silently, producing
|
||||||
|
# an incomplete discovery with no clear termination signal to the frontend.
|
||||||
|
raise RuntimeError(f"DB flush failed after rack {rack_host}: {exc}") from exc
|
||||||
await _emit("disc_rack_done", {
|
await _emit("disc_rack_done", {
|
||||||
"rack_name": rack_name, "rack_host": rack_host,
|
"rack_name": rack_name, "rack_host": rack_host,
|
||||||
"found": len(devices_found), "ssh_ok": True,
|
"found": len(devices_found), "ssh_ok": True,
|
||||||
|
|
|
||||||
|
|
@ -2,13 +2,20 @@
|
||||||
SSEManager — pub/sub hub for Server-Sent Events.
|
SSEManager — pub/sub hub for Server-Sent Events.
|
||||||
|
|
||||||
Task runner calls publish() → all connected SSE clients watching that task_id receive the event.
|
Task runner calls publish() → all connected SSE clients watching that task_id receive the event.
|
||||||
|
|
||||||
|
Replay buffer: each task_id stores the last REPLAY_MAXLEN events so that clients who connect
|
||||||
|
slightly after publishing starts (race between POST /discover and GET /events) still receive
|
||||||
|
the initial events (disc_server, disc_rack_start, etc.).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
|
from collections import deque
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
|
||||||
|
REPLAY_MAXLEN = 200
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
@dataclass
|
||||||
class SSEEvent:
|
class SSEEvent:
|
||||||
|
|
@ -25,11 +32,23 @@ class SSEManager:
|
||||||
def __init__(self) -> None:
|
def __init__(self) -> None:
|
||||||
# task_id -> list[Queue]
|
# task_id -> list[Queue]
|
||||||
self._subscribers: dict[str, list[asyncio.Queue]] = {}
|
self._subscribers: dict[str, list[asyncio.Queue]] = {}
|
||||||
|
# task_id -> deque of recent events for late-joining subscribers
|
||||||
|
self._replay: dict[str, deque[SSEEvent]] = {}
|
||||||
self._lock = asyncio.Lock()
|
self._lock = asyncio.Lock()
|
||||||
|
|
||||||
|
def clear_replay(self, task_id: str) -> None:
|
||||||
|
"""Call before starting a new task to discard stale events from a previous run."""
|
||||||
|
self._replay.pop(task_id, None)
|
||||||
|
|
||||||
async def subscribe(self, task_id: str, queue: asyncio.Queue) -> None:
|
async def subscribe(self, task_id: str, queue: asyncio.Queue) -> None:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
self._subscribers.setdefault(task_id, []).append(queue)
|
self._subscribers.setdefault(task_id, []).append(queue)
|
||||||
|
# Replay buffered events so late subscribers catch up
|
||||||
|
for event in self._replay.get(task_id, []):
|
||||||
|
try:
|
||||||
|
queue.put_nowait(event)
|
||||||
|
except asyncio.QueueFull:
|
||||||
|
break
|
||||||
|
|
||||||
async def unsubscribe(self, task_id: str, queue: asyncio.Queue) -> None:
|
async def unsubscribe(self, task_id: str, queue: asyncio.Queue) -> None:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
|
|
@ -43,6 +62,8 @@ class SSEManager:
|
||||||
|
|
||||||
async def publish(self, event: SSEEvent) -> None:
|
async def publish(self, event: SSEEvent) -> None:
|
||||||
async with self._lock:
|
async with self._lock:
|
||||||
|
buf = self._replay.setdefault(event.task_id, deque(maxlen=REPLAY_MAXLEN))
|
||||||
|
buf.append(event)
|
||||||
queues = list(self._subscribers.get(event.task_id, []))
|
queues = list(self._subscribers.get(event.task_id, []))
|
||||||
for queue in queues:
|
for queue in queues:
|
||||||
try:
|
try:
|
||||||
|
|
|
||||||
|
|
@ -148,8 +148,9 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete
|
||||||
// ignore parse errors
|
// ignore parse errors
|
||||||
}
|
}
|
||||||
},
|
},
|
||||||
onerror() {
|
onerror(err) {
|
||||||
ctrl.abort()
|
ctrl.abort()
|
||||||
|
throw err
|
||||||
},
|
},
|
||||||
})
|
})
|
||||||
|
|
||||||
|
|
|
||||||
Loading…
Add table
Reference in a new issue