perf(store): read the operations journal from its end

planfile/core/jsonl_tail.py

@@ -0,0 +1,97 @@
+"""Bounded newest-first reads of append-only JSONL journals.
+
+An append-only journal is written at its end and read from its end, so a reader
+that starts at byte zero pays for the entire history to answer a question about
+the last few hundred rows. This walks backwards in chunks and stops as soon as
+it has enough, with a byte budget so a filter that matches nothing recent cannot
+turn into a full scan.
+"""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from pathlib import Path
+
+#: Default reverse-walk budget.
+#:
+#: An unfiltered read fills its limit from the last few kilobytes, so the budget
+#: only ever matters to a filtered query, where it decides how far back that
+#: filter may search. 256 MiB covers well over the recent history of the largest
+#: observed journal (379 MB) while still turning the unbounded worst case into a
+#: bounded one. A filtered query returns the matches inside this window, newest
+#: first; it is not a promise of every match in history.
+OPERATIONS_TAIL_MAX_BYTES = 256 * 1024 * 1024
+
+_CHUNK_BYTES = 1024 * 1024
+
+
+def _parse(raw: bytes) -> dict | None:
+    if not raw.strip():
+        return None
+    try:
+        row = json.loads(raw)
+    except ValueError:
+        return None
+    return row if isinstance(row, dict) else None
+
+
+def read_jsonl_tail(
+    path: Path,
+    *,
+    limit: int,
+    keep: Callable[[dict], bool] | None = None,
+    max_bytes: int = OPERATIONS_TAIL_MAX_BYTES,
+    chunk_bytes: int = _CHUNK_BYTES,
+) -> list[dict]:
+    """Return up to ``limit`` matching rows, newest first.
+
+    Blank and malformed lines are skipped, matching the forward reader this
+    replaces. A missing file is an empty result, not an error.
+    """
+    wanted = max(1, int(limit))
+    budget = max(1, int(max_bytes))
+    step = max(1024, int(chunk_bytes))
+    rows: list[dict] = []
+    try:
+        handle = path.open("rb")
+    except (FileNotFoundError, NotADirectoryError):
+        return []
+    with handle:
+        handle.seek(0, 2)
+        position = handle.tell()
+        carry = b""
+        scanned = 0
+        while position > 0 and len(rows) < wanted and scanned < budget:
+            # Clamp the chunk to what is left of the budget, so a small budget
+            # really is a small read rather than one oversized first chunk.
+            reach = min(step, budget - scanned)
+            start = max(0, position - reach)
+            handle.seek(start)
+            block = handle.read(position - start)
+            scanned += len(block)
+            buffer = block + carry
+            if start == 0:
+                # The head of the file is a complete line by definition.
+                segments = buffer.split(b"\n")
+                carry = b""
+            else:
+                boundary = buffer.find(b"\n")
+                if boundary < 0:
+                    # No line ended inside this chunk; carry it further back.
+                    carry = buffer
+                    position = start
+                    continue
+                carry = buffer[:boundary]
+                segments = buffer[boundary + 1 :].split(b"\n")
+            for raw in reversed(segments):
+                if len(rows) >= wanted:
+                    break
+                row = _parse(raw)
+                if row is None:
+                    continue
+                if keep is not None and not keep(row):
+                    continue
+                rows.append(row)
+            position = start
+    return rows

planfile/core/store.py

@@ -15,6 +15,7 @@
 import yaml
 from pydantic import BaseModel
 
+from .jsonl_tail import OPERATIONS_TAIL_MAX_BYTES, read_jsonl_tail
 from .models import TICKET_CONTRACT_VERSION, Ticket
 from .store_files import StoreFileMixin
 from .store_tickets import TicketStoreMixin
@@ -463,27 +464,36 @@ def _append_ticket_evidence_event(self, ticket_id: str, event: dict) -> None:
             handle.flush()
             os.fsync(handle.fileno())
 
-    def operational_events(self, *, limit: int = 200, ticket_id: str | None = None) -> list[dict]:
-        """Read the append-only operational journal, newest first."""
-        from collections import deque
-
-        bounded: deque[dict] = deque(maxlen=max(1, min(int(limit), 5000)))
-        try:
-            source = self._operations_path.open("r", encoding="utf-8")
-        except FileNotFoundError:
-            return []
-        with source:
-            for line in source:
-                if not line.strip():
-                    continue
-                try:
-                    row = json.loads(line)
-                except ValueError:
-                    continue
-                if ticket_id and str((row.get("event") or {}).get("ticket_id") or "") != str(ticket_id):
-                    continue
-                bounded.append(row)
-        return list(reversed(bounded))
+    def operational_events(
+        self,
+        *,
+        limit: int = 200,
+        ticket_id: str | None = None,
+        max_bytes: int = OPERATIONS_TAIL_MAX_BYTES,
+    ) -> list[dict]:
+        """Read the append-only operational journal, newest first.
+
+        The journal only grows, and every caller wants its newest end, so the
+        read walks backwards and stops as soon as it has ``limit`` rows. Reading
+        forward meant parsing the whole file to return the last few hundred
+        rows: on the observed store that was 379 MB and ~170k `json.loads` calls
+        per request, on the single-worker event loop that also serves ticket
+        writes.
+
+        ``max_bytes`` bounds the walk so a filter that matches nothing recent
+        degrades to a short read instead of a full scan; such a query returns
+        the matches inside the budget rather than every match in history.
+        """
+        return read_jsonl_tail(
+            self._operations_path,
+            limit=max(1, min(int(limit), 5000)),
+            max_bytes=max_bytes,
+            keep=(
+                None
+                if not ticket_id
+                else lambda row: str((row.get("event") or {}).get("ticket_id") or "") == str(ticket_id)
+            ),
+        )
 
     @contextmanager
     def mutation_lock(self):

tests/test_jsonl_tail.py

@@ -0,0 +1,164 @@
+"""Bounded newest-first JSONL reads must match the forward reader they replace."""
+
+from __future__ import annotations
+
+import json
+import random
+from collections import deque
+from pathlib import Path
+
+import pytest
+
+from planfile.core.jsonl_tail import read_jsonl_tail
+
+
+def forward_reference(path: Path, *, limit: int, ticket_id: str | None = None) -> list[dict]:
+    """The original full-file implementation, kept as the oracle."""
+    bounded: deque[dict] = deque(maxlen=max(1, min(int(limit), 5000)))
+    try:
+        source = path.open("r", encoding="utf-8")
+    except FileNotFoundError:
+        return []
+    with source:
+        for line in source:
+            if not line.strip():
+                continue
+            try:
+                row = json.loads(line)
+            except ValueError:
+                continue
+            if ticket_id and str((row.get("event") or {}).get("ticket_id") or "") != str(ticket_id):
+                continue
+            bounded.append(row)
+    return list(reversed(bounded))
+
+
+def keep_ticket(ticket_id: str):
+    return lambda row: str((row.get("event") or {}).get("ticket_id") or "") == str(ticket_id)
+
+
+def write_rows(path: Path, rows: list[dict], *, trailing_newline: bool = True) -> None:
+    body = "\n".join(json.dumps(row) for row in rows)
+    path.write_text(body + ("\n" if trailing_newline else ""), encoding="utf-8")
+
+
+def journal(index: int, ticket: str = "PLF-1") -> dict:
+    return {"seq": index, "event": {"ticket_id": ticket, "oql": "ticket.update"}}
+
+
+def test_missing_file_is_an_empty_result(tmp_path):
+    assert read_jsonl_tail(tmp_path / "absent.jsonl", limit=10) == []
+
+
+def test_empty_file_is_an_empty_result(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    path.write_text("", encoding="utf-8")
+    assert read_jsonl_tail(path, limit=10) == []
+
+
+def test_rows_come_back_newest_first(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    write_rows(path, [journal(i) for i in range(5)])
+    assert [row["seq"] for row in read_jsonl_tail(path, limit=10)] == [4, 3, 2, 1, 0]
+
+
+def test_limit_takes_the_newest_rows(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    write_rows(path, [journal(i) for i in range(100)])
+    assert [row["seq"] for row in read_jsonl_tail(path, limit=3)] == [99, 98, 97]
+
+
+def test_a_file_without_a_trailing_newline_keeps_its_last_row(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    write_rows(path, [journal(i) for i in range(4)], trailing_newline=False)
+    assert [row["seq"] for row in read_jsonl_tail(path, limit=2)] == [3, 2]
+
+
+def test_blank_and_malformed_lines_are_skipped(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    path.write_text(
+        "\n".join([
+            json.dumps(journal(0)),
+            "",
+            "{not json",
+            "   ",
+            json.dumps(journal(1)),
+            "[1, 2, 3]",
+        ]) + "\n",
+        encoding="utf-8",
+    )
+    assert [row["seq"] for row in read_jsonl_tail(path, limit=10)] == [1, 0]
+
+
+def test_a_filter_selects_only_matching_rows(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    rows = [journal(i, "PLF-1" if i % 2 == 0 else "PLF-2") for i in range(20)]
+    write_rows(path, rows)
+    got = read_jsonl_tail(path, limit=3, keep=keep_ticket("PLF-2"))
+    assert [row["seq"] for row in got] == [19, 17, 15]
+
+
+def test_rows_spanning_chunk_boundaries_are_not_split(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    rows = [{"seq": i, "pad": "x" * 300, "event": {"ticket_id": "PLF-1"}} for i in range(200)]
+    write_rows(path, rows)
+    # A chunk far smaller than one row forces every boundary case.
+    got = read_jsonl_tail(path, limit=200, chunk_bytes=1024, max_bytes=1 << 30)
+    assert [row["seq"] for row in got] == list(reversed(range(200)))
+
+
+def test_a_single_row_longer_than_a_chunk_is_recovered(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    write_rows(path, [{"seq": 0, "pad": "y" * 20000, "event": {"ticket_id": "PLF-1"}}])
+    got = read_jsonl_tail(path, limit=5, chunk_bytes=1024, max_bytes=1 << 30)
+    assert [row["seq"] for row in got] == [0]
+
+
+def test_the_byte_budget_stops_a_sparse_filter_early(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    # The only match sits at the head, far outside the budget.
+    rows = [journal(0, "PLF-OLD")] + [journal(i, "PLF-NEW") for i in range(1, 500)]
+    write_rows(path, rows)
+    assert read_jsonl_tail(path, limit=5, keep=keep_ticket("PLF-OLD"), max_bytes=1024) == []
+    # With room to walk the whole file, the same query finds it.
+    found = read_jsonl_tail(path, limit=5, keep=keep_ticket("PLF-OLD"), max_bytes=1 << 30)
+    assert [row["seq"] for row in found] == [0]
+
+
+def test_the_budget_never_truncates_an_unfiltered_tail_read(tmp_path):
+    path = tmp_path / "operations.jsonl"
+    write_rows(path, [journal(i) for i in range(2000)])
+    got = read_jsonl_tail(path, limit=200, max_bytes=64 * 1024)
+    assert len(got) == 200
+    assert got[0]["seq"] == 1999
+
+
+@pytest.mark.parametrize("seed", range(12))
+def test_it_agrees_with_the_forward_reader_on_random_journals(tmp_path, seed):
+    rng = random.Random(seed)
+    path = tmp_path / "operations.jsonl"
+    lines: list[str] = []
+    for index in range(rng.randint(0, 400)):
+        roll = rng.random()
+        if roll < 0.05:
+            lines.append("")
+        elif roll < 0.10:
+            lines.append("{ broken")
+        else:
+            ticket = rng.choice(["PLF-1", "PLF-2", "PLF-3"])
+            pad = "z" * rng.randint(0, 400)
+            lines.append(json.dumps({"seq": index, "pad": pad, "event": {"ticket_id": ticket}}))
+    body = "\n".join(lines)
+    path.write_text(body + ("\n" if rng.random() < 0.8 else ""), encoding="utf-8")
+
+    limit = rng.randint(1, 50)
+    for ticket_id in (None, "PLF-2"):
+        expected = forward_reference(path, limit=limit, ticket_id=ticket_id)
+        got = read_jsonl_tail(
+            path,
+            limit=limit,
+            keep=None if ticket_id is None else keep_ticket(ticket_id),
+            max_bytes=1 << 30,
+            chunk_bytes=rng.choice([1024, 4096, 1 << 20]),
+        )
+        assert got == expected, f"seed={seed} ticket={ticket_id} limit={limit}"