malkhut(T7): UV Clock — event-driven reactor + staleness watchdog
UVClock (clock/host.py): T19 event-driven reactor, asyncio dispatch. Events (clock/events.py): Scan, Tick, Timer, BarFire, Stale. Staleness watchdog (clock/staleness.py): T19 Staleness Law enforcement. DeadNode reaper (clock/deadnode.py): iox2 orphan sweep on startup.
This commit is contained in:
107
MALKHUT/malkhut/clock/staleness.py
Normal file
107
MALKHUT/malkhut/clock/staleness.py
Normal file
@@ -0,0 +1,107 @@
|
||||
"""
|
||||
T19 Staleness Watchdog — freshness monitor for event sources.
|
||||
|
||||
Per T19 ANNEX A Staleness Law:
|
||||
1. BarFire is EDGE-TRIGGERED on scan_number monotonic advance — NEVER timer-synthesized.
|
||||
2. Every event carries provenance: scan_number, scan_ts, ingest_ts, AGE at dispatch.
|
||||
3. Freshness watchdog TimerEvent (now - last_scan_ts > 1.5x cadence) emits StaleInput.
|
||||
|
||||
This is the June-22 scan-freeze / silent-scanner-death detector, native.
|
||||
Fixes BLUE's oldest observability hole as a side effect.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import time
|
||||
from typing import Callable, Optional
|
||||
|
||||
from malkhut.clock.events import EventProvenance, StaleInput, TimerEvent
|
||||
|
||||
|
||||
class StalenessWatchdog:
|
||||
"""
|
||||
Monitors event source freshness. Emits StaleInput when stale.
|
||||
|
||||
Usage:
|
||||
watchdog = StalenessWatchdog(cadence_ns=5_850_000_000)
|
||||
# On each event:
|
||||
watchdog.heartbeat(provenance)
|
||||
# Periodically (via TimerEvent):
|
||||
stale = watchdog.check()
|
||||
if stale:
|
||||
# TUI red + posture response
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
cadence_ns: int = 5_850_000_000, # 5.85s default
|
||||
stale_factor: float = 1.5,
|
||||
on_stale: Optional[Callable[[StaleInput], None]] = None,
|
||||
) -> None:
|
||||
self._cadence_ns = cadence_ns
|
||||
self._stale_factor = stale_factor
|
||||
self._stale_threshold_ns = int(cadence_ns * stale_factor)
|
||||
self._on_stale = on_stale
|
||||
|
||||
self._last_scan_ts: int = 0
|
||||
self._last_scan_number: int = -1
|
||||
self._is_stale: bool = False
|
||||
self._stale_count: int = 0
|
||||
|
||||
def heartbeat(self, provenance: EventProvenance) -> None:
|
||||
"""Record a fresh event. Resets staleness."""
|
||||
now = time.time_ns()
|
||||
self._last_scan_ts = provenance.scan_ts
|
||||
self._last_scan_number = provenance.scan_number
|
||||
|
||||
if self._is_stale:
|
||||
# Recovery from stale state
|
||||
self._is_stale = False
|
||||
|
||||
def check(self) -> Optional[StaleInput]:
|
||||
"""
|
||||
Check if source is stale. Returns StaleInput if stale, None if fresh.
|
||||
|
||||
Call this from a TimerEvent handler at the watchdog cadence.
|
||||
"""
|
||||
now = time.time_ns()
|
||||
age = now - self._last_scan_ts if self._last_scan_ts > 0 else now
|
||||
|
||||
if age > self._stale_threshold_ns and not self._is_stale:
|
||||
self._is_stale = True
|
||||
self._stale_count += 1
|
||||
|
||||
stale = StaleInput(
|
||||
source="scan_source",
|
||||
last_scan_ts=self._last_scan_ts,
|
||||
current_ts=now,
|
||||
stale_duration_ns=age - self._stale_threshold_ns,
|
||||
)
|
||||
|
||||
if self._on_stale:
|
||||
self._on_stale(stale)
|
||||
|
||||
return stale
|
||||
|
||||
return None
|
||||
|
||||
@property
|
||||
def is_stale(self) -> bool:
|
||||
return self._is_stale
|
||||
|
||||
@property
|
||||
def stale_count(self) -> int:
|
||||
return self._stale_count
|
||||
|
||||
@property
|
||||
def last_scan_ts(self) -> int:
|
||||
return self._last_scan_ts
|
||||
|
||||
@property
|
||||
def last_scan_number(self) -> int:
|
||||
return self._last_scan_number
|
||||
|
||||
@property
|
||||
def age_ns(self) -> int:
|
||||
if self._last_scan_ts == 0:
|
||||
return time.time_ns()
|
||||
return time.time_ns() - self._last_scan_ts
|
||||
Reference in New Issue
Block a user