diff --git a/graphify/querylog.py b/graphify/querylog.py index b89f419dd6..8db6ee7c5c 100644 --- a/graphify/querylog.py +++ b/graphify/querylog.py @@ -1,13 +1,15 @@ """Query logging for graphify — append-only JSONL, fail-silent.""" from __future__ import annotations +import contextlib import json import os import re -import time from datetime import datetime, timezone from pathlib import Path -from typing import Any +from typing import Any, Iterator + +from graphify.paths import _atomic_replace _NODES_RE = re.compile(r"(\d+)\s+nodes?\s+found") @@ -40,6 +42,74 @@ def nodes_from_result(result: str) -> int | None: return int(m.group(1)) if m else None +def _max_records() -> int | None: + raw = os.environ.get("GRAPHIFY_QUERY_LOG_MAX_RECORDS", "").strip() + if not raw: + return None + try: + n = int(raw) + except ValueError: + return None + return n if n > 0 else None + + +def _archive_path(path: Path) -> Path: + return path.with_name(f"{path.stem}.archive{path.suffix}") + + +@contextlib.contextmanager +def _query_log_lock(path: Path) -> Iterator[None]: + """Serialize append+rotate on POSIX (watch.py uses the same flock pattern). + + Multi-process rotation on Windows is best-effort when fcntl is unavailable. + If the lock file cannot be opened, degrades to unlocked append rather than + dropping the log line. + """ + try: + import fcntl + except ImportError: + yield + return + lock_path = path.with_name(path.name + ".lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + try: + fh = open(lock_path, "a+", encoding="utf-8") + except OSError: + yield + return + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_EX) + yield + finally: + try: + fcntl.flock(fh.fileno(), fcntl.LOCK_UN) + except OSError: + pass + fh.close() + + +def _rotate_if_needed(path: Path, max_records: int) -> None: + if not path.is_file(): + return + lines = path.read_text(encoding="utf-8").splitlines(keepends=True) + if not lines or len(lines) <= max_records: + return + overflow, keep = lines[:-max_records], lines[-max_records:] + + def _write_keep(fh) -> None: + fh.write("".join(keep)) + + _atomic_replace(path, _write_keep) + archive = _archive_path(path) + archive.parent.mkdir(parents=True, exist_ok=True) + try: + with archive.open("a", encoding="utf-8") as fh: + fh.writelines(overflow) + except OSError: + with path.open("a", encoding="utf-8") as fh: + fh.writelines(overflow) + + def log_query( *, kind: str, @@ -74,7 +144,11 @@ def log_query( if result is not None and _log_responses(): rec["response"] = result path.parent.mkdir(parents=True, exist_ok=True) - with path.open("a", encoding="utf-8") as fh: - fh.write(json.dumps(rec, ensure_ascii=False) + "\n") + with _query_log_lock(path): + with path.open("a", encoding="utf-8") as fh: + fh.write(json.dumps(rec, ensure_ascii=False) + "\n") + max_records = _max_records() + if max_records is not None: + _rotate_if_needed(path, max_records) except Exception: pass diff --git a/tests/test_querylog.py b/tests/test_querylog.py index b843550c75..8776124cf6 100644 --- a/tests/test_querylog.py +++ b/tests/test_querylog.py @@ -1,10 +1,11 @@ """Tests for graphify.querylog.""" import json import os +import threading import pytest from pathlib import Path -from graphify.querylog import log_query, nodes_from_result +from graphify.querylog import _archive_path, log_query, nodes_from_result # --------------------------------------------------------------------------- @@ -221,3 +222,147 @@ def test_log_query_writes_nothing_by_default(monkeypatch, tmp_path): monkeypatch.setattr(Path, "home", classmethod(lambda cls: tmp_path)) log_query(kind="query", question="secret internal ticket TICKET-123", corpus=".", result="1 node found") assert not (tmp_path / ".cache" / "graphify-queries.log").exists() + + +# --------------------------------------------------------------------------- +# #3051 — optional rotation via GRAPHIFY_QUERY_LOG_MAX_RECORDS +# --------------------------------------------------------------------------- + +def _rotation_env(monkeypatch, tmp_path, max_records=None): + log_file = tmp_path / "q.log" + monkeypatch.setenv("GRAPHIFY_QUERY_LOG", str(log_file)) + monkeypatch.delenv("GRAPHIFY_QUERY_LOG_DISABLE", raising=False) + if max_records is None: + monkeypatch.delenv("GRAPHIFY_QUERY_LOG_MAX_RECORDS", raising=False) + else: + monkeypatch.setenv("GRAPHIFY_QUERY_LOG_MAX_RECORDS", str(max_records)) + return log_file + + +def test_rotation_unset_keeps_all_lines(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path) + for i in range(5): + log_query(kind="query", question=f"q{i}", corpus="/g.json") + lines = log_file.read_text().splitlines() + assert len(lines) == 5 + assert not _archive_path(log_file).exists() + + +def test_rotation_trims_live_keeps_newest(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=3) + for i in range(5): + log_query(kind="query", question=f"q{i}", corpus="/g.json") + live = [json.loads(l)["question"] for l in log_file.read_text().splitlines()] + archive = [json.loads(l)["question"] for l in _archive_path(log_file).read_text().splitlines()] + assert live == ["q2", "q3", "q4"] + assert archive == ["q0", "q1"] + + +def test_rotation_invalid_env_is_noop(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records="abc") + for i in range(4): + log_query(kind="query", question=f"q{i}", corpus="/g.json") + assert len(log_file.read_text().splitlines()) == 4 + assert not _archive_path(log_file).exists() + + monkeypatch.setenv("GRAPHIFY_QUERY_LOG_MAX_RECORDS", "0") + log_query(kind="query", question="extra", corpus="/g.json") + assert len(log_file.read_text().splitlines()) == 5 + + +def test_rotation_archive_appends_across_rotations(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=2) + for i in range(5): + log_query(kind="query", question=f"q{i}", corpus="/g.json") + archive_path = _archive_path(log_file) + archived = [json.loads(l)["question"] for l in archive_path.read_text().splitlines()] + assert archived == ["q0", "q1", "q2"] + live = [json.loads(l)["question"] for l in log_file.read_text().splitlines()] + assert live == ["q3", "q4"] + + +def test_rotation_never_raises(tmp_path, monkeypatch): + bad_path = tmp_path / "is_a_dir" + bad_path.mkdir() + monkeypatch.setenv("GRAPHIFY_QUERY_LOG", str(bad_path)) + monkeypatch.setenv("GRAPHIFY_QUERY_LOG_MAX_RECORDS", "1") + monkeypatch.delenv("GRAPHIFY_QUERY_LOG_DISABLE", raising=False) + log_query(kind="query", question="q", corpus="/g.json") + + +def test_rotation_archive_after_live_replace(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=2) + log_query(kind="query", question="q0", corpus="/g.json") + log_query(kind="query", question="q1", corpus="/g.json") + + def fail_replace(src, dst): + raise OSError("simulated replace failure") + + monkeypatch.setattr(os, "replace", fail_replace) + log_query(kind="query", question="q2", corpus="/g.json") + + archive = _archive_path(log_file) + assert not archive.exists() or archive.read_text() == "" + live = [json.loads(l)["question"] for l in log_file.read_text().splitlines()] + assert live == ["q0", "q1", "q2"] + + +def test_rotation_concurrent_appends_preserved(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=5) + errors: list[Exception] = [] + + def worker(prefix: str) -> None: + try: + for i in range(5): + log_query(kind="query", question=f"{prefix}{i}", corpus="/g.json") + except Exception as exc: + errors.append(exc) + + threads = [threading.Thread(target=worker, args=(p,)) for p in ("a", "b")] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors + + questions: set[str] = set() + for path in (log_file, _archive_path(log_file)): + if path.exists(): + for line in path.read_text().splitlines(): + questions.add(json.loads(line)["question"]) + assert len(questions) == 10 + + +def test_rotation_lock_open_failure_still_appends(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=3) + real_open = open + + def selective_open(file, *args, **kwargs): + if str(file).endswith(".lock"): + raise OSError("simulated lock open failure") + return real_open(file, *args, **kwargs) + + monkeypatch.setattr("builtins.open", selective_open) + log_query(kind="query", question="q0", corpus="/g.json") + assert log_file.exists() + rec = json.loads(log_file.read_text()) + assert rec["question"] == "q0" + + +def test_rotation_archive_failure_restores_overflow(tmp_path, monkeypatch): + log_file = _rotation_env(monkeypatch, tmp_path, max_records=2) + archive = _archive_path(log_file) + real_open = Path.open + + def selective_open(self, *args, **kwargs): + if self == archive and args and args[0] == "a": + raise OSError("simulated archive failure") + return real_open(self, *args, **kwargs) + + monkeypatch.setattr(Path, "open", selective_open) + for i in range(3): + log_query(kind="query", question=f"q{i}", corpus="/g.json") + + live = [json.loads(l)["question"] for l in log_file.read_text().splitlines()] + assert live == ["q1", "q2", "q0"] + assert not archive.exists() or archive.read_text() == ""