File size: 8,373 Bytes
c0bce59 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 | """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 # 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()}
|