diff --git a/pyproject.toml b/pyproject.toml index 4249b9e..e625f61 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "pyqt-reactive" -version = "0.1.25" +version = "0.1.26" description = "React-quality reactive form generation framework for PyQt6" authors = [{name = "Tristan Simas", email = "tristan.simas@mail.mcgill.ca"}] license = {text = "MIT"} diff --git a/src/pyqt_reactive/__init__.py b/src/pyqt_reactive/__init__.py index 56778ad..862315c 100644 --- a/src/pyqt_reactive/__init__.py +++ b/src/pyqt_reactive/__init__.py @@ -19,7 +19,7 @@ - Cross-window reactive updates """ -__version__ = "0.1.25" +__version__ = "0.1.26" # Public API will be populated as modules are added __all__ = [ diff --git a/src/pyqt_reactive/utils/log_highlighter.py b/src/pyqt_reactive/utils/log_highlighter.py index c7f9cb2..2117467 100644 --- a/src/pyqt_reactive/utils/log_highlighter.py +++ b/src/pyqt_reactive/utils/log_highlighter.py @@ -45,6 +45,41 @@ def parse_line(text: str) -> list[dict]: return segments +def _escape_html(text: str) -> str: + return text.replace("&", "&").replace("<", "<").replace(">", ">") + + +def build_log_line_html(text: str) -> str: + """Render one line from the authoritative parsed highlight segments.""" + segments = sorted(parse_line(text), key=lambda segment: segment["start"]) + cursor = 0 + parts: list[str] = [''] + text_len = len(text) + + for segment in segments: + start = max(0, segment["start"]) + end = min(text_len, start + segment["length"]) + if end <= start or start < cursor: + continue + if start > cursor: + parts.append(_escape_html(text[cursor:start])) + + red, green, blue = segment["color"] + style = f"color: rgb({red},{green},{blue});" + if segment.get("bold"): + style += " font-weight: 700;" + parts.append( + f'{_escape_html(text[start:end])}' + ) + cursor = end + + if cursor < text_len: + parts.append(_escape_html(text[cursor:])) + + parts.append("") + return "".join(parts) + + def main() -> int: for line in sys.stdin: try: diff --git a/src/pyqt_reactive/utils/log_streamer.py b/src/pyqt_reactive/utils/log_streamer.py index ca7cb6a..91896c5 100644 --- a/src/pyqt_reactive/utils/log_streamer.py +++ b/src/pyqt_reactive/utils/log_streamer.py @@ -5,8 +5,8 @@ import sys from collections import deque from pathlib import Path -import re +from pyqt_reactive.utils.log_highlighter import build_log_line_html def emit(obj: dict) -> None: sys.stdout.write(json.dumps(obj, ensure_ascii=True) + "\n") @@ -24,82 +24,6 @@ def tail_lines(path: Path, max_lines: int) -> list[str]: return list(buf) -_TIMESTAMP_RE = re.compile(r"^\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}:\d{2},\d{3}") -_LOG_LEVEL_RE = re.compile(r"\b(ERROR|WARNING|INFO|DEBUG|CRITICAL)\b") -_LOGGER_NAME_RE = re.compile(r" - ([\w\.]+) - ") -_FILE_PATH_RE = re.compile(r"(?:/[\w\-\.]+)+\.py") -_PYTHON_STRING_RE = re.compile(r"[\"'](?:[^\"'\\]|\\.)*[\"']") -_NUMBER_RE = re.compile(r"\b\d+(?:\.\d+)?\b") - - -def _escape_html(text: str) -> str: - return ( - text.replace("&", "&") - .replace("<", "<") - .replace(">", ">") - ) - - -def _segments_for_line(text: str) -> list[dict]: - segments = [] - - match = _TIMESTAMP_RE.match(text) - if match: - segments.append({"start": match.start(), "length": match.end() - match.start(), "color": (105, 105, 105)}) - - for match in _LOG_LEVEL_RE.finditer(text): - level = match.group(1) - if level in ("ERROR", "CRITICAL"): - color = (255, 85, 85) - elif level == "WARNING": - color = (255, 140, 0) - else: - color = (100, 160, 210) - segments.append({"start": match.start(), "length": match.end() - match.start(), "color": color, "bold": True}) - - for match in _LOGGER_NAME_RE.finditer(text): - segments.append({"start": match.start(1), "length": match.end(1) - match.start(1), "color": (147, 112, 219)}) - - for match in _FILE_PATH_RE.finditer(text): - segments.append({"start": match.start(), "length": match.end() - match.start(), "color": (34, 139, 34)}) - - for match in _PYTHON_STRING_RE.finditer(text): - segments.append({"start": match.start(), "length": match.end() - match.start(), "color": (206, 145, 120)}) - - for match in _NUMBER_RE.finditer(text): - segments.append({"start": match.start(), "length": match.end() - match.start(), "color": (181, 206, 168)}) - - return segments - - -def _build_html(text: str) -> str: - segments = sorted(_segments_for_line(text), key=lambda s: s["start"]) - cursor = 0 - parts: list[str] = [""] - text_len = len(text) - - for seg in segments: - start = max(0, seg["start"]) - end = min(text_len, start + seg["length"]) - if end <= start or start < cursor: - continue - if start > cursor: - parts.append(_escape_html(text[cursor:start])) - - r, g, b = seg["color"] - style = f"color: rgb({r},{g},{b});" - if seg.get("bold"): - style += " font-weight: 700;" - parts.append(f"{_escape_html(text[start:end])}") - cursor = end - - if cursor < text_len: - parts.append(_escape_html(text[cursor:])) - - parts.append("") - return "".join(parts) - - def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("path") @@ -114,7 +38,7 @@ def main() -> int: chunk: list = [] for line in lines: if args.html: - chunk.append({"text": line, "html": _build_html(line)}) + chunk.append({"text": line, "html": build_log_line_html(line)}) else: chunk.append(line) if len(chunk) >= args.chunk_lines: diff --git a/src/pyqt_reactive/widgets/log_viewer.py b/src/pyqt_reactive/widgets/log_viewer.py index b9b1325..eba7206 100644 --- a/src/pyqt_reactive/widgets/log_viewer.py +++ b/src/pyqt_reactive/widgets/log_viewer.py @@ -7,9 +7,8 @@ """ import logging -import sys -import json import re +from collections import deque from typing import Optional, List, Set, Tuple from pathlib import Path @@ -23,7 +22,7 @@ from PyQt6.QtCore import ( QObject, QTimer, QFileSystemWatcher, pyqtSignal, pyqtSlot, Qt, QRegularExpression, QThread, QAbstractListModel, QModelIndex, QSize, - QThreadPool, QRunnable, QPoint, QPointF, QProcess, + QThreadPool, QRunnable, QPoint, QPointF, ) from PyQt6.QtGui import QTextCharFormat, QColor, QAction, QFont, QTextCursor, QPalette, QAbstractTextDocumentLayout, QKeySequence, QFontMetrics @@ -44,6 +43,7 @@ from typing import Dict, Tuple from pyqt_reactive.utils.log_highlight_client import LogHighlightClient, HighlightedSegmentDTO +from pyqt_reactive.utils.log_highlighter import build_log_line_html logger = logging.getLogger(__name__) @@ -1831,99 +1831,56 @@ def highlightBlock(self, text: str) -> None: class LogFileLoader(QThread): - """Background process for loading large log files without blocking UI.""" + """Interruptible worker thread for loading large log files.""" # Signals - chunk_loaded = pyqtSignal(list) # Emits list[str] chunks - load_finished = pyqtSignal() # Emits when all chunks loaded - load_failed = pyqtSignal(str) # Emits error message on failure + chunk_loaded = pyqtSignal(object, list) # Emits (loader, list[str]) + load_finished = pyqtSignal(object) # Emits loader when all chunks loaded + load_failed = pyqtSignal(object, str) # Emits (loader, error message) - def __init__(self, log_path: Path, tail_lines: int = 100_000, chunk_lines: int = 1000): - super().__init__() + def __init__( + self, + log_path: Path, + tail_lines: int = 100_000, + chunk_lines: int = 1000, + parent: Optional[QObject] = None, + ): + super().__init__(parent) self.log_path = log_path self.tail_lines = tail_lines self.chunk_lines = chunk_lines - self._process: Optional[QProcess] = None - self._buffer = "" - self._stderr_chunks: list[str] = [] - def start(self): # type: ignore[override] - """Spawn a worker process to load the file content.""" + def run(self) -> None: + """Read and highlight the bounded tail without blocking the GUI thread.""" try: - if self._process and self._process.state() != QProcess.ProcessState.NotRunning: - self._process.kill() - self._process = None - - self._buffer = "" - self._stderr_chunks = [] - - self._process = QProcess() - self._process.setProgram(sys.executable) - args = [ - "-m", - "pyqt_reactive.utils.log_streamer", - str(self.log_path), - "--tail-lines", - str(self.tail_lines), - "--chunk-lines", - str(self.chunk_lines), - "--html", - ] - logger.debug("Spawning log_streamer subprocess: %s %s", sys.executable, args) - self._process.setArguments(args) - self._process.readyReadStandardOutput.connect(self._on_stdout) - self._process.readyReadStandardError.connect(self._on_stderr) - self._process.finished.connect(self._on_finished) - self._process.start() - except Exception as e: - self.load_failed.emit(str(e)) - - def isRunning(self) -> bool: - return bool(self._process and self._process.state() != QProcess.ProcessState.NotRunning) - - def wait(self, msecs: int = 0) -> None: - if self._process: - self._process.waitForFinished(msecs) - - def _on_stdout(self) -> None: - if not self._process: - return - data = bytes(self._process.readAllStandardOutput()) - text = data.decode("utf-8", errors="replace") - if not text: - return - - self._buffer += text - while "\n" in self._buffer: - line, self._buffer = self._buffer.split("\n", 1) - if not line.strip(): - continue - try: - payload = json.loads(line) - except json.JSONDecodeError: - continue + if self.tail_lines <= 0: + self.load_finished.emit(self) + return - if payload.get("type") == "chunk": - lines = payload.get("lines", []) - if lines: - self.chunk_loaded.emit(lines) - elif payload.get("type") == "done": - self.load_finished.emit() + lines: deque[str] = deque(maxlen=self.tail_lines) + with open(self.log_path, "r", encoding="utf-8", errors="replace") as handle: + for line in handle: + if self.isInterruptionRequested(): + return + lines.append(line.rstrip("\n")) - def _on_stderr(self) -> None: - if not self._process: - return - data = bytes(self._process.readAllStandardError()) - text = data.decode("utf-8", errors="replace") - if text: - self._stderr_chunks.append(text) + chunk: list[dict[str, str]] = [] + for line in lines: + if self.isInterruptionRequested(): + return + chunk.append({"text": line, "html": build_log_line_html(line)}) + if len(chunk) >= self.chunk_lines: + self.chunk_loaded.emit(self, chunk) + chunk = [] - def _on_finished(self) -> None: - stderr = "".join(self._stderr_chunks).strip() + if chunk and not self.isInterruptionRequested(): + self.chunk_loaded.emit(self, chunk) - if stderr: - self.load_failed.emit(stderr) - return + if not self.isInterruptionRequested(): + self.load_finished.emit(self) + except Exception as exc: + if not self.isInterruptionRequested(): + self.load_failed.emit(self, str(exc)) class LogTailer(QThread): @@ -2018,7 +1975,7 @@ def __init__(self, file_manager, service_adapter, parent=None): # Update throttling to reduce UI load self._pending_lines: List[str] = [] # Buffer for pending log lines - self._update_timer = QTimer() + self._update_timer = QTimer(self) self._update_timer.setSingleShot(True) self._update_timer.timeout.connect(self._flush_pending_lines) self._update_throttle_ms = 50 # Minimum time between UI updates @@ -2035,8 +1992,10 @@ def __init__(self, file_manager, service_adapter, parent=None): self.log_tailer: Optional[LogTailer] = None # Async log tailer thread self.highlighter: Optional[LogHighlighter] = None self.file_loader: Optional[LogFileLoader] = None # Async file loader + self._file_loaders: set[LogFileLoader] = set() self.server_scan_timer: QTimer = None # Periodic ZMQ server scanning self._pending_log_to_load: Optional[Path] = None # Log to load when window is shown + self._cleaned_up = False # Process tracking for alive/dead process indication self.process_tracker = ProcessTracker() @@ -2476,9 +2435,10 @@ def switch_to_log(self, log_path: Path) -> None: # Stop current tailing self.stop_log_tailing() - # Stop any existing file loader - if self.file_loader and self.file_loader.isRunning(): - self.file_loader.wait() + # Superseded workers are interrupted without blocking the GUI. + # Their identity remains owned until QThread.finished so queued + # signals cannot outlive the worker or mutate the new selection. + self._retire_file_loader(self.file_loader) # Reset streaming state self._pending_partial_line = "" @@ -2501,18 +2461,44 @@ def switch_to_log(self, log_path: Path) -> None: self.log_model.append_lines([f"Loading log file ({file_size // 1024} KB)..."]) # Create and start async loader - self.file_loader = LogFileLoader(log_path) + self.file_loader = LogFileLoader(log_path, parent=self) + self._file_loaders.add(self.file_loader) self.file_loader.chunk_loaded.connect(self._on_file_chunk_loaded) self.file_loader.load_finished.connect(self._on_file_load_finished) self.file_loader.load_failed.connect(self._on_file_load_failed) + self.file_loader.finished.connect( + lambda loader=self.file_loader: self._release_retired_file_loader( + loader + ) + ) self.file_loader.start() except Exception as e: logger.error(f"Error switching to log {log_path}: {e}") raise - def _on_file_chunk_loaded(self, lines: list) -> None: + def _retire_file_loader(self, loader: Optional[LogFileLoader]) -> None: + """Interrupt a superseded loader while retaining its QObject lifetime.""" + if loader is None: + return + if loader is self.file_loader: + self.file_loader = None + if loader.isRunning(): + loader.requestInterruption() + return + self._release_retired_file_loader(loader) + + def _release_retired_file_loader(self, loader: LogFileLoader) -> None: + """Release a finished loader once it is no longer the active selection.""" + if loader is self.file_loader: + return + self._file_loaders.discard(loader) + loader.deleteLater() + + def _on_file_chunk_loaded(self, loader: LogFileLoader, lines: list) -> None: """Handle a chunk of lines loaded from subprocess.""" + if loader is not self.file_loader: + return try: if not lines: return @@ -2522,8 +2508,10 @@ def _on_file_chunk_loaded(self, lines: list) -> None: except Exception as e: logger.error(f"Error displaying loaded chunk: {e}") - def _on_file_load_finished(self) -> None: + def _on_file_load_finished(self, loader: LogFileLoader) -> None: """Finalize after all chunks are loaded.""" + if loader is not self.file_loader: + return try: # Update file position immediately if self.current_log_path and self.current_log_path.exists(): @@ -2577,8 +2565,14 @@ def _finish_batch_insert(self) -> None: logger.info(f"Loaded log file: {self.current_log_path}") - def _on_file_load_failed(self, error_msg: str) -> None: + def _on_file_load_failed( + self, + loader: LogFileLoader, + error_msg: str, + ) -> None: """Handle file load failure.""" + if loader is not self.file_loader: + return self.clear_log_display() self.log_model.append_lines([f"Failed to load log file: {error_msg}"]) logger.error(f"Failed to load log file: {error_msg}") @@ -2765,9 +2759,10 @@ def start_log_tailing(self, log_path: Path) -> None: def stop_log_tailing(self) -> None: """Stop current log tailing.""" if self.log_tailer: - self.log_tailer.stop() - self.log_tailer.wait(1000) # Wait up to 1 second for thread to finish + log_tailer = self.log_tailer self.log_tailer = None + log_tailer.stop() + log_tailer.wait() logger.debug("Stopped log tailing") def _on_new_content(self, new_content: str, new_file_position: int) -> None: @@ -2927,7 +2922,7 @@ def start_process_tracking(self) -> None: self._update_tracked_processes() # Process status badges are informational; avoid rescanning constantly. - self.process_update_timer = QTimer() + self.process_update_timer = QTimer(self) self.process_update_timer.timeout.connect(self.update_process_status) self.process_update_timer.start(10000) @@ -3049,47 +3044,37 @@ def on_filter_changed(self, state: int) -> None: logger.debug(f"Filter changed: show_alive_only={self.show_alive_only}") def cleanup(self) -> None: - """Cleanup all resources and background processes.""" - try: - # Stop async log tailer thread - if hasattr(self, 'log_tailer') and self.log_tailer: - self.stop_log_tailing() - - # Stop tailing timer (deprecated, kept for compatibility) - if hasattr(self, 'tail_timer') and self.tail_timer and self.tail_timer.isActive(): - self.tail_timer.stop() - self.tail_timer.deleteLater() - self.tail_timer = None - - # Stop process tracking timer - if hasattr(self, 'process_update_timer') and self.process_update_timer and self.process_update_timer.isActive(): - self.process_update_timer.stop() - self.process_update_timer.deleteLater() - self.process_update_timer = None - - # Stop file monitoring - self.stop_monitoring() - - # Clean up file detector - if hasattr(self, 'file_detector') and self.file_detector: - self.file_detector.stop_watching() - self.file_detector = None - - except Exception as e: - logger.warning(f"Error during log viewer cleanup: {e}") + """Deterministically stop every worker and timer owned by the viewer.""" + if self._cleaned_up: + return + self._cleaned_up = True - def closeEvent(self, event) -> None: - """Handle window close event.""" - # Stop async tailer - if hasattr(self, 'log_tailer') and self.log_tailer: - self.stop_log_tailing() + self._update_timer.stop() + self.stop_log_tailing() - if self.file_detector: - self.file_detector.stop_watching() - if self.tail_timer: + for loader in tuple(self._file_loaders): + if loader.isRunning(): + loader.requestInterruption() + for loader in tuple(self._file_loaders): + if loader.isRunning(): + loader.wait() + self.file_loader = None + self._file_loaders.clear() + + # Stop tailing timer (deprecated, kept for compatibility) + if self.tail_timer and self.tail_timer.isActive(): self.tail_timer.stop() - if hasattr(self, 'process_update_timer') and self.process_update_timer: + self.tail_timer = None + + if self.process_update_timer and self.process_update_timer.isActive(): self.process_update_timer.stop() + self.process_update_timer = None + + self.stop_monitoring() LogHighlightClient.shutdown() + + def closeEvent(self, event) -> None: + """Handle window close event.""" + self.cleanup() self.window_closed.emit() super().closeEvent(event) diff --git a/tests/test_log_viewer_lifecycle.py b/tests/test_log_viewer_lifecycle.py new file mode 100644 index 0000000..6342172 --- /dev/null +++ b/tests/test_log_viewer_lifecycle.py @@ -0,0 +1,147 @@ +"""Lifecycle regressions for asynchronous log loading.""" + +from __future__ import annotations + +from pathlib import Path + +from PyQt6.QtCore import qInstallMessageHandler + +from pyqt_reactive.protocols import register_log_discovery_provider +from pyqt_reactive.utils.log_highlighter import build_log_line_html +from pyqt_reactive.widgets.log_viewer import ( + LogFileInfo, + LogFileLoader, + LogViewerWindow, +) + + +def test_log_line_html_uses_authoritative_highlight_segments() -> None: + rendered = build_log_line_html( + "2026-07-29 20:00:00,000 - worker - ERROR - value < 2" + ) + + assert "ERROR" in rendered + assert "font-weight: 700" in rendered + assert "<" in rendered + assert ">2" in rendered + + +class _LogDiscoveryProvider: + def __init__(self, main_log_path: Path) -> None: + self.main_log_path = main_log_path + + def get_current_log_path(self) -> Path: + return self.main_log_path + + def discover_logs( + self, + base_log_path=None, + include_main_log: bool = True, + log_directory=None, + ) -> list[LogFileInfo]: + if not include_main_log: + return [] + return [LogFileInfo(path=self.main_log_path, log_type="main")] + + +def _disable_external_discovery(monkeypatch) -> None: + monkeypatch.setattr( + LogViewerWindow, + "_scan_subprocess_logs_async", + lambda self: None, + ) + monkeypatch.setattr(LogViewerWindow, "_scan_servers_async", lambda self: None) + monkeypatch.setattr(LogViewerWindow, "start_monitoring", lambda self: None) + monkeypatch.setattr(LogViewerWindow, "start_process_tracking", lambda self: None) + + +def test_log_file_loader_runs_as_interruptible_qthread(qtbot, tmp_path) -> None: + """The loader must use its QThread lifecycle rather than an owned process.""" + + log_path = tmp_path / "main.log" + log_path.write_text("first\nsecond\nthird\n", encoding="utf-8") + loader = LogFileLoader(log_path, tail_lines=2, chunk_lines=1) + started = [] + chunks = [] + completed = [] + loader.started.connect(lambda: started.append(True)) + loader.chunk_loaded.connect(lambda owner, lines: chunks.append((owner, lines))) + loader.load_finished.connect(completed.append) + + loader.start() + qtbot.waitUntil(lambda: completed == [loader]) + loader.wait() + + assert started == [True] + assert [chunk[1][0]["text"] for chunk in chunks] == ["second", "third"] + assert all(owner is loader for owner, _chunk in chunks) + + +def test_rapid_log_switching_retires_workers_without_qt_lifecycle_failure( + qtbot, + tmp_path, + monkeypatch, +) -> None: + """Superseded loads cannot destroy a running worker or update the new view.""" + + main_log_path = tmp_path / "main.log" + second_log_path = tmp_path / "second.log" + main_log_path.write_text( + "".join(f"main line {index}\n" for index in range(1_000)), + encoding="utf-8", + ) + second_log_path.write_text("second\n", encoding="utf-8") + register_log_discovery_provider(_LogDiscoveryProvider(main_log_path)) + _disable_external_discovery(monkeypatch) + + qt_messages = [] + previous_message_handler = qInstallMessageHandler( + lambda _kind, _context, message: qt_messages.append(message) + ) + viewer = LogViewerWindow(file_manager=None, service_adapter=None) + try: + for _index in range(10): + viewer.switch_to_log(second_log_path) + viewer.switch_to_log(main_log_path) + + qtbot.waitUntil( + lambda: ( + viewer.file_loader is not None + and not viewer.file_loader.isRunning() + and viewer.log_model.rowCount() == 1_001 + ), + timeout=5_000, + ) + + assert viewer.current_log_path == main_log_path + assert viewer.log_model.data(viewer.log_model.index(1_000, 0)) == "main line 999" + assert not [ + message + for message in qt_messages + if "Destroyed while process" in message + or "Destroyed while thread" in message + ] + finally: + viewer.cleanup() + qInstallMessageHandler(previous_message_handler) + + assert viewer.file_loader is None + assert viewer._file_loaders == set() + + +def test_cleanup_joins_an_active_file_loader(qapp, tmp_path, monkeypatch) -> None: + """Closing during initial loading leaves no live worker owned by the viewer.""" + + main_log_path = tmp_path / "main.log" + main_log_path.write_text( + "".join(f"line {index}\n" for index in range(10_000)), + encoding="utf-8", + ) + register_log_discovery_provider(_LogDiscoveryProvider(main_log_path)) + _disable_external_discovery(monkeypatch) + viewer = LogViewerWindow(file_manager=None, service_adapter=None) + + viewer.cleanup() + + assert viewer.file_loader is None + assert viewer._file_loaders == set()