mirror of
https://github.com/kuhyx/signal-bot.git
synced 2026-11-10 18:17:02 +01:00
Krzysztof kuhy Rudnicki
4af32f8bd5
Bot.listen awaited every frame to completion, so one Commons fetch or one attachment upload held up every text reply behind it (DOCS-latency.md). Each frame now runs in its own task, never more than LIMIT at once; the reader waits for a slot before spawning, so a burst applies back-pressure at the websocket instead of fanning out upstream. One person's commands stay serialised (user_lock.py): two quick !kot would otherwise both pass the cooldown gate before either recorded a use. Every served command reports its timings split into the bot's own time and everybody else's (served.py); /_selftest carries that block and tool/latency.py holds own_ms to 100 ms at p95, with the command's cooldown switched off for the run and put back afterwards. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
130 lines
3.4 KiB
Python
130 lines
3.4 KiB
Python
# Copyright (c) 2026 Krzysztof Rudnicki. MIT licence.
|
|
"""Frames as tasks: a few at once, the rest waiting at the reader."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import logging
|
|
from typing import TYPE_CHECKING
|
|
|
|
from signal_bot.inflight import LIMIT, Inflight
|
|
|
|
if TYPE_CHECKING:
|
|
import pytest
|
|
|
|
|
|
async def _settled() -> None:
|
|
"""Let every task that is ready to run take a turn."""
|
|
for _ in range(3):
|
|
await asyncio.sleep(0)
|
|
|
|
|
|
async def test_start_returns_before_the_work_is_done() -> None:
|
|
gate = asyncio.Event()
|
|
done: list[str] = []
|
|
|
|
async def work() -> None:
|
|
await gate.wait()
|
|
done.append("x")
|
|
|
|
inflight = Inflight()
|
|
await inflight.start(work())
|
|
assert inflight.count == 1
|
|
assert done == []
|
|
gate.set()
|
|
await inflight.drain()
|
|
assert done == ["x"]
|
|
assert inflight.count == 0
|
|
|
|
|
|
async def test_the_reader_waits_only_when_every_slot_is_busy() -> None:
|
|
gate = asyncio.Event()
|
|
|
|
async def work() -> None:
|
|
await gate.wait()
|
|
|
|
inflight = Inflight(limit=2)
|
|
await inflight.start(work())
|
|
await inflight.start(work())
|
|
third = asyncio.ensure_future(inflight.start(work()))
|
|
await _settled()
|
|
# Two slots, two frames in them: the third start is still waiting.
|
|
assert not third.done()
|
|
assert inflight.count == 2
|
|
gate.set()
|
|
await third
|
|
await inflight.drain()
|
|
assert inflight.count == 0
|
|
|
|
|
|
async def test_a_task_that_raises_is_logged_and_frees_its_slot(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
async def boom() -> None:
|
|
msg = "handler broke"
|
|
raise RuntimeError(msg)
|
|
|
|
async def fine() -> None:
|
|
return None
|
|
|
|
inflight = Inflight(limit=1)
|
|
with caplog.at_level(logging.ERROR, logger="signal_bot.inflight"):
|
|
await inflight.start(boom())
|
|
await inflight.drain()
|
|
# The slot came back, or this would hang.
|
|
await asyncio.wait_for(inflight.start(fine()), timeout=1)
|
|
await inflight.drain()
|
|
assert "a frame's task failed" in caplog.text
|
|
assert "handler broke" in caplog.text
|
|
|
|
|
|
async def test_cancel_stops_what_is_in_flight_without_a_log_line(
|
|
caplog: pytest.LogCaptureFixture,
|
|
) -> None:
|
|
async def forever() -> None:
|
|
await asyncio.Event().wait()
|
|
|
|
async def fine() -> None:
|
|
return None
|
|
|
|
inflight = Inflight(limit=1)
|
|
await inflight.start(forever())
|
|
with caplog.at_level(logging.ERROR, logger="signal_bot.inflight"):
|
|
# Cancelled before its first step, which is the case a wrapper's
|
|
# `finally` would miss: the slot must come back regardless.
|
|
inflight.cancel()
|
|
await inflight.drain()
|
|
assert inflight.count == 0
|
|
assert caplog.text == ""
|
|
await asyncio.wait_for(inflight.start(fine()), timeout=1)
|
|
await inflight.drain()
|
|
|
|
|
|
async def test_a_slot_held_inline_counts_against_the_same_limit() -> None:
|
|
gate = asyncio.Event()
|
|
|
|
async def work() -> None:
|
|
await gate.wait()
|
|
|
|
inflight = Inflight(limit=1)
|
|
await inflight.start(work())
|
|
|
|
async def inline() -> str:
|
|
async with inflight.slot():
|
|
return "ran"
|
|
|
|
held = asyncio.ensure_future(inline())
|
|
await _settled()
|
|
assert not held.done()
|
|
gate.set()
|
|
assert await held == "ran"
|
|
await inflight.drain()
|
|
|
|
|
|
async def test_drain_with_nothing_in_flight_returns_at_once() -> None:
|
|
await asyncio.wait_for(Inflight().drain(), timeout=1)
|
|
|
|
|
|
def test_the_default_limit_is_a_handful() -> None:
|
|
assert 1 < LIMIT <= 16
|