journal.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oc_chat/registry/source_migrations/journal.py
 
# OCT-LINT: disable=tests_exist reason="Covered by oc_chat/tests/registry/test_source_migrate_journal.py"
"""
The per-run JSONL recovery journal (Plan 142m §4) — schema v1.
 
Purpose
-------
A byte-for-byte PREIMAGE backup + a durable, crash-tolerant event log for the
destructive ``migrate`` engine. The journal is a **JSON Lines** file
(``<run-id>.jsonl``): line 1 is a header, then one preimage line per WRITE-SET
record, then per write TWO event lines (``planned_postimage`` before the
``os.replace`` — the durable rollback anchor — and ``applied`` after), then a
terminal ``resolved`` event that IS the binding state transition.
 
Every line is flushed + fsync'd. A HOST crash truncates at a line boundary (a
CONSISTENT partial, folded as-is); an in-process FAILED append (ENOSPC mid-write)
can leave a TORN final line — the APPEND-BOUNDARY RULE truncates it to the last
complete line before the next append. A published journal whose header is
unparseable, or whose body has a non-JSON complete line, is CORRUPT ⇒ UNRESOLVED
(corruption cannot be proven NOT to be recovery state).
 
Lifecycle (the sole normative definition):
- ``<run-id>.jsonl.partial`` — being built; NEVER "unresolved".
- ``<run-id>.jsonl`` — published (``os.replace``) before the first source write;
  UNRESOLVED until its tail carries a terminal ``resolved`` event.
- ``.journal/<target>/archive/<run-id>.jsonl`` — best-effort tidy move AFTER the
  ``resolved`` append (which is the real transition); ``archive/`` is NEVER scanned.
 
Responsibilities
----------------
- Build/publish the preimage journal, append fsync'd event lines, resolve+archive.
- Fold a published journal from disk (the rollback + step-5 recovery basis) and
  project its EXACT byte size for the disk preflight.
 
Contracts
---------
- Every line is flushed + fsync'd; publish is atomic (``os.replace``) BEFORE the
  first source write; the terminal ``resolved`` event is the binding state change.
- A torn final line is a consistent partial; a corrupt/unparseable journal is
  UNRESOLVED (fail-closed).
 
Diagnostics
-----------
Domain: OC_CHAT.MAIN
Levels:
    L2 - journal open/close/publish/resolve/archive
    L3 - per-line fold / append-boundary repair
    L4 - module load
 
Dependencies
------------
- oc_diagnostics (structured debug logging via _dbg)
"""
from __future__ import annotations
 
import base64
import hashlib
import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Optional
 
from oc_diagnostics import _dbg
 
_dbg("OC_CHAT.MAIN", "module", "loaded", 4)
 
JOURNAL_SCHEMA_VERSION = 1
EVENT_PLANNED_POSTIMAGE = "planned_postimage"
EVENT_APPLIED = "applied"
EVENT_RESOLVED = "resolved"
OUTCOME_SUCCESS = "success"
OUTCOME_ROLLED_BACK = "rolled_back"
 
_PARTIAL_SUFFIX = ".jsonl.partial"
_PUBLISHED_SUFFIX = ".jsonl"
_ARCHIVE_DIRNAME = "archive"
 
 
# --- IO seams (monkeypatched by tests for fsync-ordering + failure injection) --
def _fsync(fileobj: "object") -> None:
    """fsync a file object (best-effort seam; overridable in tests)."""
    os.fsync(fileobj.fileno())  # type: ignore[attr-defined]
 
 
def _replace(src: Path, dst: Path) -> None:
    """Atomic ``os.replace`` seam (overridable in tests for ordering/failure)."""
    os.replace(src, dst)
 
 
def _fsync_dir(path: Path) -> None:
    """Best-effort directory fsync (unsupported on Windows — documented)."""
    func = "_fsync_dir"
    try:
        fd = os.open(str(path), os.O_RDONLY)
    except OSError:
        return
    try:
        os.fsync(fd)
    except OSError:
        _dbg("OC_CHAT.MAIN", func, "directory fsync unsupported (platform)", 3)
    finally:
        os.close(fd)
 
 
def _dumps(obj: dict) -> str:
    """The single JSON encoder + LF convention every journal line uses (the exact
    form the step-10 disk preflight projects)."""
    return json.dumps(obj, sort_keys=True, separators=(",", ":")) + "\n"
 
 
def _write_line(fileobj: "object", obj: dict) -> None:
    """Write one JSON line, flush + fsync it (the per-line durability rule)."""
    fileobj.write(_dumps(obj))  # type: ignore[attr-defined]
    fileobj.flush()  # type: ignore[attr-defined]
    _fsync(fileobj)
 
 
# --- preimage / header --------------------------------------------------------
@dataclass(frozen=True)
class PreimageRecord:
    """One WRITE-SET record's byte-for-byte preimage (§4)."""
 
    rel_path: str
    preimage_bytes: bytes
    file_mode: Optional[int]
 
    @property
    def sha256(self) -> str:
        return hashlib.sha256(self.preimage_bytes).hexdigest()
 
    def to_line(self) -> dict:
        return {
            "rel_path": self.rel_path,
            "action": "replace",
            "exists_before": True,
            "preimage_b64": base64.b64encode(self.preimage_bytes).decode("ascii"),
            "file_mode": self.file_mode,
            "sha256": self.sha256,
        }
 
 
def build_preimage(rel_path: str, raw_bytes: bytes, file_mode: Optional[int]) -> PreimageRecord:
    """Build a preimage record from a write-set candidate's raw on-disk bytes."""
    func = "build_preimage"
    _dbg("OC_CHAT.MAIN", func, f"preimage {rel_path} ({len(raw_bytes)}B)", 3)
    return PreimageRecord(rel_path=rel_path, preimage_bytes=raw_bytes, file_mode=file_mode)
 
 
def header_dict(target_schema_version: str, source_root: str, run_id: str) -> dict:
    """The journal header line (line 1)."""
    return {
        "journal_schema_version": JOURNAL_SCHEMA_VERSION,
        "target_schema_version": target_schema_version,
        "source_root": source_root,
        "run_id": run_id,
    }
 
 
def project_journal_bytes(
    header: dict,
    preimages: list[PreimageRecord],
    postimage_shas: list[str],
) -> int:
    """Serialize the PROJECTED journal with the SAME encoder + newline convention
    and return its exact UTF-8 byte length (the step-10 disk-preflight input):
    header + one preimage line per write-set record + one ``planned_postimage`` +
    one ``applied`` per write + the terminal ``resolved`` line. Exact — never a
    scalar multiplier (per-line JSON structure dominates for many tiny records)."""
    func = "project_journal_bytes"
    lines = [_dumps(header)]
    for preimage in preimages:
        lines.append(_dumps(preimage.to_line()))
    for preimage, sha in zip(preimages, postimage_shas):
        lines.append(_dumps({"event": EVENT_PLANNED_POSTIMAGE, "rel_path": preimage.rel_path,
                             "postimage_sha256": sha}))
        lines.append(_dumps({"event": EVENT_APPLIED, "rel_path": preimage.rel_path}))
    lines.append(_dumps({"event": EVENT_RESOLVED, "outcome": OUTCOME_SUCCESS}))
    total = sum(len(line.encode("utf-8")) for line in lines)
    _dbg("OC_CHAT.MAIN", func, f"projected journal {total}B ({len(preimages)} preimages)", 3)
    return total
 
 
# --- reading / folding --------------------------------------------------------
@dataclass
class FoldedEntry:
    """A preimage entry folded with its event lines (the rollback consults these)."""
 
    preimage: dict
    planned_postimage_sha: Optional[str] = None
    applied: bool = False
 
    @property
    def rel_path(self) -> str:
        return self.preimage["rel_path"]
 
 
@dataclass(frozen=True)
class ParsedJournal:
    """A folded view of one on-disk journal (the rollback + step-5 recovery basis)."""
 
    path: Path
    header: Optional[dict]
    entries: list[FoldedEntry] = field(default_factory=list)
    resolved: bool = False
    outcome: Optional[str] = None
    corrupt: bool = False
 
    @property
    def unresolved(self) -> bool:
        """A published journal is UNRESOLVED iff corrupt OR its tail lacks a
        terminal ``resolved`` event (fail-closed: corruption ⇒ unresolved)."""
        return self.corrupt or not self.resolved
 
    @property
    def source_root(self) -> Optional[str]:
        return (self.header or {}).get("source_root")
 
    @property
    def target_schema_version(self) -> Optional[str]:
        return (self.header or {}).get("target_schema_version")
 
    def entry_for(self, rel_path: str) -> Optional[FoldedEntry]:
        for entry in self.entries:
            if entry.rel_path == rel_path:
                return entry
        return None
 
 
def read_journal(path: Path) -> ParsedJournal:
    """Read + fold a published journal from DISK (the rollback data source + the
    step-5 recovery check). A TORN final line (no trailing newline, unparseable) is
    a consistent partial — ignored, NOT corrupt; a non-JSON COMPLETE line, or an
    unparseable header, is corrupt ⇒ UNRESOLVED."""
    func = "read_journal"
    path = Path(path)
    raw = path.read_bytes()
    text = raw.decode("utf-8", errors="replace")
    if text == "":
        return ParsedJournal(path=path, header=None, corrupt=True)
    ends_with_newline = text.endswith("\n")
    segments = text.split("\n")
    if ends_with_newline:
        segments = segments[:-1]  # drop the empty trailing segment
    # The last segment is a TORN candidate only when the file did NOT end with \n.
    complete = segments if ends_with_newline else segments[:-1]
    torn_tail = None if ends_with_newline else segments[-1]
 
    header: Optional[dict] = None
    entries: list[FoldedEntry] = []
    resolved = False
    outcome: Optional[str] = None
    corrupt = False
    by_path: dict[str, FoldedEntry] = {}
 
    for index, line in enumerate(complete):
        if line == "":
            continue
        try:
            obj = json.loads(line)
        except ValueError:
            corrupt = True  # a non-JSON COMPLETE body line ⇒ corrupt
            continue
        if index == 0:
            if isinstance(obj, dict) and "journal_schema_version" in obj:
                header = obj
            else:
                corrupt = True
            continue
        if not isinstance(obj, dict):
            corrupt = True
            continue
        event = obj.get("event")
        rel_path = str(obj.get("rel_path", ""))
        if event == EVENT_RESOLVED:
            resolved = True
            outcome = obj.get("outcome")
        elif event == EVENT_PLANNED_POSTIMAGE:
            entry = by_path.get(rel_path)
            if entry is not None:
                entry.planned_postimage_sha = obj.get("postimage_sha256")
        elif event == EVENT_APPLIED:
            entry = by_path.get(rel_path)
            if entry is not None:
                entry.applied = True
        elif "preimage_b64" in obj:
            entry = FoldedEntry(preimage=obj)
            entries.append(entry)
            by_path[entry.rel_path] = entry
        else:
            corrupt = True
    if torn_tail not in (None, ""):
        try:
            json.loads(torn_tail)
        except ValueError:
            _dbg("OC_CHAT.MAIN", func, "torn final line — consistent partial", 3)
    if header is None:
        corrupt = True
    _dbg("OC_CHAT.MAIN", func, f"folded {path.name}: resolved={resolved} corrupt={corrupt}", 2)
    return ParsedJournal(
        path=path, header=header, entries=entries, resolved=resolved,
        outcome=outcome, corrupt=corrupt,
    )
 
 
def preimage_bytes_of(entry: FoldedEntry) -> bytes:
    """Decode a folded entry's preimage bytes (base64)."""
    return base64.b64decode(entry.preimage["preimage_b64"])
 
 
def scan_unresolved(journal_root: Path) -> list[ParsedJournal]:
    """Step 5: return every UNRESOLVED published journal under
    ``journal_root/<target>/`` — scanned NON-recursively per target dir, so
    ``archive/`` is never scanned. Recovery state is a ROOT-level hazard (ANY
    target). ``.partial`` files are excluded."""
    func = "scan_unresolved"
    journal_root = Path(journal_root)
    unresolved: list[ParsedJournal] = []
    if not journal_root.is_dir():
        return unresolved
    for target_dir in sorted(journal_root.iterdir()):
        if not target_dir.is_dir():
            continue
        for entry in sorted(target_dir.iterdir()):
            if not entry.is_file():
                continue  # archive/ (a dir) is skipped; non-recursive
            name = entry.name
            if not name.endswith(_PUBLISHED_SUFFIX) or name.endswith(_PARTIAL_SUFFIX):
                continue
            parsed = read_journal(entry)
            if parsed.unresolved:
                unresolved.append(parsed)
    _dbg("OC_CHAT.MAIN", func, f"{len(unresolved)} unresolved journal(s)", 2)
    return unresolved
 
 
# --- writing ------------------------------------------------------------------
class JournalWriter:
    """Builds, publishes, appends-to, resolves and archives one run's journal."""
 
    def __init__(self, target_dir: Path, run_id: str, header: dict) -> None:
        self.target_dir = Path(target_dir)
        self.run_id = run_id
        self.header = header
        self.partial_path = self.target_dir / f"{run_id}{_PARTIAL_SUFFIX}"
        self.final_path = self.target_dir / f"{run_id}{_PUBLISHED_SUFFIX}"
        self.published = False
 
    def build(self, preimages: list[PreimageRecord]) -> None:
        """Write header + all preimage lines to ``<run-id>.jsonl.partial`` (fsync
        per line). Raises ``OSError`` on a write failure (the caller cleans up)."""
        func = "JournalWriter.build"
        self.target_dir.mkdir(parents=True, exist_ok=True)
        with open(self.partial_path, "w", encoding="utf-8", newline="\n") as fileobj:
            _write_line(fileobj, self.header)
            for preimage in preimages:
                _write_line(fileobj, preimage.to_line())
        _dbg("OC_CHAT.MAIN", func, f"built partial ({len(preimages)} preimages)", 2)
 
    def publish(self) -> None:
        """Atomically publish ``.partial`` → ``.jsonl`` (BEFORE the first source
        write) + best-effort directory fsync."""
        func = "JournalWriter.publish"
        _replace(self.partial_path, self.final_path)
        _fsync_dir(self.target_dir)
        self.published = True
        _dbg("OC_CHAT.MAIN", func, "journal published", 2)
 
    def cleanup_partial(self) -> None:
        """Best-effort delete of the ``.partial`` on a build/publish failure (a
        surviving ``.partial`` never refuses — only a published ``.jsonl`` does)."""
        func = "JournalWriter.cleanup_partial"
        try:
            self.partial_path.unlink(missing_ok=True)
        except OSError:
            _dbg("OC_CHAT.MAIN", func, "partial cleanup failed (harmless)", 1, override=True)
 
    def _repair_torn_tail(self) -> None:
        """APPEND-BOUNDARY RULE: before appending, truncate any torn final line
        (an in-process failed append can leave partial bytes with no newline) so
        the next append never fuses into a non-JSON line."""
        func = "JournalWriter._repair_torn_tail"
        raw = self.final_path.read_bytes()
        if raw == b"" or raw.endswith(b"\n"):
            return
        last_newline = raw.rfind(b"\n")
        keep = raw[: last_newline + 1] if last_newline != -1 else b""
        with open(self.final_path, "wb") as fileobj:
            fileobj.write(keep)
            fileobj.flush()
            _fsync(fileobj)
        _dbg("OC_CHAT.MAIN", func, "truncated torn final line", 3)
 
    def append_event(self, event: dict) -> None:
        """Append one event line (fsync'd) after repairing any torn tail."""
        self._repair_torn_tail()
        with open(self.final_path, "a", encoding="utf-8", newline="\n") as fileobj:
            _write_line(fileobj, event)
 
    def planned_postimage(self, rel_path: str, postimage_sha256: str) -> None:
        """Append the durable ``planned_postimage`` anchor (BEFORE the replace)."""
        self.append_event({"event": EVENT_PLANNED_POSTIMAGE, "rel_path": rel_path,
                           "postimage_sha256": postimage_sha256})
 
    def applied(self, rel_path: str) -> None:
        """Append the ``applied`` event (AFTER a successful replace)."""
        self.append_event({"event": EVENT_APPLIED, "rel_path": rel_path})
 
    def resolve(self, outcome: str) -> None:
        """Append + fsync the terminal ``resolved`` event — the BINDING state
        transition (a resolved journal never refuses, even if never archived)."""
        func = "JournalWriter.resolve"
        self.append_event({"event": EVENT_RESOLVED, "outcome": outcome})
        _dbg("OC_CHAT.MAIN", func, f"resolved ({outcome})", 2)
 
    def archive(self) -> Path:
        """Best-effort tidy move to ``archive/<run-id>.jsonl`` AFTER ``resolve``.
        Returns the archived path on success, else the resolved-in-place path (an
        archive-move failure is L2-logged and harmless)."""
        func = "JournalWriter.archive"
        archive_dir = self.target_dir / _ARCHIVE_DIRNAME
        try:
            archive_dir.mkdir(parents=True, exist_ok=True)
            dest = archive_dir / f"{self.run_id}{_PUBLISHED_SUFFIX}"
            _replace(self.final_path, dest)
            _dbg("OC_CHAT.MAIN", func, "archived", 2)
            return dest
        except OSError:
            _dbg("OC_CHAT.MAIN", func, "archive move failed (harmless; resolved-in-place)", 2)
            return self.final_path
 
 
__all__ = [
    "JOURNAL_SCHEMA_VERSION",
    "EVENT_PLANNED_POSTIMAGE",
    "EVENT_APPLIED",
    "EVENT_RESOLVED",
    "OUTCOME_SUCCESS",
    "OUTCOME_ROLLED_BACK",
    "PreimageRecord",
    "FoldedEntry",
    "ParsedJournal",
    "JournalWriter",
    "build_preimage",
    "header_dict",
    "project_journal_bytes",
    "read_journal",
    "preimage_bytes_of",
    "scan_unresolved",
]