From 4c422741c5a94ed3508bdd554b35e0815ec1dac8 Mon Sep 17 00:00:00 2001
From: dv <dv@caps.caps>
Date: Wed, 15 Apr 2026 16:53:58 +0300
Subject: [PATCH] =?UTF-8?q?=D1=84=D0=B8=D0=BA=D1=81=20sse?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 backend/app/routers/objects.py                |  5 +++++
 .../app/services/discovery/orchestrator.py    |  7 +++++--
 backend/app/services/sse.py                   | 21 +++++++++++++++++++
 frontend/src/components/DiscoveryDrawer.tsx   |  3 ++-
 4 files changed, 33 insertions(+), 3 deletions(-)

diff --git a/backend/app/routers/objects.py b/backend/app/routers/objects.py
index 91f268a..9b4d384 100644
--- a/backend/app/routers/objects.py
+++ b/backend/app/routers/objects.py
@@ -256,6 +256,11 @@ async def sync_devices(
 
 async def _run_discovery_background(obj_id: int, task_id: str) -> None:
     """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:
         await sse_manager.publish(SSEEvent(type=event_type, task_id=task_id, data=data))
diff --git a/backend/app/services/discovery/orchestrator.py b/backend/app/services/discovery/orchestrator.py
index 742a57f..1bcfbd8 100644
--- a/backend/app/services/discovery/orchestrator.py
+++ b/backend/app/services/discovery/orchestrator.py
@@ -147,7 +147,7 @@ async def discover_via_ssh(
             try:
                 await db.flush()
             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", {
                 "rack_name": rack_name, "rack_host": rack_host, "found": 0, "ssh_ok": False
             })
@@ -171,7 +171,10 @@ async def discover_via_ssh(
         try:
             await db.flush()
         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", {
             "rack_name": rack_name, "rack_host": rack_host,
             "found": len(devices_found), "ssh_ok": True,
diff --git a/backend/app/services/sse.py b/backend/app/services/sse.py
index af6e0d9..30f8b25 100644
--- a/backend/app/services/sse.py
+++ b/backend/app/services/sse.py
@@ -2,13 +2,20 @@
 SSEManager — pub/sub hub for Server-Sent Events.
 
 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 json
+from collections import deque
 from dataclasses import dataclass, field
 from datetime import datetime, timezone
 
+REPLAY_MAXLEN = 200
+
 
 @dataclass
 class SSEEvent:
@@ -25,11 +32,23 @@ class SSEManager:
     def __init__(self) -> None:
         # task_id -> list[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()
 
+    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 with self._lock:
             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 with self._lock:
@@ -43,6 +62,8 @@ class SSEManager:
 
     async def publish(self, event: SSEEvent) -> None:
         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, []))
         for queue in queues:
             try:
diff --git a/frontend/src/components/DiscoveryDrawer.tsx b/frontend/src/components/DiscoveryDrawer.tsx
index 847c5e6..b82fb5f 100644
--- a/frontend/src/components/DiscoveryDrawer.tsx
+++ b/frontend/src/components/DiscoveryDrawer.tsx
@@ -148,8 +148,9 @@ export function DiscoveryDrawer({ open, onClose, objectName, taskId, onComplete
           // ignore parse errors
         }
       },
-      onerror() {
+      onerror(err) {
         ctrl.abort()
+        throw err
       },
     })