monitor.heartbeat

def heartbeat_status(last_ping_at: Optional[int], now: int, interval_seconds: int, grace_seconds: int) -> HeartbeatStatus

HeartbeatState = Literal["new", "up", "late", "down"]

@dataclass(frozen=True)
class HeartbeatStatus:
    """Where a heartbeat stands and when it changes next."""

    state: HeartbeatState
    #: when the next ping is due: lastPingAt + intervalSeconds; null when new
    due_at: Optional[int]
    #: the last second it counts as late: dueAt + graceSeconds; null when new
    down_at: Optional[int]
    #: now - dueAt when late or down, else 0
    overdue_seconds: int

from fune.monitor.heartbeat import heartbeat_status  # monitor.heartbeat@^1

from typing import Optional

from .monitor_heartbeat_types import HeartbeatStatus


def _whole(name: str, v: object) -> None:
    if not isinstance(v, int) or isinstance(v, bool):
        raise ValueError("%s must be whole seconds, received %r" % (name, v))


def heartbeat_status(last_ping_at: Optional[int], now: int, interval_seconds: int, grace_seconds: int) -> HeartbeatStatus:
    """New, up, late or down, as healthchecks.io names them. Up through the
    due second itself, late through the last second of the grace time, down
    after. A ping stamped in the future (clock skew between the job and the
    monitor) is simply up: its due time is later still."""
    if last_ping_at is not None:
        _whole("lastPingAt", last_ping_at)
    _whole("now", now)
    _whole("intervalSeconds", interval_seconds)
    _whole("graceSeconds", grace_seconds)
    if interval_seconds < 1:
        raise ValueError("intervalSeconds must be at least 1, received %d" % interval_seconds)
    if grace_seconds < 0:
        raise ValueError("graceSeconds must not be negative, received %d" % grace_seconds)
    if last_ping_at is None:
        return HeartbeatStatus(state="new", due_at=None, down_at=None, overdue_seconds=0)
    due_at = last_ping_at + interval_seconds
    down_at = due_at + grace_seconds
    if now <= due_at:
        return HeartbeatStatus(state="up", due_at=due_at, down_at=down_at, overdue_seconds=0)
    return HeartbeatStatus(state="late" if now <= down_at else "down", due_at=due_at, down_at=down_at, overdue_seconds=now - due_at)

Install

fune build

With that line in your source, in a Python project (language python in fune.project), fune build resolves it and nothing else, pins them in fune.lock, downloads only the Python package of each, and builds the code above into your project’s .fune/build, one readable file per capability with a header linking back here. Or pin a range in fune.project and build in one step:

fune add monitor.heartbeat

The manifest, vectors and README with only the Python implementation. Install it without the registry with fune add ./monitor.heartbeat-1.0.0-python.fune, or fetch it from a terminal with fune pull monitor.heartbeat@1.0.0:python.

The whole function, every language, is one file too: monitor.heartbeat-1.0.0.fune, 12,774 bytes, sha256 670066e850a8ae45f319f6644290afddae8eaec2c718edba53e6b8a7a51a80f3. It installs into a project of any language.

# fune: before monitor.heartbeat

# fune: after monitor.heartbeat

# fune: step monitor.heartbeat after <n|label>

More from the author

where `dueAt = lastPingAt + intervalSeconds` and `downAt = dueAt + graceSeconds`. The due second itself is still up, and the last second of the grace time is still late; the state changes the second after. With no grace time a check goes straight from up to down.

`overdueSeconds` is `now - dueAt` when late or down (how long past due, for "late by 2m" in an alert), and 0 otherwise. `dueAt` and `downAt` are returned so the caller can schedule its next evaluation instead of polling.

## Edge cases

- A ping stamped after `now` (the job's clock is ahead of the monitor's) is `up`, overdue 0: its due time is later still. It is not an error, because clock skew is normal and refusing it would hide a healthy job. - `paused` is not a state here: pausing is the caller's decision, not something the ping times say. - Only simple period schedules are covered; a cron-expression schedule gives `dueAt` from the expression instead.

## Errors

- `intervalSeconds must be at least 1, received X` - `graceSeconds must not be negative, received X` - `NAME must be whole seconds, received X` for a fractional time

## Sources

- healthchecks.io documentation, "Check states" and "Period and Grace Time": https://healthchecks.io/docs/ and https://healthchecks.io/docs/configuring_checks/

Files