stroblmeandClaude Opus 5
60d7ec81c0
A wheel whose top-level module is `app` collides with anything else in a user's venv, so the package that is about to be published takes the name it is published under. Only the Python package moves; the repo, the Docker WORKDIR and the compose project keep theirs. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
84 lines
2.8 KiB
Python
84 lines
2.8 KiB
Python
"""Engine self-observation: event-loop lag watchdog and the deliberate exit.
|
|
|
|
Docker's restart policy only fires when the process exits, so a wedged event
|
|
loop would otherwise stay "up" forever. The watchdog measures how late a
|
|
sleeping task wakes up — the standard loop-lag trick — and feeds the deep
|
|
health endpoint; `engine_fatal` is the deliberate handoff to the outer
|
|
supervisor when a clean restart beats limping on.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
import os
|
|
import time
|
|
from collections import deque
|
|
|
|
from fluksio.flow.events import EventBus
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
INTERVAL = 1.0
|
|
EWMA_ALPHA = 0.2
|
|
# One late wake-up is a busy moment; several in a row is a blocked loop.
|
|
DEGRADED_LAG_S = 5.0
|
|
DEGRADED_STRIKES = 3
|
|
# Sustained lag above this marks the engine degraded in /health.
|
|
DEGRADED_EWMA_MS = 200.0
|
|
|
|
|
|
class LoopWatchdog:
|
|
"""Measures event-loop lag and reports it as engine health."""
|
|
|
|
def __init__(self, events: EventBus | None = None) -> None:
|
|
self._events = events
|
|
self.ewma_ms = 0.0
|
|
self._window: deque[tuple[float, float]] = deque() # (monotonic ts, lag ms)
|
|
self._strikes = 0
|
|
|
|
async def run(self) -> None:
|
|
while True:
|
|
before = time.monotonic()
|
|
await asyncio.sleep(INTERVAL)
|
|
self._record(max(0.0, time.monotonic() - before - INTERVAL))
|
|
|
|
def _record(self, lag_s: float) -> None:
|
|
lag_ms = lag_s * 1000.0
|
|
self.ewma_ms += EWMA_ALPHA * (lag_ms - self.ewma_ms)
|
|
now = time.monotonic()
|
|
self._window.append((now, lag_ms))
|
|
while self._window and self._window[0][0] < now - 60.0:
|
|
self._window.popleft()
|
|
if lag_s >= DEGRADED_LAG_S:
|
|
self._strikes += 1
|
|
if self._strikes == DEGRADED_STRIKES and self._events is not None:
|
|
logger.warning("event loop lagging: %.1fs late", lag_s)
|
|
self._events.publish(
|
|
{
|
|
"type": "engine_degraded",
|
|
"reason": f"event loop lag {lag_s:.1f}s",
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
else:
|
|
self._strikes = 0
|
|
|
|
@property
|
|
def degraded(self) -> bool:
|
|
return self.ewma_ms > DEGRADED_EWMA_MS
|
|
|
|
def snapshot(self) -> dict[str, float]:
|
|
return {
|
|
"ewma": round(self.ewma_ms, 1),
|
|
"max_60s": round(max((lag for _, lag in self._window), default=0.0), 1),
|
|
}
|
|
|
|
|
|
def engine_fatal(reason: str, events: EventBus | None = None) -> None:
|
|
"""Log, tell whoever still listens, and exit so Docker restarts us clean."""
|
|
logger.critical("engine fatal: %s", reason)
|
|
if events is not None:
|
|
events.publish({"type": "engine_fatal", "reason": reason, "ts": time.time()})
|
|
os._exit(1)
|