From c91b14c20a3b680097b58602e5803824094e8ac7 Mon Sep 17 00:00:00 2001 From: akshit Date: Mon, 31 Aug 2026 21:19:05 +0530 Subject: [PATCH 1/3] Add optional query log rotation (#3051) Bound live JSONL size when GRAPHIFY_QUERY_LOG_MAX_RECORDS is set. Overflow records append to a sibling archive file. Default off. --- graphify/querylog.py | 36 ++++++++++++++++++++++ tests/test_querylog.py | 68 +++++++++++++++++++++++++++++++++++++++++- 2 files changed, 103 insertions(+), 1 deletion(-) diff --git a/graphify/querylog.py b/graphify/querylog.py index b89f419dd6..262442f468 100644 --- a/graphify/querylog.py +++ b/graphify/querylog.py @@ -40,6 +40,39 @@ 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}") + + +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: + return + if len(lines) <= max_records: + return + overflow, keep = lines[:-max_records], lines[-max_records:] + archive = _archive_path(path) + archive.parent.mkdir(parents=True, exist_ok=True) + with archive.open("a", encoding="utf-8") as fh: + fh.writelines(overflow) + tmp = path.with_name(path.name + ".tmp") + tmp.write_text("".join(keep), encoding="utf-8") + os.replace(tmp, path) + + def log_query( *, kind: str, @@ -76,5 +109,8 @@ def log_query( 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") + 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..bccf55fb89 100644 --- a/tests/test_querylog.py +++ b/tests/test_querylog.py @@ -4,7 +4,7 @@ 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 +221,69 @@ 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") From b46c46f575ce97ebfbe9619956029821b08f9a2b Mon Sep 17 00:00:00 2001 From: akshit Date: Tue, 1 Sep 2026 01:57:17 +0530 Subject: [PATCH 2/3] Address bot review: flock query log rotation Serialize append and rotate under POSIX flock on the log lock file. Write the live log before appending overflow to the archive so a failed replace cannot duplicate archived records. --- graphify/querylog.py | 48 ++++++++++++++++++++++++++++++------------ tests/test_querylog.py | 44 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 13 deletions(-) diff --git a/graphify/querylog.py b/graphify/querylog.py index 262442f468..786bf47fa6 100644 --- a/graphify/querylog.py +++ b/graphify/querylog.py @@ -1,13 +1,13 @@ """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 _NODES_RE = re.compile(r"(\d+)\s+nodes?\s+found") @@ -55,22 +55,43 @@ 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. + try: + import fcntl + except ImportError: + yield + return + lock_path = path.with_name(path.name + ".lock") + lock_path.parent.mkdir(parents=True, exist_ok=True) + fh = open(lock_path, "a+", encoding="utf-8") + 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: - return - if len(lines) <= max_records: + if not lines or len(lines) <= max_records: return overflow, keep = lines[:-max_records], lines[-max_records:] + tmp = path.with_name(path.name + ".tmp") + tmp.write_text("".join(keep), encoding="utf-8") + os.replace(tmp, path) archive = _archive_path(path) archive.parent.mkdir(parents=True, exist_ok=True) with archive.open("a", encoding="utf-8") as fh: fh.writelines(overflow) - tmp = path.with_name(path.name + ".tmp") - tmp.write_text("".join(keep), encoding="utf-8") - os.replace(tmp, path) def log_query( @@ -107,10 +128,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") - max_records = _max_records() - if max_records is not None: - _rotate_if_needed(path, max_records) + 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 bccf55fb89..bf142e2c91 100644 --- a/tests/test_querylog.py +++ b/tests/test_querylog.py @@ -1,6 +1,7 @@ """Tests for graphify.querylog.""" import json import os +import threading import pytest from pathlib import Path @@ -287,3 +288,46 @@ def test_rotation_never_raises(tmp_path, monkeypatch): 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 From 59df1d644132c11eec3c4af2b7e132804af685d3 Mon Sep 17 00:00:00 2001 From: akshit Date: Tue, 1 Sep 2026 02:09:36 +0530 Subject: [PATCH 3/3] Harden query log rotation: atomic replace, lock degrade, archive restore Use paths._atomic_replace for live rewrites, degrade to unlocked append when the lock file cannot be opened, and re-append overflow to the live log if archive append fails after replace. --- graphify/querylog.py | 32 ++++++++++++++++++++++++-------- tests/test_querylog.py | 35 +++++++++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 8 deletions(-) diff --git a/graphify/querylog.py b/graphify/querylog.py index 786bf47fa6..8db6ee7c5c 100644 --- a/graphify/querylog.py +++ b/graphify/querylog.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any, Iterator +from graphify.paths import _atomic_replace + _NODES_RE = re.compile(r"(\d+)\s+nodes?\s+found") @@ -57,8 +59,12 @@ def _archive_path(path: Path) -> Path: @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. + """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: @@ -66,7 +72,11 @@ def _query_log_lock(path: Path) -> Iterator[None]: return lock_path = path.with_name(path.name + ".lock") lock_path.parent.mkdir(parents=True, exist_ok=True) - fh = open(lock_path, "a+", encoding="utf-8") + try: + fh = open(lock_path, "a+", encoding="utf-8") + except OSError: + yield + return try: fcntl.flock(fh.fileno(), fcntl.LOCK_EX) yield @@ -85,13 +95,19 @@ def _rotate_if_needed(path: Path, max_records: int) -> None: if not lines or len(lines) <= max_records: return overflow, keep = lines[:-max_records], lines[-max_records:] - tmp = path.with_name(path.name + ".tmp") - tmp.write_text("".join(keep), encoding="utf-8") - os.replace(tmp, path) + + 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) - with archive.open("a", encoding="utf-8") as fh: - fh.writelines(overflow) + 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( diff --git a/tests/test_querylog.py b/tests/test_querylog.py index bf142e2c91..8776124cf6 100644 --- a/tests/test_querylog.py +++ b/tests/test_querylog.py @@ -331,3 +331,38 @@ def worker(prefix: str) -> None: 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() == ""