"""Packed append-only transition journal."""
import hashlib
import json
import os
import re
import struct
import uuid
from collections.abc import Iterator, Mapping
from dataclasses import dataclass
from pathlib import Path
from typing import Any, BinaryIO, Self
from ._util import fsync_directory, json_bytes, visibility_attempts
from .errors import FormatError, WorkspaceCorruptionError, WorkspaceUnavailableError
from .models import DEFAULT_JOURNAL_SEGMENT_BYTES, to_base36
SEGMENT_HEADER = b"HTTK-HWJ-V1\n"
_LENGTH = struct.Struct(">Q")
_REF_PATTERN = re.compile(
r"w(?P<writer>[0-9a-f]{32})-s(?P<segment>[0-9a-z]{1,7})-o(?P<offset>[0-9a-z]{1,13})"
r"-l(?P<length>[0-9a-z]{1,13})-h(?P<checksum>[0-9a-f]{32})"
)
#: Problem codes a frame read reports. A transient code may simply mean that
#: the extended segment has not become visible on this client yet, so a reader
#: retries it until the visibility deadline expires; the rest are damage.
TRANSIENT_FRAME_PROBLEMS = frozenset({"missing_segment", "short_read", "undecodable_frame"})
#: The largest frame a segment walk will believe a length prefix about. It
#: keeps a corrupted length from turning a repair walk into a huge allocation.
MAXIMUM_FRAME_BYTES = 16 * 1024 * 1024
class _FrameProblem(Exception):
"""One frame could not be read, with a stable machine-readable code."""
def __init__(self, code: str, message: str) -> None:
super().__init__(message)
self.code = code
@property
def transient(self) -> bool:
"""Report whether waiting for the filesystem could still fix this."""
return self.code in TRANSIENT_FRAME_PROBLEMS
class JournalWriter:
"""The exclusive journal writer for one manager incarnation."""
def __init__(
self,
control_dir: Path,
*,
writer_id: str | None = None,
durable: bool = True,
maximum_segment_bytes: int = DEFAULT_JOURNAL_SEGMENT_BYTES,
) -> None:
self.control_dir = control_dir
self.writer_id = writer_id or str(uuid.uuid4())
writer_uuid = uuid.UUID(self.writer_id)
if str(writer_uuid) != self.writer_id:
raise ValueError("writer_id must be a canonical UUID")
self.durable = durable
self.maximum_segment_bytes = maximum_segment_bytes
self._writer_dir = control_dir / "journal" / self.writer_id
self._writer_dir.mkdir(parents=True, exist_ok=False)
self._segment_number = 0
self._handle = self._open_segment(self._segment_number)
def _segment_path(self, number: int) -> Path:
return self._writer_dir / f"{to_base36(number)}.hwj"
def _open_segment(self, number: int) -> BinaryIO:
if number > (1 << 32) - 1:
raise WorkspaceCorruptionError("journal segment number exhausted")
path = self._segment_path(number)
handle = path.open("x+b")
handle.write(SEGMENT_HEADER)
handle.flush()
if self.durable:
os.fsync(handle.fileno())
fsync_directory(path.parent)
return handle
def _rotate_if_needed(self, frame_bytes: int) -> None:
if self._handle.tell() == len(SEGMENT_HEADER):
return
if self._handle.tell() + frame_bytes <= self.maximum_segment_bytes:
return
self._handle.close()
self._segment_number += 1
self._handle = self._open_segment(self._segment_number)
def append(self, record: Mapping[str, object]) -> str:
"""Append *record* and return its canonical ``hwref-v1`` reference."""
payload = json_bytes(dict(record))
length_bytes = _LENGTH.pack(len(payload))
checksum = hashlib.sha256(length_bytes + payload).digest()
frame = length_bytes + payload + checksum + length_bytes
self._rotate_if_needed(len(frame))
offset = self._handle.tell()
self._handle.write(frame)
self._handle.flush()
if self.durable:
os.fsync(self._handle.fileno())
return encode_record_ref(self.writer_id, self._segment_number, offset, len(payload), checksum)
def close(self) -> None:
self._handle.close()
def __enter__(self) -> Self:
return self
def __exit__(self, exc_type: object, exc: object, traceback: object) -> None:
self.close()
[docs]
def encode_record_ref(writer_id: str, segment: int, offset: int, length: int, checksum: bytes) -> str:
"""Encode one canonical ``hwref-v1`` reference."""
return (
f"w{writer_id.replace('-', '')}"
f"-s{to_base36(segment)}"
f"-o{to_base36(offset)}"
f"-l{to_base36(length)}"
f"-h{checksum[:16].hex()}"
)
[docs]
def parse_record_ref(record_ref: str) -> tuple[str, int, int, int, str]:
"""Parse one canonical ``hwref-v1`` reference."""
match = _REF_PATTERN.fullmatch(record_ref)
if match is None:
raise FormatError(f"invalid hwref-v1 record reference: {record_ref!r}")
writer_hex = match.group("writer")
writer_id = str(uuid.UUID(hex=writer_hex))
segment = int(match.group("segment"), 36)
offset = int(match.group("offset"), 36)
length = int(match.group("length"), 36)
return writer_id, segment, offset, length, match.group("checksum")
[docs]
def segment_path(control_dir: Path, writer_id: str, segment: int) -> Path:
"""Return the segment file one record reference names."""
return control_dir / "journal" / writer_id / f"{to_base36(segment)}.hwj"
def _read_frame(path: Path, offset: int, expected_length: int, expected_prefix: str) -> dict[str, Any]:
"""Read and verify one referenced frame, or report why it is unreadable."""
try:
handle = path.open("rb")
except FileNotFoundError as exc:
raise _FrameProblem("missing_segment", f"journal segment is not present: {path}") from exc
with handle:
header = handle.read(len(SEGMENT_HEADER))
if len(header) != len(SEGMENT_HEADER):
raise _FrameProblem("short_read", "short journal segment header")
if header != SEGMENT_HEADER:
raise _FrameProblem("invalid_header", f"invalid journal segment header: {path}")
handle.seek(offset)
length_bytes = handle.read(_LENGTH.size)
if len(length_bytes) != _LENGTH.size:
raise _FrameProblem("short_read", "short frame length")
(length,) = _LENGTH.unpack(length_bytes)
if length != expected_length:
raise _FrameProblem("length_mismatch", "record reference length disagrees with journal frame")
payload = handle.read(length)
checksum = handle.read(32)
trailer = handle.read(_LENGTH.size)
if len(payload) != length or len(checksum) != 32 or len(trailer) != _LENGTH.size:
raise _FrameProblem("short_read", "short journal frame")
if trailer != length_bytes:
raise _FrameProblem("trailer_mismatch", "journal frame trailer disagrees with header")
if hashlib.sha256(length_bytes + payload).digest() != checksum:
raise _FrameProblem("checksum_mismatch", "journal frame checksum mismatch")
if checksum[:16].hex() != expected_prefix:
raise _FrameProblem("reference_mismatch", "record reference checksum mismatch")
try:
value = json.loads(payload)
except (json.JSONDecodeError, UnicodeError) as exc:
raise _FrameProblem("undecodable_frame", f"journal frame is not decodable JSON: {exc}") from exc
if not isinstance(value, dict):
raise _FrameProblem("not_an_object", "journal record is not a JSON object")
return value
[docs]
def read_record(control_dir: Path, record_ref: str, *, deadline_seconds: float | None = None) -> dict[str, Any]:
"""Read and verify a journal record, retrying visibility-short reads.
A frame that is absent, short, or undecodable may be an extension of a
segment that has not reached this client yet, so it is retried with bounded
backoff until *deadline_seconds* — the workspace's configured visibility
deadline — expires. Damage that no amount of waiting can repair is reported
at once.
"""
writer_id, segment, offset, expected_length, expected_prefix = parse_record_ref(record_ref)
path = segment_path(control_dir, writer_id, segment)
last_error: _FrameProblem | None = None
for _ in visibility_attempts(deadline_seconds):
try:
return _read_frame(path, offset, expected_length, expected_prefix)
except _FrameProblem as exc:
if not exc.transient:
raise WorkspaceCorruptionError(str(exc)) from exc
last_error = exc
raise WorkspaceUnavailableError(f"journal record is not coherently visible: {record_ref}") from last_error
[docs]
@dataclass(frozen=True)
class RecordVerification:
"""The outcome of reading one referenced frame without raising."""
record_ref: str
frame: dict[str, Any] | None
problem: str | None
detail: str
@property
def ok(self) -> bool:
"""Report whether the referenced frame was read and verified."""
return self.frame is not None
[docs]
def verify_record(control_dir: Path, record_ref: str, *, deadline_seconds: float | None = None) -> RecordVerification:
"""Read one referenced frame, reporting damage rather than raising.
This is the reading half of a workspace check: it distinguishes a segment
that is gone from one that is truncated, corrupt, or simply not holding the
frame the reference names, which is what a repair decision needs.
"""
try:
writer_id, segment, offset, expected_length, expected_prefix = parse_record_ref(record_ref)
except (FormatError, ValueError) as exc:
return RecordVerification(record_ref, None, "invalid_record_ref", str(exc))
path = segment_path(control_dir, writer_id, segment)
last_error: _FrameProblem | None = None
for _ in visibility_attempts(deadline_seconds):
try:
frame = _read_frame(path, offset, expected_length, expected_prefix)
except _FrameProblem as exc:
last_error = exc
if exc.transient:
continue
break
except OSError as exc:
return RecordVerification(record_ref, None, "unreadable_segment", str(exc))
return RecordVerification(record_ref, frame, None, "")
if last_error is None: # pragma: no cover - a schedule always has one attempt
return RecordVerification(record_ref, None, "unreadable_frame", "the frame was never probed")
return RecordVerification(record_ref, None, last_error.code, str(last_error))
[docs]
@dataclass(frozen=True)
class JournalFrame:
"""One intact frame found by walking a segment from its header."""
record_ref: str
writer_id: str
segment: int
offset: int
frame: dict[str, Any]
[docs]
def iter_segment_frames(path: Path, writer_id: str, segment: int) -> Iterator[JournalFrame]:
"""Yield every intact frame of one segment.
The walk is deliberately forgiving. A damaged frame whose framing is still
intact is skipped, because the frames behind it remain locatable and are
exactly what a repair is looking for; a torn or partially visible tail is
the normal state of a segment a live writer is appending to and simply ends
the walk.
"""
try:
handle = path.open("rb")
except OSError:
return
with handle:
if handle.read(len(SEGMENT_HEADER)) != SEGMENT_HEADER:
return
while True:
offset = handle.tell()
length_bytes = handle.read(_LENGTH.size)
if len(length_bytes) != _LENGTH.size:
return
(length,) = _LENGTH.unpack(length_bytes)
if length > MAXIMUM_FRAME_BYTES:
return
payload = handle.read(length)
checksum = handle.read(32)
trailer = handle.read(_LENGTH.size)
if len(payload) != length or len(checksum) != 32 or trailer != length_bytes:
return
if hashlib.sha256(length_bytes + payload).digest() != checksum:
continue
try:
value = json.loads(payload)
except (json.JSONDecodeError, UnicodeError):
continue
if not isinstance(value, dict):
continue
yield JournalFrame(
record_ref=encode_record_ref(writer_id, segment, offset, length, checksum),
writer_id=writer_id,
segment=segment,
offset=offset,
frame=value,
)
[docs]
def iter_journal_frames(control_dir: Path) -> Iterator[JournalFrame]:
"""Yield every intact frame of every segment of every writer."""
journal = control_dir / "journal"
if not journal.is_dir():
return
for writer_dir in sorted(journal.iterdir()):
if not writer_dir.is_dir():
continue
try:
writer_id = str(uuid.UUID(writer_dir.name))
except ValueError:
continue
if writer_id != writer_dir.name:
continue
for path in sorted(writer_dir.glob("*.hwj")):
try:
segment = int(path.stem, 36)
except ValueError:
continue
yield from iter_segment_frames(path, writer_id, segment)