ledger.py

Go to the documentation of this file.

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
# oct/timing/ledger.py
 
"""
Purpose
-------
Persist stage timings.
 
Responsibilities
----------------
- Keep local timing data bounded, private and advisory.
 
Diagnostics
-----------
Domain: GENERAL
Levels:
    L1 -- instrumentation failures
    L2 -- lifecycle
    L3 -- semantic details
    L4 -- deep tracing
 
Contracts
---------
- Timing instrumentation never changes the wrapped operation's outcome.
- Only the fixed timing schema is persisted; command arguments are never stored.
 
Dependencies
------------
- oc_diagnostics (structured timing diagnostics)
- oct (version and timing contracts)
- click (lazy; walks the registered command groups to name the stage)
"""
from __future__ import annotations
 
from contextlib import contextmanager
from dataclasses import asdict, dataclass
from datetime import datetime, timezone
import json
import os
from pathlib import Path
import re
import socket
import time
from typing import Iterator
from oc_diagnostics import _dbg, dbg_timed
from oct import __version__
 
@dataclass
class TimingSample:
    """Mutable measurements with a fixed, argument-free serialized shape."""
    ts: str
    host: str
    cpus: int
    stage: str
    kind: str = "stage"
    wall_ms: int = 0
    cpu_ms: int = 0
    files: int | None = None
    scope: str | None = None
    outcome: int | str = "ok"
    version: str = __version__
 
def _failure() -> None:
    """Report instrumentation failure without exposing paths or caller data."""
    func = "_failure"
    try:
        _dbg("GENERAL", func, "SYSTEM_ERROR: timing ledger unavailable", 1)
    except Exception:
        # Diagnostics itself may be unavailable; the operation remains primary.
        return
 
def _token(value: object) -> str | None:
    """Accept short tokens only, never free text or paths."""
    return value if isinstance(value, str) and re.fullmatch(r"[a-zA-Z0-9_.-]{1,64}", value) else None
 
def _append(root: Path, sample: TimingSample) -> None:
    """Serialize one bounded record under a nonblocking interprocess lock."""
    func = "_append"
    path = root / ".option_c" / "logs" / "timings.jsonl"
    path.parent.mkdir(parents=True, exist_ok=True)
    with (path.parent / "timings.lock").open("a+b") as lock:
        lock.seek(0, 2)
        if lock.tell() == 0:
            lock.write(b"0")
            lock.flush()
        lock.seek(0)
        if os.name == "nt":
            import msvcrt
            msvcrt.locking(lock.fileno(), msvcrt.LK_NBLCK, 1)
        else:
            import fcntl
            fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        try:
            data = asdict(sample)
            data["stage"] = _token(sample.stage) or "unknown"
            data["scope"] = _token(sample.scope)
            data["kind"] = sample.kind if sample.kind in ("stage", "wait") else "stage"
            data["outcome"] = sample.outcome if type(sample.outcome) is int or sample.outcome in ("ok", "error") else "error"
            data["files"] = sample.files if type(sample.files) is int and sample.files >= 0 else None
            line = json.dumps(data, separators=(",", ":")) + "\n"
            with path.open("a+", encoding="utf-8", newline="\n") as stream:
                stream.seek(0)
                lines = stream.readlines()
                if len(lines) > 5000:
                    stream.seek(0)
                    stream.truncate()
                    stream.writelines((lines + [line])[-4000:])
                else:
                    if lines and not lines[-1].endswith("\n"):
                        stream.write("\n")
                    stream.write(line)
        finally:
            lock.seek(0)
            if os.name == "nt":
                msvcrt.locking(lock.fileno(), msvcrt.LK_UNLCK, 1)
            else:
                fcntl.flock(lock.fileno(), fcntl.LOCK_UN)
 
def read_records(root: Path) -> list[dict]:
    """Read complete JSON records, ignoring damaged or incomplete lines."""
    func = "read_records"
    path = Path(root) / ".option_c" / "logs" / "timings.jsonl"
    try:
        with path.open(encoding="utf-8") as stream:
            rows = []
            for line in stream:
                try:
                    row = json.loads(line)
                except ValueError:
                    _dbg("GENERAL", func, "VALIDATION_ERROR: invalid timing row", 3)
                    continue
                if isinstance(row, dict):
                    rows.append(row)
            return rows
    except FileNotFoundError:
        return []
    except (OSError, UnicodeError):
        _failure()
        return []
 
@contextmanager
def timed_stage(root: Path, stage: str, *, kind: str = "stage", scope: str | None = None, files: int | None = None, outcome: int | str = "ok") -> Iterator[TimingSample]:
    """Measure one operation and preserve all success and exception semantics."""
    func = "timed_stage"
    sample = TimingSample(datetime.now(timezone.utc).isoformat(), socket.gethostname(), os.cpu_count() or 1, _token(stage) or "unknown", kind=kind, scope=scope, files=files, outcome=outcome)
    # _depth=3: past this generator and contextlib's frame, so the ENTER and
    # EXIT records are located at the code running the stage, not here.
    # oc_diagnostics < 2.2.0 has no _depth (and located them in contextlib).
    try:
        diagnostic = dbg_timed("GENERAL", func, label=sample.stage, _depth=3)
    except TypeError:
        diagnostic = dbg_timed("GENERAL", func, label=sample.stage)
    active = False
    failed = False
    try:
        diagnostic.__enter__()
        active = True
    except Exception:
        failed = True
    started = time.perf_counter_ns()
    cpu = time.process_time_ns()
    try:
        yield sample
    except BaseException as exc:
        code = getattr(exc, "exit_code", getattr(exc, "code", None))
        if isinstance(exc, SystemExit) and code is None:
            code = 0
        sample.outcome = (code if type(code) is int else 1) if type(outcome) is int else "error"
        raise
    finally:
        sample.wall_ms = max(0, (time.perf_counter_ns() - started) // 1_000_000)
        sample.cpu_ms = max(0, (time.process_time_ns() - cpu) // 1_000_000)
        if active:
            try:
                diagnostic.__exit__(None, None, None)
            except Exception:
                failed = True
        try:
            _append(Path(root), sample)
        except Exception:
            failed = True
        if failed:
            _failure()
 
def command_stage(ctx: object) -> str:
    """Use registered command names only; never serialize arbitrary argv."""
    import click
    args = list(ctx._protected_args) + list(ctx.args)
    command = ctx.command
    parts = []
    while args and isinstance(command, click.Group):
        child = command.get_command(ctx, args[0])
        if child is None:
            break
        parts.append(child.name)
        args.pop(0)
        command = child
    return ".".join(parts) or "oct"