Health Checks

broker_health_check

broker_health_check(broker: HealthProvider | Any, config: HealthCheckConfig | None = None, management_client: Any = None) -> BrokerHealthResult

Check broker health status (sync).

Parameters:

Name	Type	Description	Default
broker	HealthProvider | Any	A broker implementing :class:HealthProvider (typed properties) or a legacy broker exposing private attributes (_started, _transport, _worker_pool).	required
config	HealthCheckConfig | None	Optional :class:HealthCheckConfig to tune thresholds.	None
management_client	Any	Optional :class:~rabbitkit.management.RabbitManagementClient (sync .health_check()). When given, its result is folded in as an additional signal: this process may hold a perfectly live connection to one node while the rest of a partitioned cluster is unreachable, which the process-local checks alone cannot detect. A failing management check downgrades an otherwise-HEALTHY result to DEGRADED (never overrides an UNHEALTHY local result). Omit for the original process-local-only behavior.	None

Returns:

Type	Description
BrokerHealthResult	BrokerHealthResult with status HEALTHY (started, connected, not
BrokerHealthResult	blocked, all consumers active), DEGRADED (started but connection
BrokerHealthResult	blocked (L15), consumers missing, pool backlog high, or — with
BrokerHealthResult	management_client — the management API reports the node
BrokerHealthResult	unhealthy), or UNHEALTHY (not started or not connected).

Source code in src/rabbitkit/health.py

def broker_health_check(
    broker: HealthProvider | Any,
    config: HealthCheckConfig | None = None,
    management_client: Any = None,
) -> BrokerHealthResult:
    """Check broker health status (sync).

    Args:
        broker: A broker implementing :class:`HealthProvider` (typed
            properties) or a legacy broker exposing private attributes
            (``_started``, ``_transport``, ``_worker_pool``).
        config: Optional :class:`HealthCheckConfig` to tune thresholds.
        management_client: Optional :class:`~rabbitkit.management.RabbitManagementClient`
            (sync ``.health_check()``). When given, its result is folded in
            as an additional signal: this process may hold a perfectly live
            connection to one node while the rest of a partitioned cluster is
            unreachable, which the process-local checks alone cannot detect.
            A failing management check downgrades an otherwise-HEALTHY result
            to DEGRADED (never overrides an UNHEALTHY local result). Omit for
            the original process-local-only behavior.

    Returns:
        BrokerHealthResult with status HEALTHY (started, connected, not
        blocked, all consumers active), DEGRADED (started but connection
        blocked (L15), consumers missing, pool backlog high, or — with
        ``management_client`` — the management API reports the node
        unhealthy), or UNHEALTHY (not started or not connected).
    """
    result = _local_broker_health_check(broker, config=config)
    if management_client is not None and result.status != HealthStatus.UNHEALTHY:
        result = _apply_management_check(result, management_client.health_check())
    return result

broker_liveness / broker_readiness

broker_liveness(broker: HealthProvider | Any, wedged_timeout: float = 60.0) -> bool

Liveness probe — is the broker process alive and not hard-wedged?

Parameters:

Name	Type	Description	Default
broker	HealthProvider | Any	A broker implementing :class:HealthProvider (typed properties) or a legacy broker exposing private attributes (_started, _wedged, last_heartbeat).	required
wedged_timeout	float	Seconds without a heartbeat before liveness fails.	60.0

Liveness fails when:

• the broker is not started (_started/started False/absent), OR
• an explicit _wedged flag is set to True (transports/brokers may set this on a hard fault), OR
• a last_heartbeat is present and now - last_heartbeat > wedged_timeout (a heartbeat was recorded via :func:mark_heartbeat but has gone stale, meaning the I/O loop is wedged).

A transient broker/transport disconnect is not a liveness failure: the process is still running and can recover. Use :func:broker_readiness to decide whether to route traffic.

When no last_heartbeat attribute exists, liveness falls back to the _started / _wedged checks only (backwards compatible).

Source code in src/rabbitkit/health.py

def broker_liveness(broker: HealthProvider | Any, wedged_timeout: float = 60.0) -> bool:
    """Liveness probe — is the broker process alive and not hard-wedged?

    Args:
        broker: A broker implementing :class:`HealthProvider` (typed
            properties) or a legacy broker exposing private attributes
            (``_started``, ``_wedged``, ``last_heartbeat``).
        wedged_timeout: Seconds without a heartbeat before liveness fails.

    Liveness fails when:

    - the broker is not started (``_started``/``started`` False/absent), OR
    - an explicit ``_wedged`` flag is set to ``True`` (transports/brokers may
      set this on a hard fault), OR
    - a ``last_heartbeat`` is present and ``now - last_heartbeat >
      wedged_timeout`` (a heartbeat was recorded via :func:`mark_heartbeat`
      but has gone stale, meaning the I/O loop is wedged).

    A transient broker/transport disconnect is *not* a liveness failure: the
    process is still running and can recover. Use :func:`broker_readiness` to
    decide whether to route traffic.

    When no ``last_heartbeat`` attribute exists, liveness falls back to the
    ``_started`` / ``_wedged`` checks only (backwards compatible).
    """
    started = _get_started(broker)
    if not started:
        return False
    if _get(broker, "wedged", "_wedged", False):
        return False
    last_heartbeat = _get_last_heartbeat(broker)
    if last_heartbeat is not None:
        if time.monotonic() - last_heartbeat > wedged_timeout:
            return False
    return True

broker_readiness(broker: HealthProvider | Any, config: HealthCheckConfig | None = None, management_client: Any = None) -> bool

Readiness probe — is the broker ready to serve traffic right now?

Parameters:

Name	Type	Description	Default
broker	HealthProvider | Any	A broker implementing :class:HealthProvider (typed properties) or a legacy broker exposing private attributes.	required
config	HealthCheckConfig | None	Optional :class:HealthCheckConfig to tune thresholds.	None
management_client	Any	Optional :class:~rabbitkit.management.RabbitManagementClient — see :func:broker_health_check. A failing management check fails readiness even if this process's own connection looks fine.	None

Requires: health check not UNHEALTHY, transport connected, connection not blocked by broker flow control (L15), and every registered route has an active (live) consumer. Use this for load-balancer / ingress gating; use :func:broker_liveness for restart decisions.

Source code in src/rabbitkit/health.py

def broker_readiness(
    broker: HealthProvider | Any,
    config: HealthCheckConfig | None = None,
    management_client: Any = None,
) -> bool:
    """Readiness probe — is the broker ready to serve traffic right now?

    Args:
        broker: A broker implementing :class:`HealthProvider` (typed
            properties) or a legacy broker exposing private attributes.
        config: Optional :class:`HealthCheckConfig` to tune thresholds.
        management_client: Optional :class:`~rabbitkit.management.RabbitManagementClient`
            — see :func:`broker_health_check`. A failing management check
            fails readiness even if this process's own connection looks fine.

    Requires: health check not UNHEALTHY, transport connected, connection
    not blocked by broker flow control (L15), and every registered route
    has an active (live) consumer. Use this for load-balancer / ingress
    gating; use :func:`broker_liveness` for restart decisions.
    """
    result = broker_health_check(broker, config=config, management_client=management_client)
    if result.status == HealthStatus.UNHEALTHY:
        return False
    if not result.connected:
        return False
    # L15: a blocked connection can't publish -- not ready for traffic even
    # though it's technically still "connected" and may still have live
    # consumers.
    if result.blocked:
        return False
    # A failing management check downgrades to DEGRADED rather than
    # UNHEALTHY (this process's own connection may be fine), but a
    # partitioned/unreachable node is still not ready for traffic.
    if "management_check" in result.details:
        return False
    # M-SRE3: every route must have a live consumer. The health check already
    # verified transport connectivity above.
    return result.consumer_count == result.route_count

broker_liveness_async(broker: HealthProvider | Any, wedged_timeout: float = 60.0) -> bool async

Async variant of :func:broker_liveness.

Source code in src/rabbitkit/health.py

async def broker_liveness_async(broker: HealthProvider | Any, wedged_timeout: float = 60.0) -> bool:
    """Async variant of :func:`broker_liveness`."""
    return broker_liveness(broker, wedged_timeout=wedged_timeout)

broker_readiness_async(broker: HealthProvider | Any, config: HealthCheckConfig | None = None, management_client: Any = None) -> bool async

Async variant of :func:broker_readiness.

Uses :func:broker_health_check_async (management_client.health_check_async()) rather than delegating to the sync broker_readiness — the sync management check makes a blocking network call, which must not run on the event loop.

Source code in src/rabbitkit/health.py

async def broker_readiness_async(
    broker: HealthProvider | Any,
    config: HealthCheckConfig | None = None,
    management_client: Any = None,
) -> bool:
    """Async variant of :func:`broker_readiness`.

    Uses :func:`broker_health_check_async` (``management_client.health_check_async()``)
    rather than delegating to the sync ``broker_readiness`` — the sync
    management check makes a blocking network call, which must not run on
    the event loop.
    """
    result = await broker_health_check_async(broker, config=config, management_client=management_client)
    if result.status == HealthStatus.UNHEALTHY:
        return False
    if not result.connected:
        return False
    if result.blocked:
        return False
    if "management_check" in result.details:
        return False
    return result.consumer_count == result.route_count

HealthStatus / BrokerHealthResult

HealthStatus

Bases: str, Enum

Health status levels.

Source code in src/rabbitkit/health.py

class HealthStatus(str, enum.Enum):
    """Health status levels."""

    HEALTHY = "healthy"
    DEGRADED = "degraded"
    UNHEALTHY = "unhealthy"

BrokerHealthResult dataclass

Result of a broker health check.

Source code in src/rabbitkit/health.py

@dataclass(frozen=True, slots=True)
class BrokerHealthResult:
    """Result of a broker health check."""

    status: HealthStatus
    started: bool = False
    connected: bool = False
    consumer_count: int = 0
    route_count: int = 0
    worker_pool_pending: int = 0
    blocked: bool = False
    details: dict[str, Any] = field(default_factory=dict)

HealthProvider Protocol

HealthProvider

Bases: Protocol

Broker health surface — typed alternative to private-attribute probing.

Brokers that expose these read-only properties satisfy the protocol and :mod:rabbitkit.health will use them directly. Brokers that still use private attributes (_started, _transport, ...) are supported via the deprecation fallback in :func:rabbitkit.health._get.

Source code in src/rabbitkit/core/protocols.py

@runtime_checkable
class HealthProvider(Protocol):
    """Broker health surface — typed alternative to private-attribute probing.

    Brokers that expose these read-only properties satisfy the protocol and
    :mod:`rabbitkit.health` will use them directly. Brokers that still use
    private attributes (``_started``, ``_transport``, ...) are supported via
    the deprecation fallback in :func:`rabbitkit.health._get`.
    """

    @property
    def started(self) -> bool:
        """Whether the broker has been started."""
        ...

    @property
    def connected(self) -> bool:
        """Whether the underlying transport is connected."""
        ...

    @property
    def consumer_count(self) -> int:
        """Number of routes with an active (live) consumer."""
        ...

    @property
    def route_count(self) -> int:
        """Total number of registered routes."""
        ...

    @property
    def worker_pool_pending(self) -> int:
        """Current worker-pool backlog (pending tasks)."""
        ...

    @property
    def last_heartbeat(self) -> float | None:
        """Last liveness heartbeat (monotonic seconds), or None."""
        ...

Attributes

started: bool property

Whether the broker has been started.

connected: bool property

Whether the underlying transport is connected.

consumer_count: int property

Number of routes with an active (live) consumer.

route_count: int property

Total number of registered routes.

worker_pool_pending: int property

Current worker-pool backlog (pending tasks).

last_heartbeat: float | None property

Last liveness heartbeat (monotonic seconds), or None.

HealthWatcher

Opt-in push-style health notifications with debounced transitions. On Kubernetes keep probes primary; this is for bare metal/VMs and direct pager/webhook wiring.

HealthWatcher

Opt-in push-style health notifications (sync, daemon-thread poller).

Polls :func:broker_health_check every interval seconds and fires on_change(old, new, result) when the status transitions -- but only after debounce consecutive identical readings, so a single flapping poll never pages anyone. Callback exceptions are logged, never raised, and never stall the loop.

Positioning: for deployments that aren't (only) Kubernetes -- bare metal, VMs, direct pager/webhook wiring. On k8s, keep :func:broker_liveness/:func:broker_readiness probes as the primary signal; this watcher complements, never replaces, them.

When collector is given (any MetricsCollector with set_gauge), every poll also emits a rabbitkit_health_state gauge (0 healthy / 1 degraded / 2 unhealthy), so Prometheus users get a state series without writing a callback.

clock and sleeper are injectable for tests (no wall-clock sleeps in the test suite -- a hard-won deflaking lesson).

Source code in src/rabbitkit/health.py

class HealthWatcher:
    """Opt-in push-style health notifications (sync, daemon-thread poller).

    Polls :func:`broker_health_check` every *interval* seconds and fires
    ``on_change(old, new, result)`` when the status transitions -- but only
    after *debounce* consecutive identical readings, so a single flapping
    poll never pages anyone. Callback exceptions are logged, never raised,
    and never stall the loop.

    Positioning: for deployments that aren't (only) Kubernetes -- bare
    metal, VMs, direct pager/webhook wiring. On k8s, keep
    :func:`broker_liveness`/:func:`broker_readiness` probes as the primary
    signal; this watcher complements, never replaces, them.

    When *collector* is given (any ``MetricsCollector`` with ``set_gauge``),
    every poll also emits a ``rabbitkit_health_state`` gauge
    (0 healthy / 1 degraded / 2 unhealthy), so Prometheus users get a state
    series without writing a callback.

    *clock* and *sleeper* are injectable for tests (no wall-clock sleeps in
    the test suite -- a hard-won deflaking lesson).
    """

    _GAUGE_VALUES: typing.ClassVar[dict[HealthStatus, int]] = {
        HealthStatus.HEALTHY: 0,
        HealthStatus.DEGRADED: 1,
        HealthStatus.UNHEALTHY: 2,
    }

    def __init__(
        self,
        broker: HealthProvider | Any,
        *,
        interval: float = 10.0,
        on_change: Any = None,
        management_client: Any = None,
        config: HealthCheckConfig | None = None,
        debounce: int = 2,
        collector: Any = None,
        gauge_name: str = "rabbitkit_health_state",
    ) -> None:
        if interval <= 0:
            raise ValueError(f"HealthWatcher interval must be > 0, got {interval}")
        if debounce < 1:
            raise ValueError(f"HealthWatcher debounce must be >= 1, got {debounce}")
        self._broker = broker
        self._interval = interval
        self._on_change = on_change
        self._management_client = management_client
        self._config = config
        self._debounce = debounce
        self._collector = collector
        self._gauge_name = gauge_name

        self._current: HealthStatus | None = None  # last CONFIRMED status
        self._candidate: HealthStatus | None = None
        self._candidate_count = 0
        self._thread: Any = None
        self._stop_event: Any = None

    @property
    def current_status(self) -> HealthStatus | None:
        """Last debounce-confirmed status (None until the first confirmation)."""
        return self._current

    def _tick(self) -> None:
        """One poll: read health, then run the shared debounced state machine."""
        result = broker_health_check(
            self._broker, config=self._config, management_client=self._management_client
        )
        self._apply(result)

    def _apply(self, result: BrokerHealthResult) -> None:
        """Debounced state machine on an already-obtained result (shared with
        the async variant)."""
        if self._collector is not None:
            try:
                self._collector.set_gauge(self._gauge_name, {}, self._GAUGE_VALUES[result.status])
            except Exception:  # pragma: no cover — collector bugs never stall the loop
                logger.exception("HealthWatcher gauge emission raised")

        status = result.status
        if status == self._current:
            # Confirmed state re-observed; reset any half-built candidate.
            self._candidate = None
            self._candidate_count = 0
            return
        if status != self._candidate:
            self._candidate = status
            self._candidate_count = 0
        self._candidate_count += 1
        if self._candidate_count < self._debounce:
            return
        old, self._current = self._current, status
        self._candidate = None
        self._candidate_count = 0
        if self._on_change is not None:
            try:
                self._on_change(old, status, result)
            except Exception:
                logger.exception("HealthWatcher on_change callback raised")

    def start(self) -> None:
        """Start the daemon poller thread. Idempotent."""
        import threading

        if self._thread is not None and self._thread.is_alive():
            return
        self._stop_event = threading.Event()
        stop = self._stop_event

        def _loop() -> None:
            while not stop.wait(timeout=self._interval):
                try:
                    self._tick()
                except Exception:  # pragma: no cover — defensive; _tick guards itself
                    logger.exception("HealthWatcher tick raised")

        self._thread = threading.Thread(target=_loop, name="rabbitkit-health-watcher", daemon=True)
        self._thread.start()

    def stop(self, timeout: float = 5.0) -> None:
        """Stop the poller (bounded join). Idempotent."""
        if self._stop_event is not None:
            self._stop_event.set()
        if self._thread is not None:
            self._thread.join(timeout=timeout)
            self._thread = None

Attributes

current_status: HealthStatus | None property

Last debounce-confirmed status (None until the first confirmation).

Methods:

start() -> None

Start the daemon poller thread. Idempotent.

Source code in src/rabbitkit/health.py

def start(self) -> None:
    """Start the daemon poller thread. Idempotent."""
    import threading

    if self._thread is not None and self._thread.is_alive():
        return
    self._stop_event = threading.Event()
    stop = self._stop_event

    def _loop() -> None:
        while not stop.wait(timeout=self._interval):
            try:
                self._tick()
            except Exception:  # pragma: no cover — defensive; _tick guards itself
                logger.exception("HealthWatcher tick raised")

    self._thread = threading.Thread(target=_loop, name="rabbitkit-health-watcher", daemon=True)
    self._thread.start()

stop(timeout: float = 5.0) -> None

Stop the poller (bounded join). Idempotent.

Source code in src/rabbitkit/health.py

def stop(self, timeout: float = 5.0) -> None:
    """Stop the poller (bounded join). Idempotent."""
    if self._stop_event is not None:
        self._stop_event.set()
    if self._thread is not None:
        self._thread.join(timeout=timeout)
        self._thread = None

AsyncHealthWatcher

Bases: HealthWatcher

Async variant of :class:HealthWatcher — an asyncio task instead of a thread, and the management check (if any) awaited via :func:broker_health_check_async so it never blocks the event loop.

Source code in src/rabbitkit/health.py

class AsyncHealthWatcher(HealthWatcher):
    """Async variant of :class:`HealthWatcher` — an asyncio task instead of
    a thread, and the management check (if any) awaited via
    :func:`broker_health_check_async` so it never blocks the event loop."""

    async def _tick_async(self) -> None:
        result = await broker_health_check_async(
            self._broker, config=self._config, management_client=self._management_client
        )
        self._apply(result)

    async def run(self) -> None:
        """Poll forever (cancel the task to stop)."""
        import asyncio

        while True:
            await asyncio.sleep(self._interval)
            try:
                await self._tick_async()
            except Exception:  # pragma: no cover — defensive; _tick guards itself
                logger.exception("HealthWatcher tick raised")

Methods:

run() -> None async

Poll forever (cancel the task to stop).

Source code in src/rabbitkit/health.py

async def run(self) -> None:
    """Poll forever (cancel the task to stop)."""
    import asyncio

    while True:
        await asyncio.sleep(self._interval)
        try:
            await self._tick_async()
        except Exception:  # pragma: no cover — defensive; _tick guards itself
            logger.exception("HealthWatcher tick raised")