orchestrata_ultimate.kernel.heartbeat_pump

Reference › orchestrata_ultimate › kernel › heartbeat_pump

The roster heartbeat pump of SPEC-001 (volk-comms-heartbeat), used by cli_main when it boots the monolithic deployment. HeartbeatPump takes a health probe (the seam that offers record_heartbeat, get_volk_health and is_registered), a zero-argument roster callable and an optional tracer, and keeps the registered volks marked alive.

beat_once asks the roster for the current alias list on every sweep, so registration changes show up at once, and calls record_heartbeat on the probe for each alias. It is best-effort per alias: an exception is logged and the sweep continues, and the method returns the number of heartbeats recorded. start launches a background thread and is idempotent; the thread loop waits on a stop event for the interval and then beats; stop sets the event and joins the thread within a timeout. The interval comes from VOLK_HEARTBEAT_INTERVAL_S, read from the environment mapping passed in or from the process environment. The class is traced to REQ-001-017 and REQ-001-018. Status changes caused by missed or restored heartbeats reach the event bus through volk_status_bridge.

At a glance

Class diagram

Dependency diagram

API

Classes

HeartbeatPump

Decorators: traces(spec='001', story='US-001', reqs=['REQ-001-018', 'REQ-001-017']) · Line 36

Beats a heartbeat for every alias in the live roster on a fixed interval.

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/src/orchestrata_ultimate/kernel/heartbeat_pump.py

"""SPEC-001 volk-comms-heartbeat — the roster heartbeat pump (closes G1).

Liveness for in-process ``BaseAgent`` volks was conflated with dispatch
recency: without a pump, an idle-but-alive volk (no dispatch traffic since
its last heartbeat) decays to ``INACTIVE`` after ``heartbeat_timeout_seconds``
even though nothing is actually wrong — there is no separate volk process to
push heartbeats from, so "the hosting runtime is up AND the handler is
registered" IS the correct liveness signal. ``HeartbeatPump`` beats a
heartbeat for every alias in the live roster on a fixed interval, keeping
idle volks healthy. Busy/failed remains ``BaseAgent.status``'s job.

Threads, not asyncio — matches the codebase's existing thread posture under
free-threaded 3.14t (decision 2, plans/volk-comms-heartbeat-design.md §4.1).
"""

from __future__ import annotations

import logging
import os
import threading
from collections.abc import Callable, Mapping

from orchestrata_ultimate.core.base.traces import traces
from orchestrata_ultimate.core.protocols.volk_health import VolkHealthProbe
from orchestrata_ultimate.core.tracing.tracer import Tracer

logger = logging.getLogger(__name__)

# Tunable, not an infra endpoint key — the literal default is permitted
# under RUL-003's inverse rule (only REQUIRED infra endpoint keys must raise
# when unset).
DEFAULT_HEARTBEAT_INTERVAL_S = "15"


@traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-017"])
class HeartbeatPump:
    """Beats a heartbeat for every alias in the live roster on a fixed interval.

    Composed of a :class:`~orchestrata_ultimate.core.protocols.volk_health.VolkHealthProbe`
    (typically ``agents.volk_registry.VolkRegistry``, which satisfies it
    structurally) and a *roster* callable returning the currently-live alias
    list. ``start``/``stop`` run the beat loop on a daemon
    ``threading.Thread``; ``beat_once`` is also callable directly (e.g. from
    a test, or a one-shot sweep).
    """

    def __init__(
        self,
        probe: VolkHealthProbe,
        roster: Callable[[], list[str]],
        tracer: Tracer | None = None,
        env: Mapping[str, str] | None = None,
    ) -> None:
        """Bind the pump to *probe* and *roster*; read the beat interval.

        Args:
            probe: Health seam the pump beats against
                (``record_heartbeat``/``get_volk_health``/``is_registered``).
            roster: Zero-arg callable returning the currently-live alias list;
                consulted fresh on every beat so it reflects registration
                churn.
            tracer: Optional span/log emitter; defaults to the process-wide
                Singleton (:meth:`Tracer.default`).
            env: Optional environment mapping consulted for
                ``VOLK_HEARTBEAT_INTERVAL_S``; defaults to ``os.environ``
                when omitted. This is a tunable, not an infra endpoint key,
                so a literal default is permitted here.
        """
        self._probe = probe
        self._roster = roster
        self._tracer = tracer if tracer is not None else Tracer.default()
        source: Mapping[str, str] = env if env is not None else os.environ
        self._interval_s = float(
            source.get("VOLK_HEARTBEAT_INTERVAL_S", DEFAULT_HEARTBEAT_INTERVAL_S)
        )
        self._thread: threading.Thread | None = None
        self._stop_event = threading.Event()

    @traces(spec="001", story="US-001", reqs=["REQ-001-018", "REQ-001-017"])
    def beat_once(self) -> int:
        """Record one heartbeat for every alias currently in the roster.

        Best-effort per alias (REQ-001-017): a failing or unregistered alias
        is logged and skipped, never aborts the sweep over the rest of the
        roster.

        Returns:
            The number of aliases whose heartbeat was successfully recorded.
        """
        with self._tracer.span("HeartbeatPump.beat_once"):
            beaten = 0
            for alias in self._roster():
                try:
                    if self._probe.record_heartbeat(alias):
                        beaten += 1
                except Exception:
                    logger.exception("Heartbeat pump failed to beat volk %s", alias)
            return beaten

    def start(self) -> None:
        """Start the background beat loop; idempotent."""
        if self._thread is not None and self._thread.is_alive():
            return
        self._stop_event.clear()
        self._thread = threading.Thread(
            target=self._run,
            name="volk-heartbeat-pump",
            daemon=True,
        )
        self._thread.start()

    def stop(self, timeout: float = 5.0) -> None:
        """Stop the background beat loop and join the thread.

        Args:
            timeout: Max seconds to wait for the thread to join.
        """
        self._stop_event.set()
        if self._thread is not None:
            self._thread.join(timeout=timeout)
        self._thread = None

    def _run(self) -> None:
        """Beat on a fixed interval until :meth:`stop` sets the stop event."""
        while not self._stop_event.wait(self._interval_s):
            self.beat_once()