claude code agent
03d1de2954
F1 [WARNING] pytest-asyncio was not declared, so the 6 new async stage-1 tests
ERROR on a clean checkout (async def not natively supported) — zero regression
protection, masked by a globally-installed plugin. Added pytest-asyncio to
requirements.txt + asyncio_mode=auto to tests/pytest.ini. Verified on a fresh venv
from requirements alone: the tests collect and run (180 passed).
F2 [WARNING] The /raw_json endpoint's get_messages was the one remaining live RPC
without a timeout, violating the stage-1 DoD ('every Telegram RPC is bounded').
Wrapped it in asyncio.wait_for(..., 30) mirroring PostParser.get_post. (It is not
under the tg_rpc gate, so its blast radius was one request, not the app.)
F3 [WARNING] The worker test globally no-op'd asyncio.sleep, so the dedicated
except FloodWait branch was indistinguishable from the generic handler — deleting
it kept the test green. The sleep stub now records delays and the test asserts the
FloodWait backoff of 6 (=min(1+5,900)), distinct from the success path's 2.
F5 [low] The tricky 'gate outside, timeout inside' nesting was open-coded at 3
sites (each re-deriving the invariant). Extracted tg_rpc_bounded(timeout) into
tg_throttle (using asyncio.timeout()); the 3 sites now use it, so a future call
site cannot silently wrap the gate entry and reopen the hang-under-backpressure.
F4 [low] Documented TG_RPC_TIMEOUT in the dockercompose.yml environment block
next to the other TG_RPC_* knobs.
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
93 lines
3.3 KiB
Python
93 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
# flake8: noqa
|
|
# pylint: disable=broad-exception-caught, global-statement, missing-function-docstring, missing-class-docstring
|
|
# pylint: disable=logging-fstring-interpolation, line-too-long
|
|
|
|
import os
|
|
import time
|
|
import asyncio
|
|
import logging
|
|
from contextlib import asynccontextmanager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _parse_int_env(name: str, default: int, minimum: int) -> int:
|
|
"""Parse an int env var, falling back to default on missing/invalid/out-of-range value."""
|
|
raw = os.getenv(name)
|
|
if raw is None or raw.strip() == "":
|
|
return default
|
|
try:
|
|
value = int(raw)
|
|
except ValueError:
|
|
logger.warning(f"tg_throttle: {name} is not a valid integer ({raw!r}); using default {default}")
|
|
return default
|
|
if value < minimum:
|
|
logger.warning(f"tg_throttle: {name}={value} below minimum {minimum}; using default {default}")
|
|
return default
|
|
return value
|
|
|
|
|
|
# Global throttle for live Telegram MTProto RPC calls. Serializes bursts (e.g. miniflux
|
|
# batching ~47 feeds at once) so they do not trip Telegram's FLOOD_WAIT (420).
|
|
_CONCURRENCY = _parse_int_env("TG_RPC_CONCURRENCY", 1, 1) # max concurrent Telegram RPCs
|
|
_MIN_INTERVAL = _parse_int_env("TG_RPC_MIN_INTERVAL_MS", 500, 0) / 1000.0 # min gap between RPC starts (seconds)
|
|
|
|
_sem = asyncio.Semaphore(_CONCURRENCY)
|
|
_lock = asyncio.Lock()
|
|
_last_start = 0.0
|
|
|
|
logger.info(f"tg_throttle: initialized (concurrency={_CONCURRENCY}, min_interval={_MIN_INTERVAL*1000:.0f}ms)")
|
|
|
|
|
|
class _TgRpcGate:
|
|
"""Async context manager that caps concurrency and enforces a minimum spacing between RPC starts."""
|
|
|
|
async def __aenter__(self):
|
|
await _sem.acquire()
|
|
global _last_start
|
|
try:
|
|
async with _lock:
|
|
wait = _last_start + _MIN_INTERVAL - time.monotonic()
|
|
if wait > 0:
|
|
await asyncio.sleep(wait)
|
|
_last_start = time.monotonic()
|
|
except BaseException:
|
|
# Do not leak a semaphore permit if cancelled while waiting for the spacing.
|
|
_sem.release()
|
|
raise
|
|
return self
|
|
|
|
async def __aexit__(self, exc_type, exc, tb):
|
|
_sem.release()
|
|
return False
|
|
|
|
|
|
def tg_rpc():
|
|
"""Return an async context manager that throttles a single live Telegram RPC call."""
|
|
return _TgRpcGate()
|
|
|
|
|
|
@asynccontextmanager
|
|
async def tg_rpc_bounded(timeout: float):
|
|
"""Throttle a live Telegram RPC AND bound it with a timeout, correctly nested.
|
|
|
|
The single tricky invariant this centralizes: the timeout must bound ONLY the
|
|
RPC body, never the gate ENTRY — timing out the `_sem.acquire()` / spacing wait
|
|
would turn legitimate queue backpressure (e.g. ~47 feeds queueing) into false
|
|
timeouts. So the gate is the OUTER context and the timeout is the INNER one; a
|
|
TimeoutError raised inside propagates out through the gate's `__aexit__`, which
|
|
releases the permit (no leak). Call as:
|
|
|
|
async with tg_rpc_bounded(Config["tg_rpc_timeout"]):
|
|
result = await client.get_chat(channel_id)
|
|
|
|
Every gated+bounded RPC uses this so no call site re-derives the nesting by hand
|
|
(getting it wrong silently reopens the hang-under-backpressure class).
|
|
"""
|
|
async with _TgRpcGate():
|
|
async with asyncio.timeout(timeout):
|
|
yield
|