From c8d3d0f52965a8e95b244f2694b1f9b1c7bb6a90 Mon Sep 17 00:00:00 2001 From: Romulo Quidute Filho Date: Tue, 29 Sep 2026 10:06:12 -0300 Subject: [PATCH] Fix stream_sink re-entrancy, log yield interval, and trailing log drop logger.add(stream_sink, ...) was missing the enqueue=True its own comment claimed, so it ran synchronously on whichever thread logged the message - including the SSE-server thread logging its own send progress in logs_http_server.py, re-entering the same sink/queue from inside itself. Adding enqueue=True moves it to loguru's dedicated worker thread, closing that path. Since stream_sink now runs asynchronously, stop_log_streaming() calls logger.complete() before stopping the handler, so queued log lines aren't dropped by add_log_entry() turning into a no-op (or the client set being torn down) while they're still in flight to the worker. Lowered LOG_RECORD_YIELD_INTERVAL from 200 to 50. The backend sends up to LOG_RECORDS_BROADCAST_CHUNK_SIZE=200 records per websocket message (test_ui_observer.py), so the old value only yielded once per message; the lower value makes a single large message yield multiple times instead of running uninterrupted end to end. Part of https://github.com/project-chip/certification-tool/issues/1119 --- th_cli/test_run/logging.py | 15 +++++++++++++-- th_cli/test_run/websocket.py | 7 +++++-- 2 files changed, 18 insertions(+), 4 deletions(-) diff --git a/th_cli/test_run/logging.py b/th_cli/test_run/logging.py index becc7f5..8d160b2 100644 --- a/th_cli/test_run/logging.py +++ b/th_cli/test_run/logging.py @@ -75,8 +75,13 @@ def stream_sink(message): # Silently fail to avoid disrupting logging pass - # Add sink with enqueue=True to prevent re-entrancy and catch=True to suppress errors - logger.add(stream_sink, format="{message}", catch=True) + # enqueue=True: run stream_sink on loguru's dedicated worker + # thread rather than synchronously on the caller. Without it, + # this sink runs inline on whichever thread logged the message - + # including the SSE-server thread when it logs its own send + # progress (logs_http_server.py), which then re-enters this same + # sink/queue from inside itself. + logger.add(stream_sink, format="{message}", enqueue=True, catch=True) logger.info(f"Real-time log streaming enabled: {viewer_url}") except Exception as e: @@ -92,6 +97,12 @@ def stop_log_streaming(): if _log_stream_handler: try: + # stream_sink runs via enqueue=True, on loguru's worker thread. + # complete() blocks until everything already queued has been + # delivered, so the last log lines aren't dropped by stopping + # the handler (which makes add_log_entry() a no-op) while some + # are still in flight. + logger.complete() _log_stream_handler.stop() logger.info("Log streaming stopped") except Exception as e: diff --git a/th_cli/test_run/websocket.py b/th_cli/test_run/websocket.py index 9ab33c7..8207693 100644 --- a/th_cli/test_run/websocket.py +++ b/th_cli/test_run/websocket.py @@ -95,8 +95,11 @@ # Yield to the event loop every N log records while processing one batch, so # a very large batch doesn't block the websocket read loop for its entire -# duration. -LOG_RECORD_YIELD_INTERVAL = 200 +# duration. The backend sends up to LOG_RECORDS_BROADCAST_CHUNK_SIZE=200 +# records per message (test_ui_observer.py), so the old value of 200 only +# yielded once per message; lowering it means a single large message yields +# multiple times instead of running uninterrupted end to end. +LOG_RECORD_YIELD_INTERVAL = 50 # TestRun states that are not yet finished, mirroring the backend's own # TestRun.completed() contract (state not in [PENDING, EXECUTING]). Anything