From 41cf074bafc02bc0ca8adcdd7da7256419434fa0 Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 18:35:54 +0100 Subject: [PATCH 1/3] Capture asyncio, alembic and warnings logs and install crash hooks Route asyncio/alembic/py.warnings through the same file, buffer and level handling as the DigiScript loggers, and install sys/threading/asyncio exception hooks that log tracebacks and increment digiscript_unhandled_errors_total{source}. Co-Authored-By: Claude Sonnet 5.5 --- server/digi_server/logger.py | 94 +++++++++++++ server/main.py | 8 +- server/test/digi_server/test_logger.py | 185 ++++++++++++++++++++++++- 3 files changed, 285 insertions(+), 2 deletions(-) diff --git a/server/digi_server/logger.py b/server/digi_server/logger.py index 3c076c5c7..50689a524 100644 --- a/server/digi_server/logger.py +++ b/server/digi_server/logger.py @@ -1,7 +1,10 @@ import logging +import sys +import threading from logging.handlers import RotatingFileHandler from typing import Optional +from prometheus_client import Counter from tornado.log import LogFormatter from utils.log_buffer import ( @@ -16,9 +19,100 @@ logging.getLogger("tornado.access"), logging.getLogger("tornado.application"), logging.getLogger("tornado.general"), + # Stdlib/third-party loggers whose output would otherwise only reach stderr + # (e.g. "Exception in callback ..." from fire-and-forget loop callbacks). + logging.getLogger("asyncio"), + logging.getLogger("alembic"), + logging.getLogger("py.warnings"), logger, ] +UNHANDLED_ERRORS = Counter( + "digiscript_unhandled_errors_total", + "Unhandled exceptions caught by the global crash hooks", + ["source"], +) + +# Originals saved when the hooks are installed; empty while not installed. +_saved_hooks: dict = {} + + +def _count_unhandled(source: str) -> None: + UNHANDLED_ERRORS.labels(source=source).inc() + + +def _sys_excepthook(exc_type, exc_value, exc_tb): + if issubclass(exc_type, (KeyboardInterrupt, SystemExit)): + original = _saved_hooks.get("sys") or sys.__excepthook__ + original(exc_type, exc_value, exc_tb) + return + _count_unhandled("sys_excepthook") + logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_tb)) + + +def _threading_excepthook(args): + if issubclass(args.exc_type, SystemExit): + return + _count_unhandled("thread") + thread_name = args.thread.name if args.thread else "" + logger.error( + "Unhandled exception in thread %s", + thread_name, + exc_info=(args.exc_type, args.exc_value, args.exc_traceback), + ) + + +def _asyncio_exception_handler(loop, context): + exc = context.get("exception") + if isinstance(exc, (KeyboardInterrupt, SystemExit)): + loop.default_exception_handler(context) + return + _count_unhandled("asyncio") + message = context.get("message", "Unhandled exception in event loop") + extras = "".join( + f"\n{key}: {context[key]!r}" + for key in ("future", "task", "handle", "protocol", "transport") + if key in context + ) + exc_info = (type(exc), exc, exc.__traceback__) if exc is not None else None + logger.error("%s%s", message, extras, exc_info=exc_info) + + +def install_exception_hooks(loop=None) -> None: + """Route otherwise-invisible crashes into the DigiScript log. + + Installs ``sys.excepthook`` and ``threading.excepthook`` (once per process) + and the given loop's exception handler, and enables + ``logging.captureWarnings`` so ``warnings.warn`` reaches the ``py.warnings`` + logger. Each unhandled error increments + ``digiscript_unhandled_errors_total{source}``. ``KeyboardInterrupt`` and + ``SystemExit`` keep their default behaviour. Safe to call repeatedly. + + :param loop: Event loop to attach the exception handler to, or ``None`` to + skip the loop handler. + """ + if not _saved_hooks: + _saved_hooks["sys"] = sys.excepthook + _saved_hooks["thread"] = threading.excepthook + sys.excepthook = _sys_excepthook + threading.excepthook = _threading_excepthook + logging.captureWarnings(True) + if loop is not None: + loop.set_exception_handler(_asyncio_exception_handler) + + +def uninstall_exception_hooks() -> None: + """Restore the hooks replaced by :func:`install_exception_hooks`. + + The event loop handler is per-loop and is not touched. + """ + if not _saved_hooks: + return + sys.excepthook = _saved_hooks["sys"] + threading.excepthook = _saved_hooks["thread"] + _saved_hooks.clear() + logging.captureWarnings(False) + def get_logger(name: Optional[str] = None): if name: diff --git a/server/main.py b/server/main.py index ccbf5b02a..c6158e31d 100755 --- a/server/main.py +++ b/server/main.py @@ -32,7 +32,11 @@ def ensure_writable_db_path(path): from digi_server.app_server import DigiScriptServer -from digi_server.logger import add_logging_level, get_logger +from digi_server.logger import ( + add_logging_level, + get_logger, + install_exception_hooks, +) add_logging_level("TRACE", logging.DEBUG - 5) @@ -51,6 +55,8 @@ def ensure_writable_db_path(path): async def main(): + install_exception_hooks(asyncio.get_running_loop()) + # Parse command-line options try: parse_command_line() diff --git a/server/test/digi_server/test_logger.py b/server/test/digi_server/test_logger.py index 617be3742..42d048162 100644 --- a/server/test/digi_server/test_logger.py +++ b/server/test/digi_server/test_logger.py @@ -1,7 +1,22 @@ +import asyncio +import gc import logging +import sys +import threading import unittest +import warnings -from digi_server.logger import CLIENT_LEVEL_MAP, map_client_level +from tornado.ioloop import IOLoop + +from digi_server.logger import ( + ALL_LOGGERS, + CLIENT_LEVEL_MAP, + UNHANDLED_ERRORS, + install_exception_hooks, + logger, + map_client_level, + uninstall_exception_hooks, +) class TestMapClientLevel(unittest.TestCase): @@ -38,3 +53,171 @@ def test_case_insensitive(self): def test_client_level_map_contains_all_expected_keys(self): expected = {"TRACE", "DEBUG", "INFO", "WARN", "ERROR", "SILENT"} self.assertEqual(expected, set(CLIENT_LEVEL_MAP.keys())) + + +class _ListHandler(logging.Handler): + def __init__(self): + super().__init__(logging.DEBUG) + self.records = [] + + def emit(self, record): + self.records.append(record) + + +class TestExceptionHooks(unittest.IsolatedAsyncioTestCase): + def setUp(self): + self._excepthook = sys.excepthook + self._thread_hook = threading.excepthook + self._warnings_captured = logging._warnings_showwarning is not None + self._levels = {lg.name: lg.level for lg in ALL_LOGGERS} + self.handler = _ListHandler() + for lg in ALL_LOGGERS: + lg.addHandler(self.handler) + logger.setLevel(logging.DEBUG) + uninstall_exception_hooks() + + def tearDown(self): + uninstall_exception_hooks() + sys.excepthook = self._excepthook + threading.excepthook = self._thread_hook + logging.captureWarnings(self._warnings_captured) + for lg in ALL_LOGGERS: + lg.removeHandler(self.handler) + lg.setLevel(self._levels[lg.name]) + + @staticmethod + def _count(source): + return UNHANDLED_ERRORS.labels(source=source)._value.get() + + def test_all_loggers_cover_stdlib_sources(self): + names = {lg.name for lg in ALL_LOGGERS} + self.assertTrue({"asyncio", "alembic", "py.warnings", "DigiScript"} <= names) + self.assertNotIn("sqlalchemy.engine", names) + self.assertNotIn("root", names) + + def test_install_is_idempotent_and_restores_originals(self): + install_exception_hooks() + first = sys.excepthook + install_exception_hooks() + self.assertIs(first, sys.excepthook) + self.assertIsNot(self._excepthook, sys.excepthook) + uninstall_exception_hooks() + self.assertIs(self._excepthook, sys.excepthook) + self.assertIs(self._thread_hook, threading.excepthook) + + def test_sys_excepthook_logs_critical_and_counts(self): + install_exception_hooks() + before = self._count("sys_excepthook") + try: + raise ValueError("boom-sys") + except ValueError: + sys.excepthook(*sys.exc_info()) + self.assertEqual(before + 1, self._count("sys_excepthook")) + records = [r for r in self.handler.records if r.name == "DigiScript"] + self.assertEqual(1, len(records)) + self.assertEqual(logging.CRITICAL, records[0].levelno) + self.assertIn("boom-sys", repr(records[0].exc_info[1])) + + def test_sys_excepthook_keeps_default_for_keyboard_interrupt(self): + calls = [] + sys.excepthook = lambda *a: calls.append(a) + install_exception_hooks() + before = self._count("sys_excepthook") + sys.excepthook(KeyboardInterrupt, KeyboardInterrupt(), None) + self.assertEqual(1, len(calls)) + self.assertEqual(before, self._count("sys_excepthook")) + self.assertEqual([], self.handler.records) + + def test_thread_excepthook_logs_and_counts(self): + install_exception_hooks() + before = self._count("thread") + + def target(): + raise RuntimeError("boom-thread") + + t = threading.Thread(target=target, name="obs-test-thread") + t.start() + t.join() + self.assertEqual(before + 1, self._count("thread")) + records = [r for r in self.handler.records if r.name == "DigiScript"] + self.assertEqual(1, len(records)) + self.assertEqual(logging.ERROR, records[0].levelno) + self.assertIn("obs-test-thread", records[0].getMessage()) + self.assertIn("boom-thread", repr(records[0].exc_info[1])) + + async def test_loop_callback_exception_logged_with_traceback(self): + loop = asyncio.get_running_loop() + install_exception_hooks(loop) + before = self._count("asyncio") + done = asyncio.Event() + + def bad_callback(): + loop.call_soon(done.set) + raise RuntimeError("boom-callback") + + # Raw asyncio callbacks (call_soon/call_later) reach the loop handler. + # IOLoop.add_callback catches exceptions itself and logs them to + # tornado.application, which is already attached to the same handlers. + loop.call_soon(bad_callback) + await asyncio.wait_for(done.wait(), 5) + await asyncio.sleep(0.05) + self.assertEqual(before + 1, self._count("asyncio")) + records = [ + r for r in self.handler.records if "boom-callback" in repr(r.exc_info) + ] + # Exactly one record carries the traceback (tornado's own logger may + # also report the callback; filter to ours and the asyncio logger). + ours = [r for r in records if r.name == "DigiScript"] + self.assertEqual(1, len(ours)) + self.assertEqual(logging.ERROR, ours[0].levelno) + self.assertIsNotNone(ours[0].exc_info[2]) + self.assertEqual([], [r for r in records if r.name == "asyncio"]) + + async def test_ioloop_add_callback_exception_logged_once(self): + loop = asyncio.get_running_loop() + install_exception_hooks(loop) + before = self._count("asyncio") + done = asyncio.Event() + + def bad_callback(): + loop.call_soon(done.set) + raise RuntimeError("boom-add-callback") + + IOLoop.current().add_callback(bad_callback) + await asyncio.wait_for(done.wait(), 5) + await asyncio.sleep(0.05) + records = [ + r for r in self.handler.records if "boom-add-callback" in repr(r.exc_info) + ] + self.assertEqual(1, len(records)) + self.assertEqual("tornado.application", records[0].name) + self.assertEqual(before, self._count("asyncio")) + + async def test_unretrieved_task_exception_logged_once(self): + loop = asyncio.get_running_loop() + install_exception_hooks(loop) + before = self._count("asyncio") + + async def bad(): + raise RuntimeError("boom-task") + + task = asyncio.create_task(bad()) + await asyncio.sleep(0.05) + del task + gc.collect() + await asyncio.sleep(0.05) + self.assertEqual(before + 1, self._count("asyncio")) + records = [r for r in self.handler.records if "boom-task" in repr(r.exc_info)] + self.assertEqual(1, len(records)) + self.assertIn("Task exception was never retrieved", records[0].getMessage()) + self.assertIn("future: Date: Thu, 22 Oct 2026 23:17:46 +0100 Subject: [PATCH 2/3] Harden exception hooks: lock, chaining, loop/warnings restore, pre-registered metric Co-Authored-By: Claude Sonnet 5.5 --- server/digi_server/logger.py | 166 ++++++++++++++++++------- server/main.py | 4 + server/test/digi_server/test_logger.py | 118 +++++++++++++++++- 3 files changed, 243 insertions(+), 45 deletions(-) diff --git a/server/digi_server/logger.py b/server/digi_server/logger.py index 50689a524..74c93f398 100644 --- a/server/digi_server/logger.py +++ b/server/digi_server/logger.py @@ -1,6 +1,7 @@ import logging import sys import threading +import weakref from logging.handlers import RotatingFileHandler from typing import Optional @@ -32,86 +33,167 @@ "Unhandled exceptions caught by the global crash hooks", ["source"], ) - -# Originals saved when the hooks are installed; empty while not installed. +UNHANDLED_ERROR_SOURCES = ("sys_excepthook", "asyncio", "thread") +# Pre-register every label series at 0 so Prometheus increase() sees the very +# first error for each source. +for _source in UNHANDLED_ERROR_SOURCES: + UNHANDLED_ERRORS.labels(source=_source) + +# Guards install/uninstall; the state below is only touched while holding it. +_hooks_lock = threading.RLock() +# Previous hooks / warning state, saved at install; empty while not installed. _saved_hooks: dict = {} +# Event loops we attached a handler to, mapped to the handler they had before. +_saved_loop_handlers: "weakref.WeakKeyDictionary" = weakref.WeakKeyDictionary() def _count_unhandled(source: str) -> None: UNHANDLED_ERRORS.labels(source=source).inc() +def _chain(previous, default, *args) -> None: + """Call a pre-existing hook, skipping interpreter defaults and ourselves. + + The defaults only print to stderr, which would duplicate the log record. + """ + if previous is None or previous is default: + return + try: + previous(*args) + except Exception: # noqa: BLE001 - a broken foreign hook must not mask the crash + logger.exception("Chained exception hook failed") + + def _sys_excepthook(exc_type, exc_value, exc_tb): + previous = _saved_hooks.get("sys") + if previous is _sys_excepthook: + previous = None if issubclass(exc_type, (KeyboardInterrupt, SystemExit)): - original = _saved_hooks.get("sys") or sys.__excepthook__ - original(exc_type, exc_value, exc_tb) + (previous or sys.__excepthook__)(exc_type, exc_value, exc_tb) return _count_unhandled("sys_excepthook") - logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_tb)) + try: + logger.critical("Uncaught exception", exc_info=(exc_type, exc_value, exc_tb)) + finally: + _chain(previous, sys.__excepthook__, exc_type, exc_value, exc_tb) def _threading_excepthook(args): + previous = _saved_hooks.get("thread") + if previous is _threading_excepthook: + previous = None if issubclass(args.exc_type, SystemExit): + _chain(previous, threading.__excepthook__, args) return _count_unhandled("thread") - thread_name = args.thread.name if args.thread else "" - logger.error( - "Unhandled exception in thread %s", - thread_name, - exc_info=(args.exc_type, args.exc_value, args.exc_traceback), - ) + try: + thread_name = args.thread.name if args.thread else "" + logger.error( + "Unhandled exception in thread %s", + thread_name, + exc_info=(args.exc_type, args.exc_value, args.exc_traceback), + ) + finally: + _chain(previous, threading.__excepthook__, args) + + +def _safe_repr(value) -> str: + try: + return repr(value) + except Exception: # noqa: BLE001 - arbitrary __repr__ must not break logging + return f"" def _asyncio_exception_handler(loop, context): exc = context.get("exception") + previous = _saved_loop_handlers.get(loop) + if previous is _asyncio_exception_handler: + previous = None if isinstance(exc, (KeyboardInterrupt, SystemExit)): loop.default_exception_handler(context) return - _count_unhandled("asyncio") - message = context.get("message", "Unhandled exception in event loop") - extras = "".join( - f"\n{key}: {context[key]!r}" - for key in ("future", "task", "handle", "protocol", "transport") - if key in context - ) - exc_info = (type(exc), exc, exc.__traceback__) if exc is not None else None - logger.error("%s%s", message, extras, exc_info=exc_info) + try: + message = context.get("message", "Unhandled exception in event loop") + extras = "".join( + f"\n{key}: {_safe_repr(context[key])}" + for key in ("future", "task", "handle", "protocol", "transport") + if key in context + ) + if exc is None: + # Message-only context (e.g. slow callback): a diagnostic, not a crash. + logger.warning("%s%s", message, extras) + else: + _count_unhandled("asyncio") + logger.error( + "%s%s", + message, + extras, + exc_info=(type(exc), exc, exc.__traceback__), + ) + except Exception: # noqa: BLE001 - the handler itself must never raise + try: + loop.default_exception_handler(context) + except Exception: # noqa: BLE001, S110 + pass + if previous is not None: + _chain(previous, None, loop, context) def install_exception_hooks(loop=None) -> None: """Route otherwise-invisible crashes into the DigiScript log. - Installs ``sys.excepthook`` and ``threading.excepthook`` (once per process) - and the given loop's exception handler, and enables - ``logging.captureWarnings`` so ``warnings.warn`` reaches the ``py.warnings`` - logger. Each unhandled error increments - ``digiscript_unhandled_errors_total{source}``. ``KeyboardInterrupt`` and - ``SystemExit`` keep their default behaviour. Safe to call repeatedly. + Installs ``sys.excepthook`` and ``threading.excepthook`` and the given + loop's exception handler, and enables ``logging.captureWarnings`` so + ``warnings.warn`` reaches the ``py.warnings`` logger. Each unhandled error + increments ``digiscript_unhandled_errors_total{source}``. Pre-existing hooks + are chained to after logging (interpreter defaults are skipped to avoid + duplicate stderr output); ``KeyboardInterrupt`` and ``SystemExit`` keep + their default behaviour. Idempotent and thread-safe. :param loop: Event loop to attach the exception handler to, or ``None`` to skip the loop handler. """ - if not _saved_hooks: - _saved_hooks["sys"] = sys.excepthook - _saved_hooks["thread"] = threading.excepthook - sys.excepthook = _sys_excepthook - threading.excepthook = _threading_excepthook - logging.captureWarnings(True) - if loop is not None: - loop.set_exception_handler(_asyncio_exception_handler) + with _hooks_lock: + if not _saved_hooks: + _saved_hooks["warnings"] = logging._warnings_showwarning is not None + if sys.excepthook is not _sys_excepthook: + _saved_hooks["sys"] = sys.excepthook + else: + _saved_hooks["sys"] = sys.__excepthook__ + if threading.excepthook is not _threading_excepthook: + _saved_hooks["thread"] = threading.excepthook + else: + _saved_hooks["thread"] = threading.__excepthook__ + sys.excepthook = _sys_excepthook + threading.excepthook = _threading_excepthook + logging.captureWarnings(True) + if loop is not None and loop not in _saved_loop_handlers: + current = loop.get_exception_handler() + if current is not _asyncio_exception_handler: + _saved_loop_handlers[loop] = current + loop.set_exception_handler(_asyncio_exception_handler) def uninstall_exception_hooks() -> None: - """Restore the hooks replaced by :func:`install_exception_hooks`. + """Undo :func:`install_exception_hooks`. - The event loop handler is per-loop and is not touched. + Restores ``sys.excepthook``, ``threading.excepthook``, the previous + handler of every loop we attached to (unless closed) and the prior + ``captureWarnings`` state. Idempotent and thread-safe. """ - if not _saved_hooks: - return - sys.excepthook = _saved_hooks["sys"] - threading.excepthook = _saved_hooks["thread"] - _saved_hooks.clear() - logging.captureWarnings(False) + with _hooks_lock: + if not _saved_hooks: + return + sys.excepthook = _saved_hooks["sys"] + threading.excepthook = _saved_hooks["thread"] + logging.captureWarnings(_saved_hooks["warnings"]) + for loop, previous in list(_saved_loop_handlers.items()): + if not loop.is_closed() and ( + loop.get_exception_handler() is _asyncio_exception_handler + ): + loop.set_exception_handler(previous) + _saved_loop_handlers.clear() + _saved_hooks.clear() def get_logger(name: Optional[str] = None): diff --git a/server/main.py b/server/main.py index c6158e31d..5b3fffb4f 100755 --- a/server/main.py +++ b/server/main.py @@ -55,6 +55,10 @@ def ensure_writable_db_path(path): async def main(): + # Installed first so crashes during startup are counted and reach stderr + # (logging's last-resort handler). The durable file log and in-app buffer + # handlers are attached later by DigiScriptServer.configure(), so errors in + # this early window are not written to the log file. install_exception_hooks(asyncio.get_running_loop()) # Parse command-line options diff --git a/server/test/digi_server/test_logger.py b/server/test/digi_server/test_logger.py index 42d048162..9dd256070 100644 --- a/server/test/digi_server/test_logger.py +++ b/server/test/digi_server/test_logger.py @@ -11,6 +11,7 @@ from digi_server.logger import ( ALL_LOGGERS, CLIENT_LEVEL_MAP, + UNHANDLED_ERROR_SOURCES, UNHANDLED_ERRORS, install_exception_hooks, logger, @@ -89,9 +90,22 @@ def tearDown(self): def _count(source): return UNHANDLED_ERRORS.labels(source=source)._value.get() + def test_label_series_preregistered_at_zero(self): + samples = { + s.labels["source"]: s.value + for m in UNHANDLED_ERRORS.collect() + for s in m.samples + if s.name == "digiscript_unhandled_errors_total" + } + self.assertEqual( + {"sys_excepthook", "asyncio", "thread"}, set(UNHANDLED_ERROR_SOURCES) + ) + for source in UNHANDLED_ERROR_SOURCES: + self.assertIn(source, samples) + def test_all_loggers_cover_stdlib_sources(self): names = {lg.name for lg in ALL_LOGGERS} - self.assertTrue({"asyncio", "alembic", "py.warnings", "DigiScript"} <= names) + self.assertLessEqual({"asyncio", "alembic", "py.warnings", "DigiScript"}, names) self.assertNotIn("sqlalchemy.engine", names) self.assertNotIn("root", names) @@ -129,6 +143,8 @@ def test_sys_excepthook_keeps_default_for_keyboard_interrupt(self): self.assertEqual([], self.handler.records) def test_thread_excepthook_logs_and_counts(self): + # Replace pytest's own thread hook so it doesn't flag the intended crash. + threading.excepthook = lambda _args: None install_exception_hooks() before = self._count("thread") @@ -213,11 +229,107 @@ async def bad(): self.assertIn("future: took 1.0s"}) + self.assertEqual(before, self._count("asyncio")) + records = [r for r in self.handler.records if r.name == "DigiScript"] + self.assertEqual(1, len(records)) + self.assertEqual(logging.WARNING, records[0].levelno) + + async def test_asyncio_handler_survives_failing_repr(self): + class BadRepr: + def __repr__(self): + raise RuntimeError("repr failed") + + loop = asyncio.get_running_loop() + install_exception_hooks(loop) + before = self._count("asyncio") + loop.call_exception_handler( + { + "message": "boom", + "exception": RuntimeError("boom-repr"), + "future": BadRepr(), + } + ) + self.assertEqual(before + 1, self._count("asyncio")) + records = [r for r in self.handler.records if "boom-repr" in repr(r.exc_info)] + self.assertEqual(1, len(records)) + self.assertIn("unrepresentable BadRepr", records[0].getMessage()) From c6f4bace3467e51a77e26b5fb7420bc62d2533e9 Mon Sep 17 00:00:00 2001 From: Tim Bradgate Date: Thu, 22 Oct 2026 23:55:05 +0100 Subject: [PATCH 3/3] Make exception-hook race test deterministic and tighten uninstall/chaining Co-Authored-By: Claude Sonnet 5.5 --- server/digi_server/logger.py | 23 ++++----- server/test/digi_server/test_logger.py | 65 +++++++++++++++++++++++--- 2 files changed, 70 insertions(+), 18 deletions(-) diff --git a/server/digi_server/logger.py b/server/digi_server/logger.py index 74c93f398..23c71afa4 100644 --- a/server/digi_server/logger.py +++ b/server/digi_server/logger.py @@ -52,7 +52,7 @@ def _count_unhandled(source: str) -> None: def _chain(previous, default, *args) -> None: - """Call a pre-existing hook, skipping interpreter defaults and ourselves. + """Call a pre-existing hook, skipping the interpreter default. The defaults only print to stderr, which would duplicate the log record. """ @@ -66,8 +66,6 @@ def _chain(previous, default, *args) -> None: def _sys_excepthook(exc_type, exc_value, exc_tb): previous = _saved_hooks.get("sys") - if previous is _sys_excepthook: - previous = None if issubclass(exc_type, (KeyboardInterrupt, SystemExit)): (previous or sys.__excepthook__)(exc_type, exc_value, exc_tb) return @@ -80,8 +78,6 @@ def _sys_excepthook(exc_type, exc_value, exc_tb): def _threading_excepthook(args): previous = _saved_hooks.get("thread") - if previous is _threading_excepthook: - previous = None if issubclass(args.exc_type, SystemExit): _chain(previous, threading.__excepthook__, args) return @@ -107,10 +103,11 @@ def _safe_repr(value) -> str: def _asyncio_exception_handler(loop, context): exc = context.get("exception") previous = _saved_loop_handlers.get(loop) - if previous is _asyncio_exception_handler: - previous = None if isinstance(exc, (KeyboardInterrupt, SystemExit)): - loop.default_exception_handler(context) + if previous is not None: + previous(loop, context) + else: + loop.default_exception_handler(context) return try: message = context.get("message", "Unhandled exception in event loop") @@ -184,10 +181,14 @@ def uninstall_exception_hooks() -> None: with _hooks_lock: if not _saved_hooks: return - sys.excepthook = _saved_hooks["sys"] - threading.excepthook = _saved_hooks["thread"] + # Only restore hooks that are still ours; never clobber a wrapper + # installed after us. + if sys.excepthook is _sys_excepthook: + sys.excepthook = _saved_hooks["sys"] + if threading.excepthook is _threading_excepthook: + threading.excepthook = _saved_hooks["thread"] logging.captureWarnings(_saved_hooks["warnings"]) - for loop, previous in list(_saved_loop_handlers.items()): + for loop, previous in _saved_loop_handlers.items(): if not loop.is_closed() and ( loop.get_exception_handler() is _asyncio_exception_handler ): diff --git a/server/test/digi_server/test_logger.py b/server/test/digi_server/test_logger.py index 9dd256070..1462a27a5 100644 --- a/server/test/digi_server/test_logger.py +++ b/server/test/digi_server/test_logger.py @@ -3,11 +3,14 @@ import logging import sys import threading +import time import unittest import warnings +from unittest import mock from tornado.ioloop import IOLoop +import digi_server.logger as logger_module from digi_server.logger import ( ALL_LOGGERS, CLIENT_LEVEL_MAP, @@ -265,14 +268,62 @@ def test_reinstall_cycles_restore_originals(self): self.assertIs(self._thread_hook, threading.excepthook) def test_concurrent_install_is_race_free(self): - threads = [threading.Thread(target=install_exception_hooks) for _ in range(8)] - for t in threads: - t.start() - for t in threads: - t.join() + """Force the check-then-act interleaving: every thread passes the + "is anything saved yet" check before any of them writes.""" + + class SlowCheckDict(dict): + def __len__(self): + size = super().__len__() + time.sleep(0.05) # widen the window after the emptiness check + return size + + sentinel_sys = lambda *_a: None # noqa: E731 + sentinel_thread = lambda _a: None # noqa: E731 + sys.excepthook = sentinel_sys + threading.excepthook = sentinel_thread + barrier = threading.Barrier(4) + + def run(): + barrier.wait() + install_exception_hooks() + + with mock.patch.object(logger_module, "_saved_hooks", SlowCheckDict()): + threads = [threading.Thread(target=run) for _ in range(4)] + for t in threads: + t.start() + for t in threads: + t.join() + uninstall_exception_hooks() + # Without the lock a late thread saves one of our own hooks / defaults + # as the "original", so the sentinels would not be restored. + self.assertIs(sentinel_sys, sys.excepthook) + self.assertIs(sentinel_thread, threading.excepthook) + + def test_uninstall_does_not_clobber_later_hooks(self): + install_exception_hooks() + + def later_sys(*_a): + pass + + def later_thread(_a): + pass + + sys.excepthook = later_sys + threading.excepthook = later_thread uninstall_exception_hooks() - self.assertIs(self._excepthook, sys.excepthook) - self.assertIs(self._thread_hook, threading.excepthook) + self.assertIs(later_sys, sys.excepthook) + self.assertIs(later_thread, threading.excepthook) + + async def test_keyboard_interrupt_in_loop_uses_previous_handler(self): + loop = asyncio.get_running_loop() + calls = [] + loop.set_exception_handler(lambda lp, ctx: calls.append(ctx)) + install_exception_hooks(loop) + before = self._count("asyncio") + context = {"message": "ki", "exception": KeyboardInterrupt()} + loop.call_exception_handler(context) + self.assertEqual([context], calls) + self.assertEqual(before, self._count("asyncio")) def test_keyboard_interrupt_does_not_recurse_after_reinstall(self): calls = []