diff --git a/tests/remote/unit/test_dashboard.py b/tests/remote/unit/test_dashboard.py index 3b181a55..62af81f7 100644 --- a/tests/remote/unit/test_dashboard.py +++ b/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 diff --git a/unshackle/core/api/events.py b/unshackle/core/api/events.py index f8011eaa..51409b99 100644 --- a/unshackle/core/api/events.py +++ b/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 diff --git a/unshackle/core/api/handlers.py b/unshackle/core/api/handlers.py index 05c7fefe..0120f369 100644 --- a/unshackle/core/api/handlers.py +++ b/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={ diff --git a/unshackle/core/api/stats.py b/unshackle/core/api/stats.py index e205a226..e6b74064 100644 --- a/unshackle/core/api/stats.py +++ b/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 + "."))