"""Append-only quarantine log: every item a stage drops, with the reason. ``/quarantine.jsonl`` holds one JSON object per line:: {"details": {...}, "item_id": "...", "reason": "...", "stage": "..."} Nothing is dropped silently (CLAUDE.md rule 2): a stage appends a :class:`QuarantineRecord` for each rejected item, and reports :meth:`QuarantineLog.counts_by_reason` in its run report. Durability follows the results-file protocol of :mod:`nida.bench.protocol`: each batch is written with a single ``write`` and then flushed and fsynced, so an interruption leaves a valid prefix plus at most one partial trailing line. A partial trailing line is handled byte-wise (it may end inside a multi-byte UTF-8 character): * :meth:`QuarantineLog.load` raises :class:`QuarantineError` on it, unless ``repair=True``; * :meth:`QuarantineLog.append` refuses to write after it, because the new record would be glued to the fragment and corrupt both; * :meth:`QuarantineLog.recover` cuts the fragment, logs a warning with its size and a preview, and returns the number of bytes removed. A fragment is dropped even if it happens to be valid JSON (only the final newline was lost): a record counts as durable only once its newline is on disk. The item it described was not confirmed quarantined, so the stage reprocesses it. A malformed line anywhere *before* the last newline is never repaired: that is corruption, not an interrupted append, and is always an error. The log does not de-duplicate: a stage rerun that quarantines the same item again appends another record. Use :meth:`QuarantineLog.quarantined_ids` to skip known items on resume. """ from __future__ import annotations import json import logging import os from collections import Counter from collections.abc import Iterable from pathlib import Path from nida.data.types import QuarantineRecord __all__ = ["QUARANTINE_FILENAME", "QuarantineError", "QuarantineLog"] QUARANTINE_FILENAME = "quarantine.jsonl" _PREVIEW_BYTES = 80 _log = logging.getLogger(__name__) class QuarantineError(ValueError): """The quarantine file is malformed or a record cannot be serialised.""" def _encode(record: QuarantineRecord) -> bytes: try: text = json.dumps( { "item_id": record.item_id, "stage": record.stage, "reason": record.reason, "details": record.details, }, ensure_ascii=False, sort_keys=True, allow_nan=False, ) return (text + "\n").encode("utf-8") except (TypeError, ValueError) as err: raise QuarantineError(f"record {record.item_id!r} is not serialisable: {err}") from err def _decode(line: str, where: str) -> QuarantineRecord: try: obj = json.loads(line) except json.JSONDecodeError as err: raise QuarantineError(f"{where}: invalid JSON ({err.msg})") from err if not isinstance(obj, dict) or set(obj) != {"item_id", "stage", "reason", "details"}: raise QuarantineError( f"{where}: record must be an object with item_id/stage/reason/details" ) item_id, stage, reason, details = obj["item_id"], obj["stage"], obj["reason"], obj["details"] for name, value in (("item_id", item_id), ("stage", stage), ("reason", reason)): if not isinstance(value, str): raise QuarantineError(f"{where}: {name} must be a string") if not isinstance(details, dict): raise QuarantineError(f"{where}: details must be an object") try: return QuarantineRecord(item_id=item_id, stage=stage, reason=reason, details=details) except ValueError as err: raise QuarantineError(f"{where}: {err}") from err class QuarantineLog: """The quarantine file of one stage directory.""" def __init__(self, stage_dir: Path) -> None: self._path = stage_dir / QUARANTINE_FILENAME @property def path(self) -> Path: return self._path def append(self, records: Iterable[QuarantineRecord]) -> int: """Append a batch of records durably (one write, then flush + fsync). All records are serialised before anything is written, so a record that cannot be serialised leaves the file untouched. Creates the stage directory if needed. Returns: The number of records appended (``0`` for an empty batch; the file is then not touched). Raises: QuarantineError: if a record is not JSON-serialisable (non-finite float, unsupported type, unencodable text), or the file ends in a partial line left by an interrupted append (call :meth:`recover` first). """ encoded = [_encode(record) for record in records] if not encoded: return 0 self._path.parent.mkdir(parents=True, exist_ok=True) if self._ends_mid_line(): raise QuarantineError( f"{self._path}: file ends in an incomplete line; call recover() before appending" ) with self._path.open("ab") as handle: handle.write(b"".join(encoded)) handle.flush() os.fsync(handle.fileno()) return len(encoded) def _ends_mid_line(self) -> bool: if not self._path.exists(): return False with self._path.open("rb") as handle: handle.seek(0, os.SEEK_END) if handle.tell() == 0: return False handle.seek(-1, os.SEEK_END) return handle.read(1) != b"\n" def recover(self) -> int: """Cut a partial trailing line, if any, and return the number of bytes removed. Logs a warning (path, byte count, preview of the fragment) when something is cut. A missing or intact file is left untouched and yields ``0``. """ if not self._path.exists(): return 0 raw = self._path.read_bytes() keep = raw.rfind(b"\n") + 1 # 0 when there is no complete line fragment = raw[keep:] if not fragment: return 0 _log.warning( "%s: removing %d bytes of an incomplete trailing line (interrupted append): %r", self._path, len(fragment), fragment[:_PREVIEW_BYTES], ) with self._path.open("r+b") as handle: handle.truncate(keep) handle.flush() os.fsync(handle.fileno()) return len(fragment) def load(self, *, repair: bool = False) -> list[QuarantineRecord]: """Read all records in file order; a missing file yields ``[]``. Raises: QuarantineError: on a malformed or non-UTF-8 complete line, or on a partial trailing line unless ``repair=True`` (which cuts it via :meth:`recover`). """ if not self._path.exists(): return [] raw = self._path.read_bytes() # Split at the last newline *in bytes*: the trailing fragment must not be decoded. keep = raw.rfind(b"\n") + 1 body, trailing = raw[:keep], raw[keep:] if trailing and not repair: raise QuarantineError( f"{self._path}: incomplete trailing line ({len(trailing)} bytes); " "use repair=True or recover()" ) try: text = body.decode("utf-8") except UnicodeDecodeError as err: raise QuarantineError( f"{self._path}: complete lines are not valid UTF-8 ({err})" ) from err # Split on "\n" only: str.splitlines() would also split on U+2028/U+0085 inside JSON. lines = text.split("\n")[:-1] if text else [] records = [_decode(line, f"{self._path}:{number}") for number, line in enumerate(lines, 1)] if trailing: self.recover() return records def counts_by_reason(self) -> dict[str, int]: """Number of records per ``reason``, most frequent first (ties by reason).""" counter = Counter(record.reason for record in self.load()) return dict(sorted(counter.items(), key=lambda item: (-item[1], item[0]))) def quarantined_ids(self) -> set[str]: """Distinct ``item_id`` values in the log (for skipping known items on resume).""" return {record.item_id for record in self.load()}