fix(api): stop dashboard event and log reads racing worker threads

Readers filter a deque copy, publish takes a re-entrant lock, seq moves after the append, and poll endpoints return a cursor that never skips a record.

tests/remote/unit/test_dashboard.py

@@ -4,13 +4,16 @@ from __future__ import annotations
 
 import asyncio
 import logging
+import sys
+import threading
 import time
+from collections.abc import Callable
 from datetime import datetime, timedelta, timezone
 
 import pytest
 from aiohttp import web
 
-from unshackle.core.api.events import bus
+from unshackle.core.api.events import EventBus, bus
 from unshackle.core.api.handlers import api_key_authentication, dashboard_authentication
 from unshackle.core.api.routes import cors_middleware, setup_routes
 from unshackle.core.api.session_store import SessionStore
@@ -72,6 +75,56 @@ def test_ring_since_and_level() -> None:
     assert [r["msg"] for r in ring.since(0, "warning")] == ["m3"]
 
 
+def _hammer(write: Callable[[], object], read: Callable[[], object]) -> None:
+    """Call ``read`` on this thread while four threads call ``write`` in a loop."""
+    stop = threading.Event()
+
+    def spin() -> None:
+        while not stop.is_set():
+            write()
+
+    old_interval = sys.getswitchinterval()
+    sys.setswitchinterval(1e-6)
+    threads = [threading.Thread(target=spin, daemon=True) for _ in range(4)]
+    try:
+        for t in threads:
+            t.start()
+        for _ in range(300):
+            read()
+    finally:
+        stop.set()
+        sys.setswitchinterval(old_interval)
+        for t in threads:
+            t.join()
+
+
+def test_bus_since_survives_concurrent_publish() -> None:
+    local = EventBus(history=5000)
+    _hammer(lambda: local.publish("log", {}), lambda: local.since(0))
+    seqs = [item["seq"] for item in local.history]
+    assert seqs == list(range(seqs[0], seqs[0] + len(seqs)))
+
+
+def test_bus_publish_reenters_its_lock() -> None:
+    local = EventBus()
+
+    def nested() -> None:
+        with local._lock:
+            local.publish("log", {})
+
+    worker = threading.Thread(target=nested, daemon=True)
+    worker.start()
+    worker.join(5)
+    assert not worker.is_alive(), "publish() deadlocked re-entering its own lock"
+    assert [item["seq"] for item in local.since(0)] == [1]
+
+
+def test_ring_since_survives_concurrent_emit() -> None:
+    ring = RingLogHandler(maxlen=5000)
+    record = logging.LogRecord("test.ring", logging.INFO, __file__, 0, "m", None, None)
+    _hammer(lambda: ring.handle(record), lambda: ring.since(0))
+
+
 def test_ring_masks_host_paths_and_secrets_in_records() -> None:
     """The ring feeds both /api/logs and the SSE stream, so it must mask on the way in."""
     from pathlib import Path

unshackle/core/api/events.py

@@ -18,6 +18,7 @@ class EventBus:
         self.history: Deque[Dict[str, Any]] = deque(maxlen=history)
         self._loop: Optional[asyncio.AbstractEventLoop] = None
         self._loop_thread: Optional[threading.Thread] = None
+        self._lock = threading.RLock()
 
     def subscribe(self) -> asyncio.Queue:
         self._loop = asyncio.get_running_loop()
@@ -31,12 +32,25 @@ class EventBus:
             self._subs.remove(queue)
 
     def since(self, seq: int) -> List[Dict[str, Any]]:
-        return [item for item in self.history if item["seq"] > seq]
+        """Return the history after ``seq``.
+
+        ``deque.copy()`` is one C call, so it is atomic against an append from another thread.
+        Do not take the lock here: a finalizer that logs on the loop can then wait for the log
+        handler's lock while a worker that holds it waits for this lock, and the server stops.
+        """
+        return [item for item in self.history.copy() if item["seq"] > seq]
 
     def publish(self, event: str, data: Dict[str, Any]) -> None:
-        self.seq += 1
-        item = {"seq": self.seq, "event": event, "data": data}
-        self.history.append(item)
+        """Record and fan out one event; safe to call from any thread.
+
+        The lock is re-entrant because a log call inside it publishes again. ``seq``
+        moves only after the append, so a reader that sees it also finds the event in history.
+        """
+        with self._lock:
+            seq = self.seq + 1
+            item = {"seq": seq, "event": event, "data": data}
+            self.history.append(item)
+            self.seq = seq
         if not self._subs:
             return
         on_loop = self._loop is None or threading.current_thread() is self._loop_thread

unshackle/core/api/handlers.py

@@ -2302,6 +2302,14 @@ async def dashboard_jobs_handler(request: web.Request) -> web.Response:
     return web.json_response([job.to_dict(include_full_details=True) for job in get_download_manager().list_jobs()])
 
 
+def _poll_cursor(before: int, items: List[Dict[str, Any]]) -> int:
+    """Cursor for the next poll, from the buffer's seq read *before* its ``since()`` call.
+
+    This poll or the next one returns a record added between the two reads; no poll skips it.
+    """
+    return max(before, items[-1]["seq"]) if items else before
+
+
 async def dashboard_logs_handler(request: web.Request) -> web.Response:
     """Recent log records; `since` (seq) and `level` filter the ring buffer."""
     from unshackle.core.api.stats import ring
@@ -2310,8 +2318,9 @@ async def dashboard_logs_handler(request: web.Request) -> web.Response:
         since = int(request.query.get("since", 0))
     except ValueError:
         since = 0
+    before = ring.seq
     records = ring.since(since, request.query.get("level"), request.query.get("logger"))
-    return web.json_response({"seq": ring.seq, "records": records})
+    return web.json_response({"seq": _poll_cursor(before, records), "records": records})
 
 
 async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
@@ -2332,13 +2341,9 @@ async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
     except ValueError:
         since = 0
     buffer = session.log_buffer
-    return web.json_response(
-        {
-            "session_id": session_id,
-            "records": buffer.since(since) if buffer else [],
-            "last_seq": buffer.last_seq if buffer else 0,
-        }
-    )
+    before = buffer.last_seq if buffer else 0
+    records = buffer.since(since) if buffer else []
+    return web.json_response({"session_id": session_id, "records": records, "last_seq": _poll_cursor(before, records)})
 
 
 async def dashboard_keys_handler(request: web.Request) -> web.Response:
@@ -2630,7 +2635,11 @@ async def dashboard_events_handler(request: web.Request) -> web.StreamResponse:
         except ValueError:
             since = 0
         status = await dashboard_status_handler(request)
-        return web.json_response({"seq": bus.seq, "stats": json.loads(status.text or "{}"), "events": bus.since(since)})
+        before = bus.seq
+        events = bus.since(since)
+        return web.json_response(
+            {"seq": _poll_cursor(before, events), "stats": json.loads(status.text or "{}"), "events": events}
+        )
 
     response = web.StreamResponse(
         headers={

unshackle/core/api/stats.py

@@ -224,24 +224,27 @@ class RingLogHandler(logging.Handler):
         self.seq = 0
 
     def emit(self, record: logging.LogRecord) -> None:
-        self.seq += 1
+        """Buffer and publish one record; ``seq`` moves only after the append, as in ``EventBus``."""
+        seq = self.seq + 1
         item = {
-            "seq": self.seq,
+            "seq": seq,
             "ts": record.created,
             "level": record.levelname,
             "logger": record.name,
             "msg": redact_secrets(self.format(record)),
         }
         self.records.append(item)
+        self.seq = seq
         bus.publish("log", item)
 
     def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
+        """Filter a copy of the records: ``emit()`` appends from worker threads."""
         min_level = logging.getLevelName(level.upper()) if level else 0
         if not isinstance(min_level, int):
             min_level = 0
         return [
             r
-            for r in self.records
+            for r in self.records.copy()
             if r["seq"] > seq
             and logging.getLevelName(r["level"]) >= min_level
             and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))