| """Append-only quarantine log: every item a stage drops, with the reason. |
| |
| ``<stage_dir>/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 |
| 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() |
| |
| 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 |
| |
| 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()} |
|
|
