Fix #3051: optional query log rotation via GRAPHIFY_QUERY_LOG_MAX_RECORDS

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

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() == ""