From 32a349b2f913ecd38ef5774b61cc5a22cf932526 Mon Sep 17 00:00:00 2001 From: wakonig_k Date: Wed, 9 Sep 2026 11:16:46 +0200 Subject: [PATCH] fix(logger): fallback to stderr on loguru errors to avoid cyclic logs --- bec_widgets/applications/companion_app.py | 97 ++++++- tests/unit_tests/test_companion_app.py | 338 +++++++++++++++++++++- 2 files changed, 422 insertions(+), 13 deletions(-) diff --git a/bec_widgets/applications/companion_app.py b/bec_widgets/applications/companion_app.py index 977f4c09..7b7e438c 100644 --- a/bec_widgets/applications/companion_app.py +++ b/bec_widgets/applications/companion_app.py @@ -12,7 +12,11 @@ import os import signal import sys import traceback +from collections.abc import Callable from contextlib import redirect_stderr, redirect_stdout +from inspect import currentframe +from threading import RLock, local +from typing import TextIO import darkdetect import shiboken6 @@ -39,24 +43,87 @@ startup_profiler.mark("module imports") class SimpleFileLikeFromLogOutputFunc: - def __init__(self, log_func): + """Log complete lines on flush, bypassing logging for Loguru's own diagnostics. + + ``fallback_stream`` must bypass the logger; ``None`` discards fallback output. + Coroutine sinks must not print to redirected streams: their execution has no + synchronous logging frame to identify (their Loguru failure reports do). + """ + + def __init__( + self, log_func: Callable[[str], object], fallback_stream: TextIO | None = None + ) -> None: self._log_func = log_func self._buffer = [] + self._fallback_stream = fallback_stream + self._buffer_lock = RLock() + self._local = local() - def write(self, buffer): - self._buffer.append(buffer) + def write(self, buffer: str) -> int: + """Buffer output, sending writes made during logging to the fallback stream.""" + if self._is_logging(): + self._write_fallback(buffer) + else: + with self._buffer_lock: + self._buffer.append(buffer) + return len(buffer) - def flush(self): - lines, _, remaining = "".join(self._buffer).rpartition("\n") - if lines: + def flush(self) -> None: + """Log complete lines once, preserving the unfinished line for the next flush.""" + if self._is_logging(): + return + with self._buffer_lock: + lines, _, remaining = "".join(self._buffer).rpartition("\n") + # Detach before logging: sinks can write or flush this stream themselves. + self._buffer = [remaining] + if not lines: + return + self._local.logging = True + try: self._log_func(lines) - self._buffer = [remaining] + except Exception: + # Reporting this through the logger would retry the failing sink. + self._write_fallback(f"{lines}\n") + finally: + self._local.logging = False + + def _is_logging(self) -> bool: + if getattr(self._local, "logging", False): + return True + # Loguru reports failures to sys.stderr, also from direct logger calls and + # queued sinks, when our flush guard is inactive. It has no configurable + # diagnostic stream, so recognize its active logging/error-reporting frames. + # Match only _log in loguru._logger: logger.catch wraps ordinary application code. + frame = currentframe() + try: + while frame is not None: + module = frame.f_globals.get("__name__") + if module in ("loguru._handler", "loguru._error_interceptor") or ( + module == "loguru._logger" and frame.f_code.co_name == "_log" + ): + return True + frame = frame.f_back + return False + finally: + del frame + + def _write_fallback(self, buffer: str) -> None: + if self._fallback_stream is None: + return + try: + self._fallback_stream.write(buffer) + self._fallback_stream.flush() + except (OSError, ValueError): + # The original stream can also be unavailable (e.g. a closed parent pipe). + pass @property - def encoding(self): + def encoding(self) -> str: + """Encoding used by callers writing text to this stream.""" return "utf-8" - def close(self): + def close(self) -> None: + """Leave the original process stream open.""" return @@ -87,9 +154,15 @@ class GUIServer: bec_logger._update_sinks() bec_logger.disabled_modules = ["bec_lib.scan_items"] - with redirect_stdout(SimpleFileLikeFromLogOutputFunc(logger.info)): # type: ignore - with redirect_stderr(SimpleFileLikeFromLogOutputFunc(logger.error)): # type: ignore - self._run() + with ( + redirect_stdout( + SimpleFileLikeFromLogOutputFunc(logger.info, fallback_stream=sys.__stdout__) + ), + redirect_stderr( + SimpleFileLikeFromLogOutputFunc(logger.error, fallback_stream=sys.__stderr__) + ), + ): + self._run() def _get_service_config(self) -> ServiceConfig: if self.config: diff --git a/tests/unit_tests/test_companion_app.py b/tests/unit_tests/test_companion_app.py index cfa42752..90e6b676 100644 --- a/tests/unit_tests/test_companion_app.py +++ b/tests/unit_tests/test_companion_app.py @@ -1,8 +1,17 @@ -"""Unit tests for the GUI-server startup handshake wiring in companion_app.""" +"""Unit tests for GUI-server startup and redirected logging.""" from __future__ import annotations +import asyncio +import io +import sys +import threading +from contextlib import nullcontext from types import SimpleNamespace +from unittest.mock import Mock + +import pytest +from loguru._logger import Core, Logger from bec_widgets.applications import companion_app @@ -34,3 +43,330 @@ def test_notify_server_ready_resolves_launcher(monkeypatch): # auto-launched gui_class window) and records the final startup stage. assert calls == [("bec-gui-server", sentinel_window)] assert ("interactive", {"final": True}) in marks + + +def test_log_output_flush_preserves_partial_lines(): + messages = [] + stream = companion_app.SimpleFileLikeFromLogOutputFunc(messages.append) + + stream.write("first") + stream.flush() + assert messages == [] + + stream.write("\nsecond\npar") + assert messages == [] + stream.flush() + stream.flush() + assert messages == ["first\nsecond"] + + stream.write("tial\n") + stream.flush() + assert messages == ["first\nsecond", "partial"] + + +def test_log_output_reentrant_writes_use_original_stdout(monkeypatch): + fallback = io.StringIO() + monkeypatch.setattr(sys, "__stdout__", fallback) + messages = [] + + def log(message): + messages.append(message) + stream.write("sink diagnostic\n") + stream.flush() + + stream = companion_app.SimpleFileLikeFromLogOutputFunc(log, fallback_stream=sys.__stdout__) + # Capture the original stream when creating the adapter, before redirection. + monkeypatch.setattr(sys, "__stdout__", io.StringIO()) + stream.write("first\npartial") + stream.flush() + stream.flush() + + assert messages == ["first"] + assert fallback.getvalue() == "sink diagnostic\n" + + stream.write(" line\n") + stream.flush() + assert messages == ["first", "partial line"] + assert fallback.getvalue() == "sink diagnostic\nsink diagnostic\n" + + +def test_log_output_callback_failure_falls_back_and_recovers(): + fallback = io.StringIO() + messages = [] + + def log(message): + messages.append(message) + if len(messages) == 1: + raise RuntimeError("logger failed") + + stream = companion_app.SimpleFileLikeFromLogOutputFunc(log, fallback_stream=fallback) + stream.write("first\nsecond\npartial") + stream.flush() + stream.flush() + + assert messages == ["first\nsecond"] + assert fallback.getvalue() == "first\nsecond\n" + + stream.write(" line\n") + stream.flush() + assert messages == ["first\nsecond", "partial line"] + assert fallback.getvalue() == "first\nsecond\n" + + +@pytest.mark.parametrize("fallback_state", ["missing", "closed", "write_fails", "flush_fails"]) +def test_log_output_recovers_when_fallback_is_unavailable(monkeypatch, fallback_state): + original_stdout = io.StringIO() + monkeypatch.setattr(sys, "__stdout__", original_stdout) + fallback = None + if fallback_state == "closed": + fallback = io.StringIO() + fallback.close() + elif fallback_state in ("write_fails", "flush_fails"): + fallback = Mock(spec=io.TextIOBase) + getattr(fallback, fallback_state.removesuffix("_fails")).side_effect = BrokenPipeError( + "parent pipe closed" + ) + messages = [] + + def log(message): + messages.append(message) + if message == "first": + stream.write("logger diagnostic\n") + stream.flush() + raise RuntimeError("logger failed") + + stream = companion_app.SimpleFileLikeFromLogOutputFunc(log, fallback_stream=fallback) + stream.write("first\n") + stream.flush() + stream.write("second\n") + stream.flush() + + assert messages == ["first", "second"] + # An unavailable stderr must not accidentally use stdout as its fallback. + assert original_stdout.getvalue() == "" + if fallback_state in ("write_fails", "flush_fails"): + assert fallback.write.call_count == 2 + assert fallback.flush.call_count == (2 if fallback_state == "flush_fails" else 0) + + +def test_log_output_reentrancy_guard_is_thread_local(): + fallback = io.StringIO() + messages = [] + + def write_from_worker(): + stream.write("worker\n") + stream.flush() + + worker = threading.Thread(target=write_from_worker, daemon=True) + + def log(message): + messages.append(message) + if message == "main": + worker.start() + worker.join(timeout=2) + + stream = companion_app.SimpleFileLikeFromLogOutputFunc(log, fallback_stream=fallback) + stream.write("main\n") + stream.flush() + + assert not worker.is_alive() + assert messages == ["main", "worker"] + assert fallback.getvalue() == "" + + +def test_log_output_accepts_reentrant_writes_while_buffering(): + messages = [] + stream = companion_app.SimpleFileLikeFromLogOutputFunc(messages.append) + + class ReentrantBuffer(list): + def append(self, chunk): + # Simulate a Python signal handler writing while append holds the lock. + if chunk == "outer\n": + stream.write("inner\n") + super().append(chunk) + + stream._buffer = ReentrantBuffer() + worker = threading.Thread(target=stream.write, args=("outer\n",), daemon=True) + worker.start() + worker.join(timeout=2) + + assert not worker.is_alive() + stream.flush() + assert messages == ["inner\nouter"] + + +@pytest.fixture +def isolated_logger(): + # A separate core avoids changing or copying the application's active sinks. + logger = Logger( + core=Core(), + exception=None, + depth=0, + record=False, + lazy=False, + colors=False, + raw=False, + capture=True, + patchers=[], + extra={}, + ) + yield logger + logger.remove() + + +@pytest.mark.parametrize("run_raises", [False, True]) +@pytest.mark.parametrize("enqueue", [False, True]) +def test_server_start_logs_stderr_but_bypasses_logger_failures( + monkeypatch, isolated_logger, run_raises, enqueue +): + stdout = io.StringIO() + stderr = io.StringIO() + monkeypatch.setattr(sys, "stdout", stdout) + monkeypatch.setattr(sys, "stderr", stderr) + monkeypatch.setattr(sys, "__stdout__", stdout) + monkeypatch.setattr(sys, "__stderr__", stderr) + monkeypatch.setattr(companion_app, "logger", isolated_logger) + monkeypatch.setattr( + companion_app, + "bec_logger", + SimpleNamespace(LOGLEVEL=SimpleNamespace(INFO="INFO"), level=None, disabled_modules=[]), + ) + messages = [] + failures = [] + + def collect(message): + messages.append((message.record["level"].name, message.record["message"])) + + def broken_sink(message): + failures.append(message.record["message"]) + # Bound the old feedback loop so this regression fails without hanging. + if len(failures) <= 2: + raise OSError("broken logger sink") + + isolated_logger.add(collect, format="{message}") + isolated_logger.add( + broken_sink, level="ERROR", backtrace=False, diagnose=False, catch=True, enqueue=enqueue + ) + + def run(): + isolated_logger.error("original error") + isolated_logger.complete() + sys.stderr.flush() + print("forwarded stdout", flush=True) + print("forwarded stderr", file=sys.stderr, flush=True) + isolated_logger.complete() + sys.stderr.flush() + isolated_logger.complete() + if run_raises: + raise RuntimeError("server failed") + + server = _server() + monkeypatch.setattr(server, "_run", run) + with pytest.raises(RuntimeError, match="server failed") if run_raises else nullcontext(): + server.start() + + assert failures == ["original error", "forwarded stderr"] + assert messages == [ + ("ERROR", "original error"), + ("INFO", "forwarded stdout"), + ("ERROR", "forwarded stderr"), + ] + assert "OSError: broken logger sink" in stderr.getvalue() + assert stdout.getvalue() == "" + assert sys.stdout is stdout + assert sys.stderr is stderr + + +def test_log_output_preserves_pending_stderr_during_sink_flush(isolated_logger): + fallback = io.StringIO() + messages = [] + stream = companion_app.SimpleFileLikeFromLogOutputFunc(messages.append, fallback) + + def sink(_message): + stream.write("sink diagnostic\n") + stream.flush() + + isolated_logger.add(sink) + stream.write("application stderr\n") + isolated_logger.info("direct logger call") + + assert messages == [] + assert fallback.getvalue() == "sink diagnostic\n" + stream.flush() + assert messages == ["application stderr"] + + +@pytest.mark.parametrize("stream_name, log_level", [("stdout", "info"), ("stderr", "error")]) +def test_log_output_from_logger_patcher_bypasses_logging( + monkeypatch, isolated_logger, stream_name, log_level +): + fallback = io.StringIO() + messages = [] + patched_messages = [] + isolated_logger.add(lambda message: messages.append(message.record["message"])) + + def patcher(record): + patched_messages.append(record["message"]) + # Bound regressions that feed this diagnostic through the same patcher. + if len(patched_messages) <= 2: + print("patcher diagnostic", file=getattr(sys, stream_name), flush=True) + + patched_logger = isolated_logger.patch(patcher) + stream = companion_app.SimpleFileLikeFromLogOutputFunc( + getattr(patched_logger, log_level), fallback + ) + monkeypatch.setattr(sys, stream_name, stream) + patched_logger.info("direct logger call") + stream.flush() + + assert patched_messages == ["direct logger call"] + assert messages == ["direct logger call"] + assert fallback.getvalue() == "patcher diagnostic\n" + + +def test_log_output_bypasses_async_sink_failure(monkeypatch, isolated_logger): + fallback = io.StringIO() + messages = [] + failures = [] + isolated_logger.add(lambda message: messages.append(message.record["message"])) + stream = companion_app.SimpleFileLikeFromLogOutputFunc(isolated_logger.error, fallback) + monkeypatch.setattr(sys, "stderr", stream) + + async def broken_sink(message): + failures.append(message.record["message"]) + if len(failures) <= 2: + raise OSError("broken async sink") + + async def run(): + isolated_logger.add(broken_sink, backtrace=False, diagnose=False) + isolated_logger.error("original error") + await isolated_logger.complete() + stream.flush() + await isolated_logger.complete() + + asyncio.run(run()) + + assert failures == ["original error"] + assert messages == ["original error"] + assert "OSError: broken async sink" in fallback.getvalue() + + +def test_log_output_inside_logger_catch_still_logs_stderr(isolated_logger): + fallback = io.StringIO() + messages = [] + isolated_logger.add( + lambda message: messages.append((message.record["level"].name, str(message))) + ) + stream = companion_app.SimpleFileLikeFromLogOutputFunc(isolated_logger.error, fallback) + + @isolated_logger.catch + def application_code(): + stream.write("ordinary stderr\n") + stream.flush() + + application_code() + + assert len(messages) == 1 + assert messages[0][0] == "ERROR" + assert "ordinary stderr" in messages[0][1] + assert fallback.getvalue() == ""