"""Journal-backed delivery sink factory. build_delivery_sink( *, scanner_id: str, intake_url: str, journal_path: str | None, signal_data_schema: dict | None, # per-data-key {type, required?}; None = no validation default_signal_validity_seconds: int, # fallback signal TTL (seconds) timeout: float = 5.0, clock: Callable[[], float] = time.time, ) -> sink The returned sink is a callable: sink(candidates, tick_id=None, tick=None) `tick` is the scaffold's per-tick observability facts; the sink stamps the journal's pending count onto them (it owns the journal) and rides them on the POST it already makes. On each call it: 1. Replays journal on first call (if file-backed), re-queuing valid undelivered entries 2. Flushes still-valid undelivered entries from prior ticks (retry) 3. Wraps this tick's candidates into wire signals (skips invalid ones), stamping the producing tick's id on each 4. Drops locally superseded / expired entries 5. POSTs {scanner_id, signals:[...]} to /signals 6. Interprets the response: accepted/expired/superseded/rejected = delivered; transport failure or non-success envelope leaves entry queued for retry sink.post_error(error: dict, tick: dict | None = None) -> None POSTs {scanner_id, error} to /errors. Fire-and-forget. A failed tick's facts ride here for the same reason they ride /signals on a clean one: this is the POST that tick already makes. Module constants: SIGNALS_PATH = "/signals" ERRORS_PATH = "/errors" """ from __future__ import annotations import sys import time import uuid from typing import Any, Callable from .http_post import ERRORS_PATH, SIGNALS_PATH, TransportError, post_json from .journal import Journal from .loop_primitives import log_event from .signal_envelope import candidate_to_signal # Terminal statuses from the intake — all mean "the runtime decided, no retry" _TERMINAL_STATUSES = {"accepted", "expired", "superseded", "rejected"} def build_delivery_sink( *, scanner_id: str, intake_url: str, journal_path: str | None, signal_data_schema: dict | None, default_signal_validity_seconds: int, timeout: float = 5.0, clock: Callable[[], float] = time.time, ): """Factory: build and return the delivery sink callable.""" journal = Journal(journal_path) replayed = [False] # mutable cell; replay runs once on first invocation # Mutable cell; the id of the tick currently inside _sink, so the reject/warn # callbacks — which are built once, here — can name it. _sink sets it on entry, # before anything can read it, and both callbacks run only from inside _sink. # None when the caller passed no id, or one the wire would refuse. current_tick_id: list[str | None] = [None] # Distinct dropped-field shapes already warned about (rate-limit: a scanner that # sets a stray key every tick warns ONCE per shape, not every tick). warned_dropped_shapes: set[frozenset] = set() # Distinct unusable tick-id shapes already warned about, same rate-limit reason: a loop # handing the sink the same bad id every tick warns ONCE, not on every tick forever. warned_unusable_tick_ids: set[str] = set() # Set once per process, the first time the intake refuses a batch for carrying tick_id. # From then on the field is left off the wire and the "this intake predates the field" line # is not written again. Never cleared: an intake that refuses the field once would refuse it # again, so re-arming would buy back the guaranteed-400 round-trip on every later tick. intake_predates_tick_id = [False] # The same latch for the body-level `tick` facts. Separate from the one above because # they are separate fields on separate levels of the body: an intake may know one and # not the other, and stripping both on either refusal would throw away what still works. intake_predates_tick_facts = [False] # ------------------------------------------------------------------ # Internal helpers # ------------------------------------------------------------------ def _tick_field(tick_id: Any) -> dict: """`tick_id=...` as log kwargs, or nothing at all when no tick is nameable. An absent key is the honest answer; a null or a blank would join to nothing while looking like an id. """ return {"tick_id": tick_id} if isinstance(tick_id, str) and tick_id else {} def _signal_tick_id(signal: Any) -> Any: """The producing tick's id off a stored signal, or None. A journal line is arbitrary text on disk: `signal` may be a null, a string or a list, so this never assumes a mapping. """ return signal.get("tick_id") if isinstance(signal, dict) else None def _now_ms() -> int: return int(clock() * 1000) def _flush_prior_pending(now_ms: int) -> list[tuple]: """Return still-valid pending entries; drop expired ones with journal line. Returns list of (_Entry, produced_id) to include in the next POST. """ to_send = [] to_drop = [] for entry in list(journal.pending()): # Belt-and-braces: guard against a non-numeric valid_until that survived # into the in-memory queue (e.g. re-queued by a future code path or a # replay edge case). Drop it instead of crashing with TypeError. if not isinstance(entry.valid_until, (int, float)) or isinstance(entry.valid_until, bool): to_drop.append(entry) elif now_ms < entry.valid_until: to_send.append(entry) else: to_drop.append(entry) for entry in to_drop: journal.record_outcome(entry.id, "dropped_expired") # The producing tick's id, read off the journalled signal — the tick that # noticed the expiry did not cause it. log_event( "delivery_dropped_expired", entry_id=entry.id, asset=entry.asset, **_tick_field(_signal_tick_id(entry.signal)), ) return to_send def _usable_tick_id(tick_id: Any) -> str | None: """The tick id to stamp, or None to leave the field off the wire entirely. The intake requires a non-empty string and rejects the WHOLE post on one bad field, so an unusable id is dropped here rather than taking the tick's signals down with it. No id at all is the old call shape, not a fault, and is not logged. """ if tick_id is None or (isinstance(tick_id, str) and tick_id): return tick_id shape = repr(tick_id)[:80] if shape in warned_unusable_tick_ids: return None warned_unusable_tick_ids.add(shape) log_event("delivery_tick_id_unusable", scanner_id=scanner_id, rejected_tick_id=shape) return None def _on_reject(reason: dict) -> None: """Loud rejection: log AND POST to /errors (design §4 — never a silent drop).""" # Merged into one dict before expanding: a reject record that happens to carry # its own `tick_id` would otherwise be a duplicate kwarg — a TypeError raised # here would take down the whole tick's delivery. The scaffold's id goes last, # so the tick that actually ran the candidate is the one named. log_event("delivery_candidate_invalid", **{**reason, **_tick_field(current_tick_id[0])}) _post_error({"type": "candidate_rejected", **reason, **_tick_field(current_tick_id[0])}) def _on_warn(info: dict) -> None: """Loud-but-non-fatal: a valid signal still emitted, but author top-level fields were dropped (never lifted to the wire). Rate-limited per distinct dropped-key shape so an always-present stray key doesn't spam every tick.""" shape = frozenset(info.get("dropped", [])) if not shape or shape in warned_dropped_shapes: return warned_dropped_shapes.add(shape) log_event( "candidate_dropped_fields", scanner_id=scanner_id, dropped=info.get("dropped"), asset=info.get("asset"), **_tick_field(current_tick_id[0]), ) def _wrap_candidates(candidates, produced_at_ms: int, tick_id: str | None) -> list[tuple]: """Wrap candidates into (signal, entry_id) pairs, skipping invalid ones. Schema-rejected candidates are NOT silently dropped: candidate_to_signal calls _on_reject, which logs AND POSTs to /errors. """ result = [] for c in candidates: # Mint the stable signal_id (the journal eid) up front and thread it # through the envelope so the POSTed signal carries it. record_produced # re-uses this same id as the entry id, so wire signal_id == journal eid. eid = uuid.uuid4().hex sig = candidate_to_signal( c, produced_at_ms, signal_id=eid, signal_data_schema=signal_data_schema, default_signal_validity_seconds=default_signal_validity_seconds, on_reject=_on_reject, on_warn=_on_warn, ) if sig is None: continue # Stamped before the journal writes the entry, so a retry — later in this process or # after a restart — re-posts the id of the tick that PRODUCED the signal. Skipped # once the intake has proved it predates the field: a stamped signal is refused and # re-posted without it, and stamping again would buy that same 400 every tick. if tick_id is not None and not intake_predates_tick_id[0]: sig["tick_id"] = tick_id entry_id = journal.record_produced(sig, now_ms=produced_at_ms) result.append((sig, entry_id)) return result def _is_shape_reject(status: int, parsed: Any) -> bool: """True when the intake refused the whole POST because it did not know the shape.""" if status != 400 or not isinstance(parsed, dict): return False err = parsed.get("error") return isinstance(err, dict) and err.get("code") == "INVALID_SHAPE" def _without_tick_id(signals: list[dict]) -> list[dict]: """Copies of `signals` with `tick_id` removed. Copies, never edits: the journal keeps each entry's producing tick, so once the intake is upgraded a later retry of the same entry names it again. """ return [ {k: v for k, v in sig.items() if k != "tick_id"} if isinstance(sig, dict) else sig for sig in signals ] def _with_pending(tick: dict | None) -> dict | None: """The tick facts plus the journal's BACKLOG, which only this closure knows. Counted before this tick's own candidates are journalled, so the number is signals from EARLIER ticks that the intake never acknowledged — a scanner producing work that is not arriving. Counting this tick's own signals too would put a 1 on every healthy signal-bearing tick and say nothing. """ if tick is None: return None return {**tick, "journal_pending": len(journal.pending())} def _post_signals(signals: list[dict], tick: dict | None = None) -> tuple[int, Any]: """POST the signals list to the intake, with this tick's facts when it has any.""" body = {"scanner_id": scanner_id, "signals": signals} if tick is not None and not intake_predates_tick_facts[0]: body["tick"] = tick # Log the exact wire shape so a 400 can be matched to the field that broke. if signals: log_event( "delivery_post_signals", scanner_id=scanner_id, count=len(signals), first_signal_keys=sorted(signals[0].keys()), first_signal=signals[0], ) url = intake_url.rstrip("/") + SIGNALS_PATH return post_json(url, body, timeout) def _interpret_response(status: int, parsed: Any, pending_eids: list[str], pending_signals: list) -> bool: """Interpret POST response; journal outcomes. Returns True if successful envelope.""" if status != 200 or parsed is None: if isinstance(parsed, dict): err = parsed.get("error") or {} body_code = err.get("code") if body_code == "UNKNOWN_SCANNER": log_event("delivery_unknown_scanner", scanner_id=scanner_id, status=status) # Surface the intake's actual reject code+message (e.g. INVALID_SHAPE) # instead of burying every non-200 under "unparseable". log_event( "delivery_rejected", scanner_id=scanner_id, status=status, code=body_code, message=err.get("message"), body=repr(parsed)[:300], ) else: log_event("delivery_response_unparseable", status=status, raw=repr(parsed)[:300]) return False if not isinstance(parsed, dict) or not parsed.get("success"): log_event("delivery_response_unparseable", status=status, parsed=repr(parsed)[:200]) return False # success: true -- read per-signal statuses data = parsed.get("data") or {} per_signal = data.get("signals") if isinstance(data, dict) else None if not isinstance(per_signal, list): per_signal = [] # Match per-signal outcomes BY signal_id, NEVER by array position. The wire # signal_id IS the journal entry id, so a pending eid set is the lookup key. # Index-matching would cross-attribute outcomes (asset A's "accepted" # recorded against asset B) under out-of-order or partial responses — # capital-sensitive. Robust to: out-of-order, partial (fewer statuses than # signals), and unknown/missing signal_id (ignored safely, never applied by # position). An unmatched entry stays pending and is retried. pending = set(pending_eids) # eid -> the tick that produced it, for the per-entry lines below. Built only if one of # those lines is actually reached: every POST pays for it otherwise, and the unknown-status # branch that reads it is the rare one. produced_by: dict | None = None for sig_status in per_signal: if not isinstance(sig_status, dict): continue eid = sig_status.get("signal_id") # Missing or unknown signal_id: cannot be matched -> ignore safely. if not isinstance(eid, str) or eid not in pending: log_event("delivery_status_unmatched", signal_id=eid) continue st = sig_status.get("status") if st == "accepted": journal.record_outcome(eid, "delivered", correlationId=sig_status.get("correlationId", "")) elif st == "rejected": journal.record_outcome(eid, "rejected", reason=sig_status.get("reason", "")) elif st in ("expired", "superseded"): journal.record_outcome(eid, st) else: # Unknown status — treat as undelivered (no outcome written) if produced_by is None: produced_by = { e: _signal_tick_id(sig) for e, sig in zip(pending_eids, pending_signals) } log_event( "delivery_unknown_status", entry_id=eid, status=st, **_tick_field(produced_by.get(eid)), ) # Entries with no matching status returned stay pending (retry next tick). return True # ------------------------------------------------------------------ # The sink callable # ------------------------------------------------------------------ def _sink(candidates, tick_id=None, tick=None) -> None: tick_id = _usable_tick_id(tick_id) current_tick_id[0] = tick_id # Replay journal on very first invocation if not replayed[0]: replayed[0] = True journal.replay(_now_ms()) now_ms = _now_ms() produced_at_ms = now_ms # single timestamp for the whole tick # 1. Flush prior pending (retry candidates from older ticks) prior_entries = _flush_prior_pending(now_ms) # Read here and not below: the backlog is what EARLIER ticks left behind, and # step 2 is about to journal this tick's own candidates on top of it. tick_body = _with_pending(tick) # 2. Wrap new candidates into signals new_signal_pairs = _wrap_candidates(candidates, produced_at_ms, tick_id) # Build the ordered signal list and corresponding entry-id list: # prior entries first (retry), then new ones. # But: if a new candidate for the same asset was just produced, the # journal.record_produced already dropped the prior entry (dropped_superseded). # We need to filter out prior entries that are now settled. all_signals: list[dict] = [] all_eids: list[str] = [] for entry in prior_entries: if not journal.is_settled(entry.id): all_signals.append(entry.signal) all_eids.append(entry.id) for sig, eid in new_signal_pairs: if not journal.is_settled(eid): all_signals.append(sig) all_eids.append(eid) # 3. POST (always, even when signals==[]) — which is what makes this the channel # the tick facts ride: a quiet scanner still posts, every tick. try: status, parsed = _post_signals(all_signals, tick_body) except TransportError as exc: log_event("delivery_transport_error", error=str(exc)) # All signals remain undelivered -> retry on next tick # Empty-list no-ops: nothing to retry (no produced lines were written) return # 3b. Forward-compat guard for an intake rolled back to before tick_id. Such an intake # validates with additionalProperties:false, so it 400s the WHOLE batch on the unknown # field — and because the journal re-posts the stored signal untouched, that batch is # re-sent every tick forever, taking every fresh signal batched with it down too. Re-post # the same batch without the field so one unknown key cannot strand real signals, and # latch the discovery so later ticks leave the field off and stop paying the refused POST. # Entries already journalled with an id still reach here and are still stripped per POST. sent_tick_id = any(isinstance(sig, dict) and "tick_id" in sig for sig in all_signals) sent_tick_facts = tick_body is not None and not intake_predates_tick_facts[0] if _is_shape_reject(status, parsed) and (sent_tick_id or sent_tick_facts): if sent_tick_id and not intake_predates_tick_id[0]: intake_predates_tick_id[0] = True log_event("delivery_intake_predates_tick_id", scanner_id=scanner_id) if sent_tick_facts: intake_predates_tick_facts[0] = True log_event("delivery_intake_predates_tick_facts", scanner_id=scanner_id) try: status, parsed = _post_signals(_without_tick_id(all_signals)) except TransportError as exc: log_event("delivery_transport_error", error=str(exc)) return # 4. Interpret response # The response per-signal list corresponds only to all_signals that have # produced entries (all_eids). Empty no-ops: no eids. The signals are the ones the # journal holds — with their tick ids — so a log line still names the producing tick # even when the retry above had to send them without it. _interpret_response(status, parsed, all_eids, all_signals) def _post_error(error: dict, tick: dict | None = None) -> None: """POST an error dict, with the failed tick's facts when the caller has them. Fire-and-forget, with one exception: an intake that refuses the body because it does not know `tick` is retried once without it and latched, so a failed tick's error record is never lost to a field that only telemetry wanted. """ tick_body = _with_pending(tick) url = intake_url.rstrip("/") + ERRORS_PATH # The error belongs to exactly one tick and already names it; a body that # names none leaves the key off rather than borrowing the current one. origin = _tick_field(error.get("tick_id")) if isinstance(error, dict) else {} def _body(with_tick: bool) -> dict: body = {"scanner_id": scanner_id, "error": error} if with_tick and tick_body is not None and not intake_predates_tick_facts[0]: body["tick"] = tick_body return body try: sent_tick_facts = tick_body is not None and not intake_predates_tick_facts[0] status, parsed = post_json(url, _body(True), timeout) if sent_tick_facts and _is_shape_reject(status, parsed): intake_predates_tick_facts[0] = True log_event("delivery_intake_predates_tick_facts", scanner_id=scanner_id) post_json(url, _body(False), timeout) except TransportError as exc: log_event("delivery_error_post_failed", error=str(exc), **origin) except Exception as exc: # noqa: BLE001 log_event("delivery_error_post_failed", error=str(exc), **origin) _sink.post_error = _post_error # type: ignore[attr-defined] return _sink