signal-bot/signal_bot/latency.py

# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
"""Hold the bot's own share of a reply to its budget, against the live bot.

``DOCS-latency.md`` measured a reply from the outside and found the bot's
code was a rounding error next to the container and the upstream fetch. That
is a claim about one measurement of one command; this is the check that keeps
it true. It types one text command at the bot through ``/_selftest`` many
times over, reads the ``served`` timings each answer carries (``served.py``),
and fails if the 95th percentile of ``own_ms`` -- everything that is not a
send or an upstream request -- is over ``OWN_BUDGET_MS``.

The self-test is always the same person, so the second attempt would be
refused by the cooldown. The command's cooldown is overridden to zero for the
run through the same switch the management page uses, and the row is put
back exactly as it was, refusal or crash included.
"""

from __future__ import annotations

import math
import time
from dataclasses import dataclass
from typing import TYPE_CHECKING, Any, Final

import httpx

from signal_bot.served import OWN_BUDGET_MS

if TYPE_CHECKING:
    from collections.abc import Sequence

MS_PER_SECOND: Final = 1000
# What the report prints and what the budget is held against, in that order.
COLUMNS: Final = ("own_ms", "queue_ms", "dispatch_ms", "handler_ms", "send_ms", "round_trip_ms")
PERCENTILES: Final = (50, 95)


class LatencyError(RuntimeError):
    """The run could not measure what it was asked to."""


@dataclass(frozen=True, slots=True)
class Sample:
    """One reply's timings plus how long the whole self-test call took."""

    served: dict[str, Any]
    round_trip_ms: int

    def __getitem__(self, column: str) -> int:
        """One column of the report, whichever side of the wire it was measured on."""
        if column == "round_trip_ms":
            return self.round_trip_ms
        return int(self.served[column])


def percentile(values: Sequence[int], p: int) -> int:
    """Nearest-rank percentile: the value at least ``p``% of the samples sit at or under."""
    ordered = sorted(values)
    rank = max(1, math.ceil(p / 100 * len(ordered)))
    return ordered[rank - 1]


@dataclass(frozen=True, slots=True)
class Report:
    """Every sample of one run, and the verdict."""

    command: str
    samples: tuple[Sample, ...]
    budget_ms: int = OWN_BUDGET_MS

    def at(self, column: str, p: int) -> int:
        """One percentile of one column."""
        return percentile([sample[column] for sample in self.samples], p)

    @property
    def within_budget(self) -> bool:
        """Whether the bot's own p95 is at or under the budget."""
        return self.at("own_ms", 95) <= self.budget_ms

    def render(self) -> str:
        """A table a human reads once and a CI log keeps."""
        lines = [f"{self.command}: {len(self.samples)} replies"]
        lines.extend(
            f"  {column:<14}" + "  ".join(f"p{p}={self.at(column, p):>6} ms" for p in PERCENTILES)
            for column in COLUMNS
        )
        verdict = "within" if self.within_budget else "OVER"
        lines.append(f"  own_ms p95 {verdict} the {self.budget_ms} ms budget")
        return "\n".join(lines)


async def _selftest(client: httpx.AsyncClient, base: str, group: str, text: str) -> Sample:
    started = time.perf_counter()
    response = await client.post(f"{base}/_selftest", json={"group": group, "text": text})
    round_trip_ms = round((time.perf_counter() - started) * MS_PER_SECOND)
    body = response.json()
    if response.status_code != httpx.codes.OK or body.get("status") != "delivered":
        message = f"self-test refused: {body.get('reason', response.text)}"
        raise LatencyError(message)
    if "served" not in body:
        message = f"{text!r} is not a command the bot ran; nothing to time"
        raise LatencyError(message)
    return Sample(served=body["served"], round_trip_ms=round_trip_ms)


async def _setting(client: httpx.AsyncClient, base: str, name: str) -> dict[str, Any]:
    """The command's switch row as it stands, so it can be put back."""
    response = await client.get(f"{base}/_commands")
    response.raise_for_status()
    for row in response.json()["commands"]:
        if row["name"] == name:
            return {
                "enabled": row["enabled"],
                "disabled_until": row["disabled_until"],
                "cooldown_seconds": row["cooldown_seconds"],
            }
    message = f"the bot ran {name!r} but does not list it"
    raise LatencyError(message)


async def _put_setting(
    client: httpx.AsyncClient, base: str, name: str, setting: dict[str, Any]
) -> None:
    response = await client.put(f"{base}/_commands/{name}", json=setting)
    response.raise_for_status()


async def measure(
    client: httpx.AsyncClient,
    base: str,
    group: str,
    text: str,
    times: int,
    budget_ms: int = OWN_BUDGET_MS,
) -> Report:
    """Type ``text`` at the bot ``times`` times and report what each reply cost.

    Sequential on purpose: the point is what one reply costs when nothing
    else is going on, and a burst would measure the queue behind ourselves.
    """
    first = await _selftest(client, base, group, text)
    command = str(first.served["command"])
    original = await _setting(client, base, command)
    await _put_setting(client, base, command, {**original, "cooldown_seconds": 0})
    samples = [first]
    try:
        samples.extend([await _selftest(client, base, group, text) for _ in range(times - 1)])
    finally:
        await _put_setting(client, base, command, original)
    return Report(command=command, samples=tuple(samples), budget_ms=budget_ms)