From 51ddfb00faf2974a2bb4db11207aece2262690de Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 18:34:05 +0100 Subject: [PATCH 1/5] Add app health metrics and log previously silent collab/log-stream failures Co-Authored-By: Claude Sonnet 5.5 --- server/controllers/api/v1/logs_viewer.py | 9 +- server/controllers/ws_controller.py | 54 ++++- .../controllers/api/v1/test_logs_viewer.py | 36 ++++ server/test/controllers/test_ws_controller.py | 203 +++++++++++++++++- server/test/utils/script/test_script_room.py | 83 ++++++- server/utils/script_room_manager.py | 22 +- server/utils/web/app_metrics.py | 124 +++++++++++ 7 files changed, 519 insertions(+), 12 deletions(-) create mode 100644 server/utils/web/app_metrics.py diff --git a/server/controllers/api/v1/logs_viewer.py b/server/controllers/api/v1/logs_viewer.py index e1da05841..fe94fffd3 100644 --- a/server/controllers/api/v1/logs_viewer.py +++ b/server/controllers/api/v1/logs_viewer.py @@ -12,6 +12,7 @@ import json import logging +from digi_server.logger import get_logger from utils.log_buffer import get_client_buffer, get_server_buffer from utils.web.base_controller import BaseAPIController from utils.web.route import ApiRoute, ApiVersion @@ -234,7 +235,11 @@ async def get(self): if matched: self.write(f"data: {json.dumps(matched[0])}\n\n") await self.flush() - except Exception: # noqa: BLE001 - pass + except Exception: + # Usually the client dropping the stream mid-write; not an error. + get_logger().debug( + f"Log stream ({source}) from {self.request.remote_ip} ended on error", + exc_info=True, + ) finally: unsubscribe() diff --git a/server/controllers/ws_controller.py b/server/controllers/ws_controller.py index 17c3ceb70..b604c5f34 100644 --- a/server/controllers/ws_controller.py +++ b/server/controllers/ws_controller.py @@ -30,6 +30,7 @@ from models.show import Act, Show from models.user import User from rbac.role import Role +from utils.web import app_metrics from utils.web.base_controller import DatabaseMixin from utils.web.jwt_service import ( TOKEN_EXPIRED, @@ -68,6 +69,9 @@ ) +app_metrics.register_app_info() + + @ApiRoute("ws", ApiVersion.V1) class WebSocketController(WSMessageSizeMetricsMixin, DatabaseMixin, WebSocketHandler): # Live-show ops only the leader may send, and the method that handles each. @@ -92,8 +96,20 @@ class WebSocketController(WSMessageSizeMetricsMixin, DatabaseMixin, WebSocketHan "DISCARD_SCRIPT_DRAFT": "_op_discard_script_draft", } + # Every OP the server handles; the only values used for the ``op`` metric + # label, so a client cannot create unlimited series. + _SESSION_OPS = frozenset({"NEW_CLIENT", "REFRESH_CLIENT", "LIVE_SHOW_JUMP_TO_PAGE"}) + _KNOWN_OPS = ( + frozenset({"AUTHENTICATE", "REFRESH_TOKEN"}) + | _SESSION_OPS + | frozenset(_LEADER_OP_HANDLERS) + | frozenset(_SCRIPT_ROOM_OP_HANDLERS) + ) + def __init__(self, application, request, **kwargs): super().__init__(application, request, **kwargs) + # True between open() counting this connection and on_close() releasing it. + self._counted_connection = False self.application: DigiScriptServer = application self.current_user_id = None self.current_username: str | None = None @@ -110,6 +126,12 @@ def __init__(self, application, request, **kwargs): # not resumed (its row had gone); used for the late leadership reclaim. self._presented_uuid: Optional[str] = None + def _who(self) -> str: + """Return ``username (ip)`` (or just the ip) for log messages.""" + if self.current_username: + return f"{self.current_username} ({self.request.remote_ip})" + return str(self.request.remote_ip) + def update_session(self, user_id=None) -> bool: """Create or refresh the Session row for this connection's uuid. @@ -157,6 +179,8 @@ def check_origin(self, origin): @gen.coroutine def open(self, *args: str, **kwargs: str) -> Optional[Awaitable[None]]: + app_metrics.ws_connections.inc() + self._counted_connection = True self.__setattr__("internal_id", str(uuid4())) self._placeholder_id = self.__getattribute__("internal_id") self.application.clients.append(self) @@ -189,6 +213,9 @@ def on_close(self) -> None: if self._close_handled: return self._close_handled = True + if self._counted_connection: + self._counted_connection = False + app_metrics.ws_connections.dec() if self in self.application.clients: self.application.clients.remove(self) @@ -487,12 +514,17 @@ async def on_message(self, message: Union[str, bytes]): content = f"<{size} bytes>" else: content = raw + app_metrics.ws_messages_total.labels(app_metrics.OTHER_OP).inc() get_logger().warning( f"Malformed WS message from {self.request.remote_ip}: " f"{type(e).__name__}: {e} {content}" ) return + app_metrics.ws_messages_total.labels( + app_metrics.bounded_op(ws_op, self._KNOWN_OPS) + ).inc() + # With redaction on never log the payload: AUTHENTICATE / REFRESH_TOKEN # carry live JWTs. if self.application.digi_settings.get_sync("log_redaction"): @@ -1208,7 +1240,11 @@ async def _op_yjs_sync(self, room_manager, data: dict) -> None: try: decoded = base64.b64decode(payload) except Exception: - get_logger().warning("Invalid base64 in YJS_SYNC message") + get_logger().warning( + f"Invalid base64 in YJS_SYNC message for revision " + f"{room.revision_id} from {self._who()} " + f"(payload {len(str(payload))} chars)" + ) return if step == 1: @@ -1287,7 +1323,11 @@ async def _op_yjs_update(self, room_manager, data: dict) -> None: try: decoded = base64.b64decode(payload) except Exception: - get_logger().warning("Invalid base64 in YJS_UPDATE message") + get_logger().warning( + f"Invalid base64 in YJS_UPDATE message for revision " + f"{room.revision_id} from {self._who()} " + f"(payload {len(str(payload))} chars)" + ) return if await self._is_live_session_active(): @@ -1307,6 +1347,10 @@ async def _op_yjs_update(self, room_manager, data: dict) -> None: try: follow_up = await room.apply_update(decoded) except Exception: + get_logger().exception( + f"YJS_UPDATE: Failed to apply update for revision " + f"{room.revision_id} from {self._who()}" + ) await self.write_message( { "OP": "NOOP", @@ -1330,7 +1374,11 @@ async def _op_yjs_awareness(self, room_manager, data: dict) -> None: try: decoded = base64.b64decode(payload) except Exception: - get_logger().warning("Invalid base64 in YJS_AWARENESS message") + get_logger().debug( + f"Invalid base64 in YJS_AWARENESS message for revision " + f"{room.revision_id} from {self._who()} " + f"(payload {len(str(payload))} chars)" + ) return await room.broadcast_awareness(decoded, sender=self) diff --git a/server/test/controllers/api/v1/test_logs_viewer.py b/server/test/controllers/api/v1/test_logs_viewer.py index c721a8d00..d2e3c268c 100644 --- a/server/test/controllers/api/v1/test_logs_viewer.py +++ b/server/test/controllers/api/v1/test_logs_viewer.py @@ -1,9 +1,11 @@ """Integration tests for GET /api/v1/logs/view.""" import logging +from unittest import mock from tornado import escape +from digi_server.logger import get_logger from test.conftest import DigiScriptTestCase from utils.log_buffer import get_client_buffer, get_server_buffer @@ -373,6 +375,40 @@ def test_stream_non_admin_returns_403(self): resp = self._fetch_stream_headers(token=user_token) self.assertEqual(403, resp.code) + def test_stream_error_is_logged_at_debug(self): + """An error ending the live stream is logged instead of swallowed.""" + token = self._create_and_login_admin() + calls = [] + + def flaky_filter(entries, *args): + calls.append(1) + if len(calls) > 1: + raise RuntimeError("boom") + return [] + + self.io_loop.call_later( + 0.3, lambda: self._inject_server_entry("sse_error_trigger") + ) + with ( + mock.patch("controllers.api.v1.logs_viewer._filter_entries", flaky_filter), + self.assertLogs(get_logger(), level="DEBUG") as captured, + ): + try: + self.fetch( + "/api/v1/logs/stream?source=server", + method="GET", + headers={"Authorization": f"Bearer {token}"}, + raise_error=False, + request_timeout=1.0, + streaming_callback=lambda _chunk: None, + ) + except Exception: + pass # HTTPTimeoutError: the stream stays open otherwise + self.assertTrue( + any("Log stream (server)" in line for line in captured.output), + captured.output, + ) + def test_stream_backfill_contains_existing_entries(self): """Entries injected before connection must appear in the first chunk. diff --git a/server/test/controllers/test_ws_controller.py b/server/test/controllers/test_ws_controller.py index cd74faf83..d06efe597 100644 --- a/server/test/controllers/test_ws_controller.py +++ b/server/test/controllers/test_ws_controller.py @@ -8,11 +8,13 @@ import base64 import json import os +import weakref from datetime import timedelta from unittest import mock from unittest.mock import AsyncMock import pycrdt +from prometheus_client import REGISTRY from sqlalchemy import select from tornado.httpclient import HTTPRequest from tornado.testing import gen_test @@ -20,6 +22,7 @@ from controllers.ws_controller import WebSocketController from digi_server.logger import get_logger +from digi_server.settings import get_version from models import models from models.script import Script from models.script_draft import ScriptDraft @@ -32,7 +35,7 @@ from test.helpers.auth_fixtures import bump_token_version, craft_token, make_user from test.helpers.script_fixtures import create_show_script_revision from utils.script_room_manager import ScriptRoom -from utils.web import ws_session_lifecycle +from utils.web import app_metrics, ws_session_lifecycle from utils.web.pending_disconnects import ( disconnect_key, room_close_key, @@ -1121,6 +1124,72 @@ async def _raise_apply_update(_data): ws.close() + async def _join_collab_room(self): + ws, _ = await self._connect_and_auth(self.admin_id) + await ws.write_message( + json.dumps({"OP": "REQUEST_SCRIPT_EDIT", "DATA": {"collab": True}}) + ) + await ws.read_message() # GET_SCRIPT_CONFIG_STATUS + await ws.write_message(json.dumps({"OP": "JOIN_SCRIPT_ROOM", "DATA": {}})) + await ws.read_message() # YJS_SYNC + await ws.read_message() # ROOM_MEMBERS + await ws.read_message() # GET_SCRIPT_CONFIG_STATUS from join + return ws + + @gen_test + async def test_invalid_base64_is_logged_with_context(self): + """Invalid base64 on YJS ops is logged with revision, never the payload.""" + ws = await self._join_collab_room() + room = self._app.room_manager.get_active_room() + bad = "not*base64*secret-marker" + for op, level in ( + ("YJS_UPDATE", "WARNING"), + ("YJS_SYNC", "WARNING"), + ("YJS_AWARENESS", "DEBUG"), + ): + with self.assertLogs(get_logger(), level=level) as captured: + await ws.write_message( + json.dumps({"OP": op, "DATA": {"payload": bad, "step": 1}}) + ) + for _ in range(100): + if any(f"Invalid base64 in {op}" in m for m in captured.output): + break + await asyncio.sleep(0.02) + line = next(m for m in captured.output if f"Invalid base64 in {op}" in m) + self.assertIn(f"revision {room.revision_id}", line) + self.assertIn("127.0.0.1", line) + self.assertNotIn("secret-marker", line) + ws.close() + + @gen_test + async def test_yjs_update_apply_failure_is_logged(self): + """A failing YJS_UPDATE apply logs the exception with the revision.""" + ws = await self._join_collab_room() + room = self._app.room_manager.get_active_room() + + async def _raise_apply_update(_data): + raise ValueError("Simulated corrupt Yjs payload") + + room.apply_update = _raise_apply_update + with self.assertLogs(get_logger(), level="ERROR") as captured: + await ws.write_message( + json.dumps( + { + "OP": "YJS_UPDATE", + "DATA": {"payload": base64.b64encode(b"x").decode("ascii")}, + } + ) + ) + await ws.read_message() # COLLAB_ERROR + self.assertTrue( + any( + f"YJS_UPDATE: Failed to apply update for revision {room.revision_id}" + in m + for m in captured.output + ) + ) + ws.close() + @gen_test async def test_yjs_sync_step2_corrupt_payload_returns_collab_error(self): """A YJS_SYNC step=2 whose apply_update raises returns COLLAB_ERROR.""" @@ -1202,6 +1271,24 @@ def _set_live_session_active(self): session.commit() return show_session.id + def test_show_session_active_gauge_follows_current_session(self): + """``digiscript_show_session_active`` is 1 only while a session is live.""" + + def active(): + return REGISTRY.get_sample_value("digiscript_show_session_active") + + # Apps from earlier tests may still be alive; look at this one only. + with mock.patch.object( + app_metrics, "_applications", weakref.WeakSet([self._app]) + ): + self.assertEqual(0.0, active()) + self._set_live_session_active() + self.assertEqual(1.0, active()) + with self._app.get_db().sessionmaker() as session: + session.get(Show, self.show_id).current_session_id = None + session.commit() + self.assertEqual(0.0, active()) + @gen_test async def test_request_script_edit_blocked_by_live_session(self): """REQUEST_SCRIPT_EDIT is rejected while a live show session is running.""" @@ -3792,6 +3879,120 @@ async def test_inbound_message_size_observed(self): self.assertGreaterEqual(self._message_count(), before + 2) +class TestWSAppMetrics(_WSTestHelpers, DigiScriptTestCase): + """Connection gauge and per-OP counter. Series are process-wide: read deltas.""" + + @staticmethod + def _sample(name, labels=None): + return REGISTRY.get_sample_value(name, labels or {}) or 0.0 + + def _connections(self): + return self._sample("digiscript_ws_connections") + + def _op_count(self, op): + return self._sample("digiscript_ws_messages_total", {"op": op}) + + async def _wait_for(self, predicate): + for _ in range(100): + if predicate(): + return + await asyncio.sleep(0.02) + self.fail("condition not reached") + + @gen_test + async def test_connections_gauge_up_on_open_down_on_close(self): + before = self._connections() + ws, uuid = await self._connect_and_auth() + self.assertEqual(before + 1, self._connections()) + ws2, _ = await self._connect_and_auth() + self.assertEqual(before + 2, self._connections()) + + ws.close() + await self._wait_for(lambda: self._connections() == before + 1) + ws2.close() + await self._wait_for(lambda: self._connections() == before) + + @gen_test + async def test_double_close_decrements_once(self): + before = self._connections() + ws, uuid = await self._connect_and_auth() + handler = self._app.get_ws(uuid) + self.assertEqual(before + 1, self._connections()) + + handler.on_close() + handler.on_close() + self.assertEqual(before, self._connections()) + ws.close() + await asyncio.sleep(0.1) + self.assertEqual(before, self._connections()) + + @gen_test + async def test_close_without_open_does_not_decrement(self): + before = self._connections() + handler = WebSocketController.__new__(WebSocketController) + handler._close_handled = False + handler._counted_connection = False + handler.application = self._app + handler.request = mock.Mock(remote_ip="127.0.0.1") + handler.current_username = None + WebSocketController.on_close(handler) + self.assertEqual(before, self._connections()) + + @gen_test + async def test_reconnect_does_not_double_count(self): + """A resumed uuid is a new connection: +1 for it, -1 for the old one.""" + before = self._connections() + ws, _ = await self._connect_and_auth() + ws.close() + await self._wait_for(lambda: self._connections() == before) + ws2, _ = await self._connect_and_auth() + self.assertEqual(before + 1, self._connections()) + ws2.close() + await self._wait_for(lambda: self._connections() == before) + + @gen_test + async def test_messages_counted_by_op_and_unknown_bounded(self): + known_before = self._op_count("NEW_CLIENT") + other_before = self._op_count("OTHER") + ws, _ = await self._connect_and_auth() + await ws.write_message(json.dumps({"OP": "NEW_CLIENT", "DATA": {}})) + await ws.write_message(json.dumps({"OP": "ATTACKER_OP_1", "DATA": {}})) + await ws.write_message(json.dumps({"OP": "ATTACKER_OP_2", "DATA": {}})) + await ws.write_message("not json") + await self._wait_for(lambda: self._op_count("OTHER") >= other_before + 3) + self.assertEqual(known_before + 1, self._op_count("NEW_CLIENT")) + self.assertEqual(other_before + 3, self._op_count("OTHER")) + self.assertIsNone( + REGISTRY.get_sample_value( + "digiscript_ws_messages_total", {"op": "ATTACKER_OP_1"} + ) + ) + ws.close() + + def test_known_ops_cover_every_handler(self): + for op in ( + *WebSocketController._LEADER_OP_HANDLERS, + *WebSocketController._SCRIPT_ROOM_OP_HANDLERS, + "AUTHENTICATE", + "REFRESH_TOKEN", + "NEW_CLIENT", + "REFRESH_CLIENT", + "LIVE_SHOW_JUMP_TO_PAGE", + ): + self.assertIn(op, WebSocketController._KNOWN_OPS) + + def test_info_gauge_carries_version(self): + self.assertEqual( + 1.0, + REGISTRY.get_sample_value("digiscript_info", {"version": get_version()}), + ) + app_metrics.register_app_info() # idempotent + self.assertEqual( + 1.0, + REGISTRY.get_sample_value("digiscript_info", {"version": get_version()}), + ) + + class TestWSOversizedMessage(DigiScriptTestCase): @gen_test async def test_message_over_limit_closes_connection(self): diff --git a/server/test/utils/script/test_script_room.py b/server/test/utils/script/test_script_room.py index f668fe4ea..2c142ed7a 100644 --- a/server/test/utils/script/test_script_room.py +++ b/server/test/utils/script/test_script_room.py @@ -7,14 +7,17 @@ import asyncio import datetime import os +import tempfile import time import uuid from unittest.mock import AsyncMock, patch import pycrdt import pytest +from prometheus_client import REGISTRY from sqlalchemy import select from tornado.testing import gen_test +from tornado.websocket import WebSocketClosedError from models.script import ( ScriptLine, @@ -29,7 +32,13 @@ _ScriptTestSetup, ) from utils.script.line_to_ydoc import build_ydoc, fetch_script_line_data -from utils.script_room_manager import CHECKPOINT_INTERVAL, RoomManager, ScriptRoom +from utils.script_room_manager import ( + CHECKPOINT_INTERVAL, + RoomManager, + ScriptRoom, + _write_checkpoint_sync, +) +from utils.web import app_metrics def _make_mock_ws(): @@ -1125,3 +1134,75 @@ async def test_a_legacy_draft_gains_a_trailing_page_and_uids_on_load(self): finally: if os.path.exists(draft_path): os.remove(draft_path) + + +class TestRoomManagerMetrics: + """The ``digiscript_collab_rooms`` gauge follows the active room.""" + + @staticmethod + def _rooms(): + return REGISTRY.get_sample_value("digiscript_collab_rooms") + + @pytest.mark.asyncio + async def test_collab_rooms_gauge_tracks_create_and_close(self): + manager = RoomManager.__new__(RoomManager) + manager._room = None + manager._room_lock = asyncio.Lock() + app_metrics.register_room_manager(manager) + manager._load_or_build_doc = AsyncMock(return_value=pycrdt.Doc()) + before = self._rooms() + + await manager.get_or_create_room(7) + assert self._rooms() == before + 1 + + # Same revision again must not double count. + await manager.get_or_create_room(7) + assert self._rooms() == before + 1 + + await manager.close_active_room() + assert self._rooms() == before + # Closing again is a no-op. + await manager.close_active_room() + assert self._rooms() == before + + def test_collab_rooms_gauge_drops_on_stop(self): + manager = RoomManager.__new__(RoomManager) + manager._maintenance_handle = None + manager._room = _make_room(3) + app_metrics.register_room_manager(manager) + before = self._rooms() + + manager.stop() + + assert self._rooms() == before - 1 + + +class TestSilentFailureLogging: + """Previously silent branches now leave a log line.""" + + def test_checkpoint_cleanup_logs_unlink_failure(self): + with tempfile.TemporaryDirectory() as tmp: + target = os.path.join(tmp, "x.yjs") + with ( + patch("utils.script_room_manager.os.rename", side_effect=OSError("no")), + patch("utils.script_room_manager.os.unlink", side_effect=OSError("u")), + patch("utils.script_room_manager.get_logger") as get_logger, + ): + with pytest.raises(OSError): + _write_checkpoint_sync(b"abc", target, tmp) + warning = get_logger.return_value.warning + warning.assert_called_once() + assert "could not remove temporary file" in warning.call_args[0][0] + # The fd was already closed before the rename failed (EBADF). + get_logger.return_value.debug.assert_called_once() + + @pytest.mark.asyncio + async def test_save_progress_closed_client_logged_at_debug(self): + room = _make_room(revision_id=11) + ws = _make_mock_ws() + ws.write_message.side_effect = WebSocketClosedError() + room.add_client(ws, "editor") + with patch("utils.script_room_manager.get_logger") as get_logger: + await room._broadcast_save_progress(1, 2) + get_logger.return_value.debug.assert_called_once() + assert "11" in get_logger.return_value.debug.call_args[0][0] diff --git a/server/utils/script_room_manager.py b/server/utils/script_room_manager.py index 528cbf48e..d2c7ad439 100644 --- a/server/utils/script_room_manager.py +++ b/server/utils/script_room_manager.py @@ -39,6 +39,7 @@ ) from utils.script.ydoc_to_lines import _save_script_page, extract_lines_from_ydoc from utils.script.yjs_debug import attach_trace_observers +from utils.web import app_metrics if TYPE_CHECKING: @@ -70,12 +71,18 @@ def _write_checkpoint_sync(state: bytes, draft_path: str, draft_dir: str) -> Non except Exception: try: os.close(fd) - except OSError: - pass + except OSError as close_err: + # Usually already closed by the failing step; not the root cause. + get_logger().debug( + f"Checkpoint cleanup: could not close fd for {tmp_path}: {close_err}" + ) try: os.unlink(tmp_path) - except OSError: - pass + except OSError as unlink_err: + get_logger().warning( + f"Checkpoint cleanup: could not remove temporary file " + f"{tmp_path}: {unlink_err}" + ) raise @@ -285,7 +292,10 @@ async def _broadcast_save_progress(self, page: int, total: int) -> None: } ) except WebSocketClosedError: - pass + get_logger().debug( + f"Client disconnected during SAVE_PROGRESS broadcast " + f"in room {self.revision_id}" + ) except Exception: get_logger().debug( f"Unexpected error sending save progress " @@ -523,6 +533,8 @@ def __init__(self, application: DigiScriptServer): self._room: ScriptRoom | None = None self._room_lock: asyncio.Lock = asyncio.Lock() self._maintenance_handle = None + app_metrics.register_room_manager(self) + app_metrics.register_application(application) def start(self): self._schedule_maintenance() diff --git a/server/utils/web/app_metrics.py b/server/utils/web/app_metrics.py new file mode 100644 index 000000000..5f95b4d48 --- /dev/null +++ b/server/utils/web/app_metrics.py @@ -0,0 +1,124 @@ +"""Prometheus gauges and counters answering "is the show healthy?". + +Everything here is module-level so it is registered exactly once per process, +however many application instances are created (tests create many). The series +are exposed alongside the others at ``/debug/metrics``: + +* ``digiscript_ws_connections`` - currently open WebSocket connections. +* ``digiscript_ws_messages_total{op}`` - inbound WebSocket messages by OP. The + OP is client-controlled, so only the OPs the server knows are used as label + values; anything else is reported as ``OTHER``. +* ``digiscript_collab_rooms`` - active collaborative script rooms, read live + from the registered room managers at scrape time so it cannot drift. +* ``digiscript_show_session_active`` - 1 while a live show session is running, + read live from the database at scrape time. +* ``digiscript_info{version}`` - constant 1 carrying the application version. +""" + +from __future__ import annotations + +import weakref +from typing import TYPE_CHECKING + +from prometheus_client import Counter, Gauge + +from digi_server.logger import get_logger +from digi_server.settings import get_version +from models.show import Show + + +if TYPE_CHECKING: + from digi_server.app_server import DigiScriptServer + from utils.script_room_manager import RoomManager + + +OTHER_OP = "OTHER" + +ws_connections = Gauge( + "digiscript_ws_connections", "Currently open WebSocket connections" +) + +ws_messages_total = Counter( + "digiscript_ws_messages_total", + "Inbound WebSocket messages by OP (unknown OPs are reported as OTHER)", + labelnames=("op",), +) + +collab_rooms = Gauge( + "digiscript_collab_rooms", "Currently active collaborative script rooms" +) + +show_session_active = Gauge( + "digiscript_show_session_active", + "1 if a live show session is running, otherwise 0", +) + +app_info = Gauge( + "digiscript_info", "DigiScript application information", labelnames=("version",) +) + +_room_managers: weakref.WeakSet = weakref.WeakSet() +_applications: weakref.WeakSet = weakref.WeakSet() + + +def bounded_op(op: str, known_ops) -> str: + """Return ``op`` if the server knows it, else ``OTHER``. + + :param op: The OP taken from a client message. + :param known_ops: Collection of OPs allowed as label values. + :returns: A label value from a fixed, finite set. + """ + return op if op in known_ops else OTHER_OP + + +def register_room_manager(manager: RoomManager) -> None: + """Include ``manager``'s active room in ``digiscript_collab_rooms``. + + :param manager: The room manager to count. + """ + _room_managers.add(manager) + + +def register_application(app: DigiScriptServer) -> None: + """Include ``app``'s live show session in ``digiscript_show_session_active``. + + :param app: The application server. + """ + _applications.add(app) + + +def register_app_info() -> None: + """Set ``digiscript_info`` for the running version (idempotent).""" + app_info.labels(get_version()).set(1) + + +def _count_rooms() -> float: + return float(sum(1 for m in list(_room_managers) if m.get_active_room())) + + +def _app_has_live_session(app: DigiScriptServer) -> bool: + setting = app.digi_settings.settings.get("current_show") + show_id = setting.get_value() if setting else None + if not show_id: + return False + with app.get_db().sessionmaker() as session: + show = session.get(Show, show_id) + return bool(show and show.current_session_id) + + +def _live_session_value() -> float: + active = False + for app in list(_applications): + try: + active = active or _app_has_live_session(app) + except Exception: + get_logger().warning( + "Could not determine live show session state for metrics", + exc_info=True, + ) + return float("nan") + return 1.0 if active else 0.0 + + +collab_rooms.set_function(_count_rooms) +show_session_active.set_function(_live_session_value) From b2c89812f04dc8e1b9d6d4e9847bb54212d7d3b1 Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:15:12 +0100 Subject: [PATCH 2/5] Address review round 1: guard metric callbacks, split log stream errors, real OP coverage test Co-Authored-By: Claude Sonnet 5.5 --- server/controllers/api/v1/logs_viewer.py | 15 +++- server/controllers/ws_controller.py | 5 +- .../controllers/api/v1/test_logs_viewer.py | 27 +++++-- server/test/controllers/test_ws_controller.py | 80 ++++++++++++++++--- server/utils/script_room_manager.py | 4 + server/utils/web/app_metrics.py | 45 ++++++++--- 6 files changed, 142 insertions(+), 34 deletions(-) diff --git a/server/controllers/api/v1/logs_viewer.py b/server/controllers/api/v1/logs_viewer.py index fe94fffd3..ab27c3918 100644 --- a/server/controllers/api/v1/logs_viewer.py +++ b/server/controllers/api/v1/logs_viewer.py @@ -12,6 +12,8 @@ import json import logging +from tornado.iostream import StreamClosedError + from digi_server.logger import get_logger from utils.log_buffer import get_client_buffer, get_server_buffer from utils.web.base_controller import BaseAPIController @@ -235,10 +237,17 @@ async def get(self): if matched: self.write(f"data: {json.dumps(matched[0])}\n\n") await self.flush() - except Exception: - # Usually the client dropping the stream mid-write; not an error. + except StreamClosedError: + # The client dropped the stream mid-write; expected, not an error. get_logger().debug( - f"Log stream ({source}) from {self.request.remote_ip} ended on error", + f"Log stream ({source}) from {self.request.remote_ip} closed " + f"by the client" + ) + except Exception: + # Filtering or serialising an entry failed: a bug, not a disconnect. + get_logger().warning( + f"Log stream ({source}) from {self.request.remote_ip} ended " + f"on an unexpected error", exc_info=True, ) finally: diff --git a/server/controllers/ws_controller.py b/server/controllers/ws_controller.py index b604c5f34..1e32ca8de 100644 --- a/server/controllers/ws_controller.py +++ b/server/controllers/ws_controller.py @@ -69,9 +69,6 @@ ) -app_metrics.register_app_info() - - @ApiRoute("ws", ApiVersion.V1) class WebSocketController(WSMessageSizeMetricsMixin, DatabaseMixin, WebSocketHandler): # Live-show ops only the leader may send, and the method that handles each. @@ -96,9 +93,9 @@ class WebSocketController(WSMessageSizeMetricsMixin, DatabaseMixin, WebSocketHan "DISCARD_SCRIPT_DRAFT": "_op_discard_script_draft", } + _SESSION_OPS = frozenset({"NEW_CLIENT", "REFRESH_CLIENT", "LIVE_SHOW_JUMP_TO_PAGE"}) # Every OP the server handles; the only values used for the ``op`` metric # label, so a client cannot create unlimited series. - _SESSION_OPS = frozenset({"NEW_CLIENT", "REFRESH_CLIENT", "LIVE_SHOW_JUMP_TO_PAGE"}) _KNOWN_OPS = ( frozenset({"AUTHENTICATE", "REFRESH_TOKEN"}) | _SESSION_OPS diff --git a/server/test/controllers/api/v1/test_logs_viewer.py b/server/test/controllers/api/v1/test_logs_viewer.py index d2e3c268c..d2238668a 100644 --- a/server/test/controllers/api/v1/test_logs_viewer.py +++ b/server/test/controllers/api/v1/test_logs_viewer.py @@ -4,6 +4,7 @@ from unittest import mock from tornado import escape +from tornado.iostream import StreamClosedError from digi_server.logger import get_logger from test.conftest import DigiScriptTestCase @@ -375,15 +376,18 @@ def test_stream_non_admin_returns_403(self): resp = self._fetch_stream_headers(token=user_token) self.assertEqual(403, resp.code) - def test_stream_error_is_logged_at_debug(self): - """An error ending the live stream is logged instead of swallowed.""" + def _run_stream_with_failing_filter(self, exc): + """Stream /logs/stream with a live-entry filter that raises ``exc``. + + :returns: Tuple of (level name, message) for each ``Log stream`` record. + """ token = self._create_and_login_admin() calls = [] def flaky_filter(entries, *args): calls.append(1) if len(calls) > 1: - raise RuntimeError("boom") + raise exc return [] self.io_loop.call_later( @@ -404,10 +408,19 @@ def flaky_filter(entries, *args): ) except Exception: pass # HTTPTimeoutError: the stream stays open otherwise - self.assertTrue( - any("Log stream (server)" in line for line in captured.output), - captured.output, - ) + return [ + (r.levelname, r.getMessage()) + for r in captured.records + if "Log stream (server)" in r.getMessage() + ] + + def test_stream_client_disconnect_is_logged_at_debug(self): + records = self._run_stream_with_failing_filter(StreamClosedError()) + self.assertEqual(["DEBUG"], [level for level, _ in records]) + + def test_stream_unexpected_error_is_logged_at_warning(self): + records = self._run_stream_with_failing_filter(RuntimeError("boom")) + self.assertEqual(["WARNING"], [level for level, _ in records]) def test_stream_backfill_contains_existing_entries(self): """Entries injected before connection must appear in the first chunk. diff --git a/server/test/controllers/test_ws_controller.py b/server/test/controllers/test_ws_controller.py index d06efe597..16531661f 100644 --- a/server/test/controllers/test_ws_controller.py +++ b/server/test/controllers/test_ws_controller.py @@ -4,17 +4,20 @@ the query patterns in ws_controller.py, following our endpoint-based testing approach. """ +import ast import asyncio import base64 +import inspect import json import os +import textwrap import weakref from datetime import timedelta from unittest import mock from unittest.mock import AsyncMock import pycrdt -from prometheus_client import REGISTRY +from prometheus_client import REGISTRY, generate_latest from sqlalchemy import select from tornado.httpclient import HTTPRequest from tornado.testing import gen_test @@ -3969,17 +3972,72 @@ async def test_messages_counted_by_op_and_unknown_bounded(self): ) ws.close() - def test_known_ops_cover_every_handler(self): - for op in ( - *WebSocketController._LEADER_OP_HANDLERS, - *WebSocketController._SCRIPT_ROOM_OP_HANDLERS, - "AUTHENTICATE", - "REFRESH_TOKEN", - "NEW_CLIENT", - "REFRESH_CLIENT", - "LIVE_SHOW_JUMP_TO_PAGE", + def test_known_ops_cover_every_op_the_handler_code_compares(self): + """Every OP literal compared against ``ws_op`` in the controller is known. + + Parses the controller source, so adding an ``elif ws_op == "NEW_OP"`` + branch without updating ``_KNOWN_OPS`` fails here. + """ + tree = ast.parse(textwrap.dedent(inspect.getsource(WebSocketController))) + compared = set() + for node in ast.walk(tree): + if not ( + isinstance(node, ast.Compare) + and isinstance(node.left, ast.Name) + and node.left.id == "ws_op" + ): + continue + for comparator in node.comparators: + for const in ast.walk(comparator): + if isinstance(const, ast.Constant) and isinstance(const.value, str): + compared.add(const.value) + self.assertIn("NEW_CLIENT", compared) # the walk really finds literals + self.assertIn("AUTHENTICATE", compared) + self.assertFalse(compared - WebSocketController._KNOWN_OPS) + for table in ( + WebSocketController._LEADER_OP_HANDLERS, + WebSocketController._SCRIPT_ROOM_OP_HANDLERS, + ): + self.assertTrue(set(table) <= WebSocketController._KNOWN_OPS) + + def test_failing_gauge_callbacks_do_not_break_the_scrape(self): + bad_manager = mock.Mock() + bad_manager.get_active_room.side_effect = RuntimeError("boom") + with ( + mock.patch.object( + app_metrics, "_room_managers", weakref.WeakSet([bad_manager]) + ), + mock.patch.object( + app_metrics, "_app_has_live_session", side_effect=RuntimeError("x") + ), + mock.patch.object( + app_metrics, "_applications", weakref.WeakSet([self._app]) + ), + self.assertLogs(get_logger(), level="WARNING") as captured, + ): + output = generate_latest().decode() + self.assertIn("digiscript_collab_rooms 0.0", output) + self.assertIn("digiscript_show_session_active NaN", output) + self.assertIn("digiscript_ws_connections", output) + self.assertTrue(any("active room" in m for m in captured.output)) + self.assertTrue(any("live show session" in m for m in captured.output)) + + def test_one_failing_app_does_not_hide_a_live_session_elsewhere(self): + good, bad = mock.Mock(), mock.Mock() + + def fake(app): + if app is bad: + raise RuntimeError("x") + return True + + with ( + mock.patch.object(app_metrics, "_app_has_live_session", fake), + mock.patch.object( + app_metrics, "_applications", weakref.WeakSet([bad, good]) + ), + self.assertLogs(get_logger(), level="WARNING"), ): - self.assertIn(op, WebSocketController._KNOWN_OPS) + self.assertEqual(1.0, app_metrics._live_session_value()) def test_info_gauge_carries_version(self): self.assertEqual( diff --git a/server/utils/script_room_manager.py b/server/utils/script_room_manager.py index d2c7ad439..e687c6421 100644 --- a/server/utils/script_room_manager.py +++ b/server/utils/script_room_manager.py @@ -533,6 +533,10 @@ def __init__(self, application: DigiScriptServer): self._room: ScriptRoom | None = None self._room_lock: asyncio.Lock = asyncio.Lock() self._maintenance_handle = None + # Metrics read live state from registered instances at scrape time. The + # application is registered here (rather than in DigiScriptServer) only + # because that file is outside this change; every real app builds a + # RoomManager at start-up. app_metrics.register_room_manager(self) app_metrics.register_application(application) diff --git a/server/utils/web/app_metrics.py b/server/utils/web/app_metrics.py index 5f95b4d48..e470a4fa6 100644 --- a/server/utils/web/app_metrics.py +++ b/server/utils/web/app_metrics.py @@ -1,8 +1,13 @@ """Prometheus gauges and counters answering "is the show healthy?". -Everything here is module-level so it is registered exactly once per process, -however many application instances are created (tests create many). The series -are exposed alongside the others at ``/debug/metrics``: +The metric objects are module-level singletons, so each series is registered +with Prometheus exactly once per process however many application instances are +created (tests create many). The room managers and applications the live gauges +read from are a different matter: each instance registers itself in a weak set, +so several can be tracked at once and they drop out when garbage collected. +Every collect-time callback is guarded so that one failing gauge can never break +the ``/debug/metrics`` scrape. The series are exposed alongside the others at +``/debug/metrics``: * ``digiscript_ws_connections`` - currently open WebSocket connections. * ``digiscript_ws_messages_total{op}`` - inbound WebSocket messages by OP. The @@ -93,12 +98,23 @@ def register_app_info() -> None: def _count_rooms() -> float: - return float(sum(1 for m in list(_room_managers) if m.get_active_room())) + """Count active rooms across registered managers, skipping any that fail.""" + count = 0 + for manager in _room_managers: + try: + if manager.get_active_room(): + count += 1 + except Exception: + get_logger().warning( + f"Could not read the active room of {type(manager).__name__} " + f"for metrics", + exc_info=True, + ) + return float(count) def _app_has_live_session(app: DigiScriptServer) -> bool: - setting = app.digi_settings.settings.get("current_show") - show_id = setting.get_value() if setting else None + show_id = app.digi_settings.get_sync("current_show") if not show_id: return False with app.get_db().sessionmaker() as session: @@ -107,18 +123,29 @@ def _app_has_live_session(app: DigiScriptServer) -> bool: def _live_session_value() -> float: + """Return 1 if any registered app has a live session, else 0. + + Each app is checked independently; one that fails is logged (with its type + and id) and skipped. The value is NaN only if no app could be read and at + least one failed, so a failure is never reported as a confident 0. + """ active = False - for app in list(_applications): + failed = 0 + for app in _applications: try: active = active or _app_has_live_session(app) except Exception: + failed += 1 get_logger().warning( - "Could not determine live show session state for metrics", + f"Could not determine live show session state of " + f"{type(app).__name__} {id(app):#x} for metrics", exc_info=True, ) - return float("nan") + if not active and failed: + return float("nan") return 1.0 if active else 0.0 collab_rooms.set_function(_count_rooms) show_session_active.set_function(_live_session_value) +register_app_info() From 19a991c5f8343980c7037185e24c0324c4832339 Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:24:53 +0100 Subject: [PATCH 3/5] Make the live-session gauge log every failing app, whatever the iteration order _live_session_value short-circuited once a live app had been seen, so a later failing app was never evaluated or logged. The WeakSet iteration order is address-dependent, which made test_one_failing_app_does_not_hide_a_live_session_elsewhere pass or fail with memory layout. Evaluate every app, and make the test check both orders explicitly. Co-Authored-By: Claude Sonnet 5.5 --- server/test/controllers/test_ws_controller.py | 22 ++++++++++++------- server/utils/web/app_metrics.py | 5 ++++- 2 files changed, 18 insertions(+), 9 deletions(-) diff --git a/server/test/controllers/test_ws_controller.py b/server/test/controllers/test_ws_controller.py index 16531661f..6023b5461 100644 --- a/server/test/controllers/test_ws_controller.py +++ b/server/test/controllers/test_ws_controller.py @@ -4030,14 +4030,20 @@ def fake(app): raise RuntimeError("x") return True - with ( - mock.patch.object(app_metrics, "_app_has_live_session", fake), - mock.patch.object( - app_metrics, "_applications", weakref.WeakSet([bad, good]) - ), - self.assertLogs(get_logger(), level="WARNING"), - ): - self.assertEqual(1.0, app_metrics._live_session_value()) + # A WeakSet iterates in an address-dependent order, so check both orders + # explicitly: the failing app must be logged and the live one still count. + for apps in ([bad, good], [good, bad]): + with ( + self.subTest(order=["bad" if a is bad else "good" for a in apps]), + mock.patch.object(app_metrics, "_app_has_live_session", fake), + mock.patch.object(app_metrics, "_applications", apps), + self.assertLogs(get_logger(), level="WARNING") as captured, + ): + self.assertEqual(1.0, app_metrics._live_session_value()) + self.assertTrue( + any("live show session" in m for m in captured.output), + captured.output, + ) def test_info_gauge_carries_version(self): self.assertEqual( diff --git a/server/utils/web/app_metrics.py b/server/utils/web/app_metrics.py index e470a4fa6..d442b7ef7 100644 --- a/server/utils/web/app_metrics.py +++ b/server/utils/web/app_metrics.py @@ -133,7 +133,10 @@ def _live_session_value() -> float: failed = 0 for app in _applications: try: - active = active or _app_has_live_session(app) + # Evaluate every app (no short-circuit) so a failing app is always + # logged, whatever order the weak set happens to iterate in. + has_live_session = _app_has_live_session(app) + active = active or has_live_session except Exception: failed += 1 get_logger().warning( From 1d93a1e465073b165185f2c37657220a1f9efe6e Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:49:59 +0100 Subject: [PATCH 4/5] Pin the show_session_active NaN rule: document it and test every combination Co-Authored-By: Claude Sonnet 5.5 --- server/test/controllers/test_ws_controller.py | 31 +++++++++++++++++++ server/utils/web/app_metrics.py | 6 ++-- 2 files changed, 35 insertions(+), 2 deletions(-) diff --git a/server/test/controllers/test_ws_controller.py b/server/test/controllers/test_ws_controller.py index 6023b5461..4dcee1b3f 100644 --- a/server/test/controllers/test_ws_controller.py +++ b/server/test/controllers/test_ws_controller.py @@ -9,6 +9,7 @@ import base64 import inspect import json +import math import os import textwrap import weakref @@ -4045,6 +4046,36 @@ def fake(app): captured.output, ) + def test_live_session_value_rule_pinned(self): + """1 if any app confirms active; else NaN if any failed; else 0.""" + false_app, true_app, bad_app = mock.Mock(), mock.Mock(), mock.Mock() + + def fake(app): + if app is bad_app: + raise RuntimeError("x") + return app is true_app + + cases = ( + ([false_app, bad_app], "nan"), + ([bad_app, false_app], "nan"), + ([true_app, bad_app], 1.0), + ([bad_app, true_app], 1.0), + ([false_app, false_app], 0.0), + ([], 0.0), + ) + for apps, expected in cases: + with ( + self.subTest(apps=[id(a) for a in apps]), + mock.patch.object(app_metrics, "_app_has_live_session", fake), + mock.patch.object(app_metrics, "_applications", apps), + mock.patch.object(app_metrics, "get_logger"), + ): + value = app_metrics._live_session_value() + if expected == "nan": + self.assertTrue(math.isnan(value)) + else: + self.assertEqual(expected, value) + def test_info_gauge_carries_version(self): self.assertEqual( 1.0, diff --git a/server/utils/web/app_metrics.py b/server/utils/web/app_metrics.py index d442b7ef7..d3af0ec48 100644 --- a/server/utils/web/app_metrics.py +++ b/server/utils/web/app_metrics.py @@ -126,8 +126,10 @@ def _live_session_value() -> float: """Return 1 if any registered app has a live session, else 0. Each app is checked independently; one that fails is logged (with its type - and id) and skipped. The value is NaN only if no app could be read and at - least one failed, so a failure is never reported as a confident 0. + and id) and counted as unknown. The rule is: 1 if any app confirms a live + session; otherwise NaN if any app failed (the failed app's state is unknown, + so "no live session anywhere" is not confirmed, even if another app + confidently reported none); otherwise 0. """ active = False failed = 0 From d106e36c82ebb7e54811f2a8f2947735018472da Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Fri, 23 Oct 2026 00:01:18 +0100 Subject: [PATCH 5/5] Register the application for metrics in DigiScriptServer.__init__, not RoomManager Co-Authored-By: Claude Sonnet 5.5 --- server/digi_server/app_server.py | 4 ++++ server/test/digi_server/test_digi_server.py | 6 ++++++ server/test/utils/script/test_script_room.py | 13 ++++++++++++- server/utils/script_room_manager.py | 5 ----- 4 files changed, 22 insertions(+), 6 deletions(-) diff --git a/server/digi_server/app_server.py b/server/digi_server/app_server.py index 0a6b2f961..9e69ef77c 100644 --- a/server/digi_server/app_server.py +++ b/server/digi_server/app_server.py @@ -43,6 +43,7 @@ from utils.module_discovery import get_resource_path, is_frozen from utils.script_room_manager import RoomManager from utils.version_checker import VersionChecker +from utils.web import app_metrics from utils.web.jwt_service import JWTService from utils.web.pending_disconnects import PendingDisconnects from utils.web.request_size_metrics import observe_http_request @@ -346,6 +347,9 @@ class AlembicVersion(self._db.Model): cookie_secret=self._get_cookie_secret(), login_url="/login", ) + # Fully constructed: let the metrics read this app's live state at scrape + # time (weakly referenced, so no teardown is needed). + app_metrics.register_application(self) _METRICS_WARN_INTERVAL = 60.0 _last_metrics_warning = float("-inf") diff --git a/server/test/digi_server/test_digi_server.py b/server/test/digi_server/test_digi_server.py index 38962e128..7d1a92877 100644 --- a/server/test/digi_server/test_digi_server.py +++ b/server/test/digi_server/test_digi_server.py @@ -14,10 +14,16 @@ from models.settings import SystemSettings from models.user import User from test.conftest import DigiScriptTestCase +from utils.web import app_metrics from utils.web.request_size_metrics import http_request_body_bytes class TestDigiScriptServer(DigiScriptTestCase): + def test_server_registers_itself_for_app_metrics(self): + """Constructing a DigiScriptServer registers it (once) with app_metrics.""" + registered = [app for app in app_metrics._applications if app is self._app] + self.assertEqual([self._app], registered) + def test_debug(self): token = self._create_and_login_admin() response = self.fetch("/debug", headers={"Authorization": f"Bearer {token}"}) diff --git a/server/test/utils/script/test_script_room.py b/server/test/utils/script/test_script_room.py index 2c142ed7a..39c188f23 100644 --- a/server/test/utils/script/test_script_room.py +++ b/server/test/utils/script/test_script_room.py @@ -10,7 +10,7 @@ import tempfile import time import uuid -from unittest.mock import AsyncMock, patch +from unittest.mock import AsyncMock, MagicMock, patch import pycrdt import pytest @@ -1206,3 +1206,14 @@ async def test_save_progress_closed_client_logged_at_debug(self): await room._broadcast_save_progress(1, 2) get_logger.return_value.debug.assert_called_once() assert "11" in get_logger.return_value.debug.call_args[0][0] + + +class TestRoomManagerRegistration: + """Registration with app_metrics: managers register, applications do not.""" + + def test_room_manager_registers_itself_but_not_the_application(self): + app = MagicMock() + manager = RoomManager(app) + + assert manager in app_metrics._room_managers + assert app not in app_metrics._applications diff --git a/server/utils/script_room_manager.py b/server/utils/script_room_manager.py index e687c6421..d9bb5d43d 100644 --- a/server/utils/script_room_manager.py +++ b/server/utils/script_room_manager.py @@ -533,12 +533,7 @@ def __init__(self, application: DigiScriptServer): self._room: ScriptRoom | None = None self._room_lock: asyncio.Lock = asyncio.Lock() self._maintenance_handle = None - # Metrics read live state from registered instances at scrape time. The - # application is registered here (rather than in DigiScriptServer) only - # because that file is outside this change; every real app builds a - # RoomManager at start-up. app_metrics.register_room_manager(self) - app_metrics.register_application(application) def start(self): self._schedule_maintenance()