hms-mediaengine/packages/cluster/hms_cluster/registry.py

"""Node-Registry mit Heartbeat-Zuständen (PLAN.md §6.3, §6.5).

- Zustände je Node: online / degraded / stale / offline mit konfigurierbaren
  Schwellen (§6.5)
- doppelte node_id wird als Fehler blockiert, nie still übernommen (§6.3)
- Kategorien für die UI: discovered / paired / unknown / incompatible /
  offline werden getrennt geführt (§6.3)
- persistente node_id bleibt identisch bei IP-Wechsel; Endpunkte werden
  als „zuletzt bekannt" aktualisiert (§10.1 Node)
"""

from __future__ import annotations

import time
from dataclasses import dataclass, field
from enum import StrEnum


class NodeHealth(StrEnum):
    ONLINE = "online"
    DEGRADED = "degraded"
    STALE = "stale"
    OFFLINE = "offline"


class NodeCategory(StrEnum):
    """UI-Kategorien gemäß §6.3: gefunden, gepaart, unbekannt, inkompatibel,
    offline werden getrennt aufgeführt."""

    DISCOVERED = "discovered"
    PAIRED = "paired"
    UNKNOWN = "unknown"
    INCOMPATIBLE = "incompatible"
    OFFLINE = "offline"


class DuplicateNodeError(Exception):
    """Doppelte node_id – wird als Fehler blockiert (§6.3)."""


@dataclass
class NodeEntry:
    """Registry-Eintrag: Identität stabil, Endpunkte „zuletzt bekannt" (§10.1)."""

    node_id: str
    display_name: str
    roles: tuple[str, ...] = ()
    api_port: int = 0
    protocol_version: int = 1
    capability_digest: str = ""
    last_known_endpoints: list[str] = field(default_factory=list)
    last_heartbeat_ns: int = 0
    health: NodeHealth = NodeHealth.OFFLINE
    category: NodeCategory = NodeCategory.DISCOVERED
    clock_offset_ns: int = 0

    def record_endpoint(self, endpoint: str) -> None:
        """IP-Wechsel: node_id bleibt, Endpunkt wird aktualisiert (§6.3)."""
        if endpoint in self.last_known_endpoints:
            self.last_known_endpoints.remove(endpoint)
        self.last_known_endpoints.insert(0, endpoint)
        del self.last_known_endpoints[4:]  # die letzten 5 genügen


@dataclass
class HealthThresholds:
    """Konfigurierbare Schwellen je Zustand (§6.5).

    Heartbeat pünktlich < degraded_after_ns; verspätet, aber vorhanden
    < stale_after_ns; danach offline. Standard-Heartbeat 500 ms (§6.5).
    """

    heartbeat_interval_ns: int = 500_000_000
    degraded_after_ns: int = 2_000_000_000  # 2 s ohne Heartbeat
    stale_after_ns: int = 5_000_000_000  # 5 s ohne Heartbeat


class NodeRegistry:
    """Autoritative Liste bekannter Nodes (Coordinator-seitig)."""

    def __init__(
        self,
        thresholds: HealthThresholds | None = None,
        protocol_version: int = 1,
    ) -> None:
        self._nodes: dict[str, NodeEntry] = {}
        self._thresholds = thresholds or HealthThresholds()
        self._protocol_version = protocol_version

    def register(
        self,
        node_id: str,
        display_name: str,
        roles: tuple[str, ...] = (),
        api_port: int = 0,
        protocol_version: int = 1,
        capability_digest: str = "",
        endpoint: str = "",
    ) -> NodeEntry:
        """Neue Node oder Update bekannter Node; Doppel-ID mit
        widersprüchlicher Identität ist ein Fehler (§6.3)."""
        existing = self._nodes.get(node_id)
        if existing is not None and existing.display_name != display_name:
            raise DuplicateNodeError(
                f"node_id {node_id} bereits als {existing.display_name!r} registriert"
            )
        if existing is None:
            entry = NodeEntry(
                node_id=node_id,
                display_name=display_name,
                roles=tuple(roles),
                api_port=api_port,
                protocol_version=protocol_version,
                capability_digest=capability_digest,
            )
            self._nodes[node_id] = entry
        else:
            entry = existing
            entry.roles = tuple(roles)
            entry.api_port = api_port
            entry.capability_digest = capability_digest
        if endpoint:
            entry.record_endpoint(endpoint)
        # Inkompatible Protokollversion sichtbar kategorisieren (§6.3)
        if entry.protocol_version != self._protocol_version:
            entry.category = NodeCategory.INCOMPATIBLE
        return entry

    def record_heartbeat(self, node_id: str, clock_offset_ns: int = 0) -> None:
        entry = self._nodes.get(node_id)
        if entry is None:
            raise KeyError(f"unknown node {node_id}")
        entry.last_heartbeat_ns = time.monotonic_ns()
        entry.clock_offset_ns = clock_offset_ns

    def evaluate_health(self, node_id: str) -> NodeHealth:
        """Berechnet den Zustand aus letztem Heartbeat + Schwellen (§6.5)."""
        entry = self._nodes[node_id]
        if entry.last_heartbeat_ns == 0:
            entry.health = NodeHealth.OFFLINE
            if entry.category not in (NodeCategory.INCOMPATIBLE, NodeCategory.PAIRED):
                entry.category = NodeCategory.DISCOVERED
            return entry.health
        elapsed = time.monotonic_ns() - entry.last_heartbeat_ns
        if elapsed < self._thresholds.degraded_after_ns:
            entry.health = NodeHealth.ONLINE
        elif elapsed < self._thresholds.stale_after_ns:
            entry.health = NodeHealth.DEGRADED
        else:
            entry.health = NodeHealth.OFFLINE
            if entry.category is NodeCategory.PAIRED:
                entry.category = NodeCategory.OFFLINE  # Vertrauen bleibt, nur weg
        return entry.health

    def mark_paired(self, node_id: str) -> None:
        self._nodes[node_id].category = NodeCategory.PAIRED

    def mark_unknown(self, node_id: str) -> None:
        self._nodes[node_id].category = NodeCategory.UNKNOWN

    def get(self, node_id: str) -> NodeEntry | None:
        return self._nodes.get(node_id)

    def by_category(self, category: NodeCategory) -> list[NodeEntry]:
        return [n for n in self._nodes.values() if n.category is category]

    def all(self) -> list[NodeEntry]:
        return list(self._nodes.values())