117 lines
3.3 KiB
Python
117 lines
3.3 KiB
Python
"""State dir layout, atomic JSON files and the append-only ledger."""
|
|
import contextlib
|
|
import json
|
|
import os
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
SUBDIRS = ("bases", "jobs", "raw", "notes", "logs", "briefs")
|
|
RETRY_TRIES = 40
|
|
RETRY_SLEEP_S = 0.05
|
|
|
|
_ledger_lock = threading.Lock()
|
|
|
|
|
|
def state_dir(run) -> Path:
|
|
"""`<x>/run.md` -> `<x>/run-gremlins` (a sibling of the run log)."""
|
|
p = os.path.abspath(str(run))
|
|
if not p.lower().endswith(".md"):
|
|
raise ValueError("--run must name the run log .md file: %s" % p)
|
|
return Path(p[:-3] + "-gremlins")
|
|
|
|
|
|
def init_state(run) -> Path:
|
|
"""Create the state dir and its subdirs (idempotent)."""
|
|
state = state_dir(run)
|
|
for d in SUBDIRS:
|
|
(state / d).mkdir(parents=True, exist_ok=True)
|
|
return state
|
|
|
|
|
|
def _retry(fn, tries: int = RETRY_TRIES, sleep_s: float = RETRY_SLEEP_S):
|
|
"""Call fn(); retry PermissionError (Windows sharing violations)."""
|
|
for attempt in range(1, tries + 1):
|
|
try:
|
|
return fn()
|
|
except PermissionError:
|
|
if attempt == tries:
|
|
raise
|
|
time.sleep(sleep_s)
|
|
|
|
|
|
def atomic_write_bytes(path, data: bytes) -> None:
|
|
"""Write to a temp file in the same dir, then `os.replace` with retries."""
|
|
path = Path(path)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
tmp = path.parent / (".%s.%d.%s.tmp" % (path.name, os.getpid(), uuid.uuid4().hex[:8]))
|
|
with open(tmp, "wb") as f:
|
|
f.write(data)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
try:
|
|
_retry(lambda: os.replace(tmp, path))
|
|
except BaseException:
|
|
with contextlib.suppress(OSError):
|
|
os.remove(tmp)
|
|
raise
|
|
|
|
|
|
def read_bytes_retry(path) -> bytes:
|
|
def _read():
|
|
with open(path, "rb") as f:
|
|
return f.read()
|
|
return _retry(_read)
|
|
|
|
|
|
def read_json(p) -> dict | None:
|
|
"""The parsed file, or None if it does not exist. Corrupt content raises ValueError."""
|
|
try:
|
|
data = read_bytes_retry(p)
|
|
except FileNotFoundError:
|
|
return None
|
|
return json.loads(data.decode("utf-8"))
|
|
|
|
|
|
def write_json(p, obj) -> None:
|
|
atomic_write_bytes(p, (json.dumps(obj, indent=1, sort_keys=True) + "\n").encode("utf-8"))
|
|
|
|
|
|
def _ledger_path(run) -> Path:
|
|
return state_dir(run) / "ledger.jsonl"
|
|
|
|
|
|
def ledger_append(run, line: dict) -> None:
|
|
"""Append one JSON line with a single write; repairs a torn last line first."""
|
|
path = _ledger_path(run)
|
|
body = (json.dumps(line, sort_keys=True) + "\n").encode("utf-8")
|
|
|
|
def _append():
|
|
torn = False
|
|
with contextlib.suppress(FileNotFoundError), open(path, "rb") as r:
|
|
if r.seek(0, os.SEEK_END) > 0:
|
|
r.seek(-1, os.SEEK_END)
|
|
torn = r.read(1) != b"\n"
|
|
with open(path, "ab") as f:
|
|
f.write((b"\n" if torn else b"") + body)
|
|
with _ledger_lock:
|
|
_retry(_append)
|
|
|
|
|
|
def ledger_lines(run) -> list[dict]:
|
|
"""Every parsable line of the ledger; [] when there is none yet."""
|
|
try:
|
|
data = read_bytes_retry(_ledger_path(run))
|
|
except FileNotFoundError:
|
|
return []
|
|
out = []
|
|
for ln in data.splitlines():
|
|
try:
|
|
obj = json.loads(ln.decode("utf-8", "replace"))
|
|
except ValueError:
|
|
continue
|
|
if isinstance(obj, dict):
|
|
out.append(obj)
|
|
return out
|