"""On-disk delivery journal: append-only JSON-lines. Journal(path: str | None) path=None -> in-memory only; no file written. Never raises on corrupt file. journal.record_produced(signal, *, now_ms) -> str Appends a 'produced' line BEFORE the POST. Returns entry id (uuid4 hex). If the asset already has a pending (undelivered) entry, drops the older one with a 'dropped_superseded' outcome line and replaces it. journal.record_outcome(entry_id, outcome, **fields) -> None Appends a terminal 'outcome' line. Marks the id settled. journal.pending() -> list In-memory queue: newest undelivered per asset. Each entry exposes .id, .asset, .signal, .valid_until (attribute access). journal.replay(now_ms) -> None Re-read the file, rebuild the settled set, re-queue valid+unsettled+newest per-asset entries. Writes drop lines for stale/superseded. Skips corrupt lines (logs journal_corrupt_line). Never raises. journal.is_settled(entry_id) -> bool """ from __future__ import annotations import json import os import sys import uuid from dataclasses import dataclass from typing import Any def _log(event: str, **fields: Any) -> None: import time record = {"event": event, "ts": time.time(), **fields} print(f"[scaffold] {json.dumps(record)}", file=sys.stderr, flush=True) @dataclass class _Entry: id: str asset: str signal: dict valid_until: int class Journal: """Append-only JSONL delivery journal, in-memory or on-disk.""" def __init__(self, path: str | None) -> None: self._path = path self._settled: set[str] = set() # asset -> _Entry (newest undelivered per asset) self._queue: dict[str, _Entry] = {} # ------------------------------------------------------------------ # Write helpers # ------------------------------------------------------------------ def _append(self, record: dict) -> None: if self._path is None: return line = json.dumps(record) + "\n" try: with open(self._path, "a", encoding="utf-8") as fh: fh.write(line) except OSError as exc: # Append failed (disk full, read-only fs, permissions, I/O error). # We do NOT let this tear down the delivery sink: the safe direction # is a dropped/lost signal (fail toward NOT trading), not a crash of # the sink that would take down delivery for every other candidate. # Log explicitly so the at-risk candidate is observable. # TODO(observability): journal_append_failed must be captured by the # observability stack and alerted on — a persistent append failure # means signals are being silently dropped on this scanner. _log( "journal_append_failed", path=self._path, error=str(exc), kind=record.get("kind"), entry_id=record.get("id"), asset=record.get("asset"), ) # ------------------------------------------------------------------ # Public API # ------------------------------------------------------------------ def record_produced(self, signal: dict, *, now_ms: int) -> str: """Write a 'produced' line and enqueue the signal. Returns entry id. The entry id IS the wire `signal_id`: the journal stamps `signal_id` onto the persisted/queued signal so the POSTed signal carries the dedup key, and a replay re-uses the SAME id (no fresh id minted on restart). If the signal already carries a `signal_id` (the delivery sink threaded the eid in via the envelope), that id is re-used as the entry id; otherwise one is minted. """ existing = signal.get("signal_id") entry_id = existing if isinstance(existing, str) and existing else uuid.uuid4().hex # Stamp the id onto the wire signal so the persisted produced line and the # POSTed body both carry signal_id == entry id. signal["signal_id"] = entry_id asset = signal.get("asset", "") valid_until = signal.get("valid_until", 0) # If there's already a pending entry for this asset, drop the old one. if asset in self._queue: old = self._queue[asset] self._append({ "kind": "outcome", "id": old.id, "outcome": "dropped_superseded", }) self._settled.add(old.id) record = { "kind": "produced", "id": entry_id, "asset": asset, "produced_at": signal.get("produced_at", now_ms), "valid_until": valid_until, "signal": signal, } self._append(record) entry = _Entry(id=entry_id, asset=asset, signal=signal, valid_until=valid_until) self._queue[asset] = entry return entry_id def record_outcome(self, entry_id: str, outcome: str, **fields) -> None: """Write a terminal outcome line and mark the id settled.""" record = {"kind": "outcome", "id": entry_id, "outcome": outcome, **fields} self._append(record) self._settled.add(entry_id) # Remove from in-memory queue to_remove = [a for a, e in self._queue.items() if e.id == entry_id] for a in to_remove: del self._queue[a] def pending(self) -> list: """Return list of pending _Entry objects (newest undelivered per asset).""" return list(self._queue.values()) def is_settled(self, entry_id: str) -> bool: return entry_id in self._settled def replay(self, now_ms: int) -> None: """Re-read the journal file and rebuild the in-memory state. Re-queues entries that are: - not settled - still valid (now_ms < valid_until; boundary == is expired) - newest per asset (by produced_at) Writes drop lines for stale/superseded unsettled entries. Skips corrupt lines (logs journal_corrupt_line). Never raises. """ # Reset in-memory state before replay self._settled = set() self._queue = {} if self._path is None or not os.path.exists(self._path): return produced: dict[str, dict] = {} # id -> produced record settled: set[str] = set() # --- Pass 1: parse all lines --- try: with open(self._path, "r", encoding="utf-8", errors="replace") as fh: lines = fh.readlines() except OSError: return for line in lines: line = line.strip() if not line: continue try: obj = json.loads(line) except (json.JSONDecodeError, ValueError, UnicodeDecodeError): _log("journal_corrupt_line", path=self._path) continue if not isinstance(obj, dict): _log("journal_corrupt_line", path=self._path) continue kind = obj.get("kind") if kind == "produced": eid = obj.get("id") if eid: produced[eid] = obj elif kind == "outcome": eid = obj.get("id") if eid: settled.add(eid) self._settled = settled # --- Pass 2: find newest per asset among unsettled produced --- # unsettled produced: id -> record unsettled = {eid: rec for eid, rec in produced.items() if eid not in settled} # Build newest-per-asset map: asset -> (produced_at, id, record) newest: dict[str, tuple] = {} # asset -> (produced_at, id, record) for eid, rec in unsettled.items(): asset = rec.get("asset", "") pat = rec.get("produced_at", 0) # Guard: non-numeric produced_at cannot be compared (would TypeError). # Treat it as 0 so numeric entries always win the newest comparison; # the entry will still be candidate-winner only if it is the sole one # for its asset — and it will then be dropped at the valid_until check. if not isinstance(pat, (int, float)) or isinstance(pat, bool): pat = 0 if asset not in newest or pat > newest[asset][0]: newest[asset] = (pat, eid, rec) # For each asset, the winner is re-queued if still valid; losers get dropped. winners = {info[1] for info in newest.values()} # Entries that are unsettled but NOT the newest for their asset -> dropped_superseded for eid, rec in unsettled.items(): if eid not in winners: self._append({"kind": "outcome", "id": eid, "outcome": "dropped_superseded"}) self._settled.add(eid) # Re-queue winners: valid -> queue, expired or non-numeric -> dropped_expired for asset, (pat, eid, rec) in newest.items(): valid_until = rec.get("valid_until", 0) if not isinstance(valid_until, (int, float)) or isinstance(valid_until, bool): # Non-numeric valid_until from a corrupt/poisoned persisted line. # Drop it — do not attempt comparison (would TypeError). _log("journal_corrupt_line", path=self._path, reason="valid_until_not_numeric", entry_id=eid) self._append({"kind": "outcome", "id": eid, "outcome": "dropped_expired"}) self._settled.add(eid) continue if now_ms < valid_until: signal = rec.get("signal", {}) entry = _Entry(id=eid, asset=asset, signal=signal, valid_until=valid_until) self._queue[asset] = entry else: self._append({"kind": "outcome", "id": eid, "outcome": "dropped_expired"}) self._settled.add(eid)