From 7d6ee0271d5152c4674bf5cf855cacb53e2d6081 Mon Sep 17 00:00:00 2001 From: agent_coder Date: Sun, 5 Jul 2026 10:34:06 +0300 Subject: [PATCH] =?UTF-8?q?feat(stability):=20stage=206=20=E2=80=94=20ligh?= =?UTF-8?q?tweight=20/ping=20healthcheck=20that=20never=20touches=20Telegr?= =?UTF-8?q?am=20(#6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The container healthcheck hit /rss/...?limit=1 (5s timeout): on a cold cache RSS generation exceeds 5s, or a hung TG RPC makes it hang, so docker/autoheal restarts the container mid-download and corrupts temp files. Replace it with /ping, which reflects process/loop liveness (answers instantly, always) plus TG liveness read from the watchdog's last-probe data — issuing ZERO Telegram RPC. - telegram_client: public watchdog_last_ok_age() — seconds since the last successful watchdog probe (None if never). Pure read of the Stage-1 _wd_last_ok_monotonic field; no RPC. - api_server: /ping route (no token, no TG RPC, no SQLite, no fs scan). healthy = connected and (age is None or age < threshold). age is None right after boot => healthy (don't kill before the first probe). connected coerced to bool so the JSON "connected" field is always a bool (pre-start reports false, never null). - config: TG_PING_UNHEALTHY_AFTER knob, default interval*(failures+1)+timeout = 250s (how long until the watchdog itself gives up), env-overridable. - dockercompose.yml: healthcheck -> curl -sf http://127.0.0.1:80/ping, interval 5m, timeout 5s, retries 3, start_period 30s. Old /rss check removed, not left behind. - tests: 10 (healthy/stale/disconnected/fresh-boot/pre-start-null/no-token + the anti-regression zero-TG-RPC spy across all branches + the accessor). Co-Authored-By: Claude Opus 4.8 (1M context) --- api_server.py | 34 +++++- config.py | 11 ++ dockercompose.yml | 11 +- telegram_client.py | 11 ++ tests/mock_config.py | 1 + tests/test_stage6_healthcheck.py | 173 +++++++++++++++++++++++++++++++ 6 files changed, 237 insertions(+), 4 deletions(-) create mode 100644 tests/test_stage6_healthcheck.py diff --git a/api_server.py b/api_server.py index 6b497af..bba4f43 100644 --- a/api_server.py +++ b/api_server.py @@ -33,7 +33,7 @@ from pyrogram import errors from pyrogram.types import Message from pyrogram.enums import MessageMediaType from fastapi import FastAPI, HTTPException, Response, Request -from fastapi.responses import HTMLResponse, FileResponse +from fastapi.responses import HTMLResponse, FileResponse, JSONResponse from telegram_client import TelegramClient from config import get_settings, setup_logging from rss_generator import generate_channel_rss, generate_channel_html @@ -1065,6 +1065,38 @@ async def get_raw_post_json(channel: str, post_id: int, request: Request, token: raise HTTPException(status_code=500, detail=error_message) from e +@app.get("/ping") +async def ping() -> JSONResponse: + """Lightweight liveness probe for the container healthcheck. + + Reflects process/event-loop liveness (always answers in microseconds) and TG liveness + from the watchdog's last-probe data. It MUST NOT issue any Telegram RPC (no get_me, + no safe_get_messages), touch SQLite, or walk the filesystem — that is the whole point: + it stays instant and truthful even while a real TG RPC is hung. It only reads the + already-recorded watchdog timestamp and the is_connected bool. + """ + age = client.watchdog_last_ok_age() # seconds since last OK probe, None if never + # is_connected is None before client.start() and a bool afterwards; coerce so the JSON + # "connected" field is always a bool (never null) and the pre-start window reports false. + connected = bool(client.client.is_connected) + threshold = Config["tg_ping_unhealthy_after"] + # age is None right after boot: the watchdog hasn't run its first probe yet. Treat that + # as healthy (gate on connected only) so a freshly-started container is not killed before + # its first probe cycle — otherwise start_period would have to cover a full watchdog interval. + # Note: when the watchdog is DISABLED (TG_WATCHDOG_ENABLED=false) age stays None forever, + # so /ping degenerates to a pure connectivity check and cannot detect a stale-but-connected + # ("zombie") session — that TG-liveness signal only exists while the watchdog runs. + healthy = connected and (age is None or age < threshold) + return JSONResponse( + { + "status": "ok" if healthy else "degraded", + "connected": connected, + "last_probe_age_s": round(age, 1) if age is not None else None, + "threshold_s": threshold, + }, + status_code=200 if healthy else 503, + ) + @app.get("/health") @app.get("/health/{token}") async def health_check(request: Request, token: str | None = None) -> Response: diff --git a/config.py b/config.py index a8fc395..3cc8763 100644 --- a/config.py +++ b/config.py @@ -123,6 +123,17 @@ def get_settings() -> dict[str, Any]: "tg_watchdog_heartbeat_every": _parse_int_env("TG_WATCHDOG_HEARTBEAT_EVERY", 30), "tg_disconnect_flap_limit": _parse_int_env("TG_DISCONNECT_FLAP_LIMIT", 3), "tg_disconnect_flap_window": _parse_int_env("TG_DISCONNECT_FLAP_WINDOW", 120), + # /ping reports TG as unhealthy once the last successful watchdog probe is older than + # this many seconds. Default is derived from the watchdog cadence: it is roughly how + # long the watchdog itself would take to give up and trigger a restart — + # interval * (failures + 1) + timeout. With the defaults (60,3,10) that is 250s, so a + # transient slow probe never flaps /ping, but a genuinely stuck session (no successful + # probe for ~4 min) surfaces as 503 before/around the time the watchdog restarts. + "tg_ping_unhealthy_after": _parse_int_env( + "TG_PING_UNHEALTHY_AFTER", + _parse_int_env("TG_WATCHDOG_INTERVAL", 60) * (_parse_int_env("TG_WATCHDOG_FAILURES", 3) + 1) + + _parse_int_env("TG_WATCHDOG_TIMEOUT", 10), + ), # Media download timeout scales with file size (large videos): the per-download # timeout is clamped to [min, max] seconds, with an effective floor of # `media_download_min_speed` bytes/s (timeout ≈ file_size / min_speed). diff --git a/dockercompose.yml b/dockercompose.yml index c22711a..4fa8bda 100644 --- a/dockercompose.yml +++ b/dockercompose.yml @@ -53,10 +53,15 @@ services: com.centurylinklabs.watchtower.enable: "true" autoheal: true healthcheck: - test: ["CMD", "curl", "-s", "-f", "-o", "/dev/null", "http://127.0.0.1:80/rss/vvzvlad_lytdybr?limit=1"] - interval: 30m + # Lightweight process/loop liveness probe. /ping never touches Telegram or the + # filesystem, so it answers instantly even while a TG RPC is hung — unlike the old + # /rss?limit=1 check, which could exceed the 5s timeout on a cold cache and get the + # container restarted mid-download (corrupted temp files). TG liveness is now judged + # by the in-process watchdog, which /ping reports via its last-probe age. + test: ["CMD", "curl", "-sf", "http://127.0.0.1:80/ping"] + interval: 5m timeout: 5s - retries: 2 + retries: 3 start_period: 30s start_interval: 5s diff --git a/telegram_client.py b/telegram_client.py index da9793e..6d4f547 100644 --- a/telegram_client.py +++ b/telegram_client.py @@ -259,6 +259,17 @@ class TelegramClient: # Emergency termination os._exit(1) + def watchdog_last_ok_age(self) -> float | None: + """Seconds since the last successful watchdog probe, or None if none succeeded yet. + + Reads only the already-recorded monotonic timestamp set by the watchdog loop; it + never issues a Telegram RPC, so it is safe to call from the hot /ping path even + while a real RPC is hung. + """ + if self._wd_last_ok_monotonic is None: + return None + return time.monotonic() - self._wd_last_ok_monotonic + async def safe_get_messages(self, channel_id, post_id, max_retries=2): """Wrapper with retry logic for auth errors""" for attempt in range(max_retries): diff --git a/tests/mock_config.py b/tests/mock_config.py index d6bacfe..cfed846 100644 --- a/tests/mock_config.py +++ b/tests/mock_config.py @@ -33,6 +33,7 @@ def get_settings(): "tg_watchdog_heartbeat_every": 30, "tg_disconnect_flap_limit": 3, "tg_disconnect_flap_window": 120, + "tg_ping_unhealthy_after": 250, "media_download_timeout_min": 120, "media_download_timeout_max": 1800, "media_download_min_speed": 256 * 1024, diff --git a/tests/test_stage6_healthcheck.py b/tests/test_stage6_healthcheck.py new file mode 100644 index 0000000..8f54b63 --- /dev/null +++ b/tests/test_stage6_healthcheck.py @@ -0,0 +1,173 @@ +# flake8: noqa +# pylint: disable=protected-access, missing-function-docstring, missing-class-docstring +# pylint: disable=redefined-outer-name, logging-fstring-interpolation, line-too-long +# pylance: disable=reportMissingImports, reportMissingModuleSource +""" +Stage 6 (lightweight /ping healthcheck) regression tests. + +Covers: +- /ping returns 200 "ok" when connected and the last probe is recent (age < threshold). +- /ping returns 503 "degraded" when connected but the last probe is stale (age > threshold). +- /ping returns 503 "degraded" when disconnected, regardless of probe age. +- /ping returns 200 "ok" on a fresh boot (age is None) while connected — a freshly-started + container must NOT be killed before the watchdog's first probe. +- ANTI-REGRESSION (the critical invariant): /ping issues ZERO Telegram RPC. A spy on the + fake client's get_me / safe_get_messages proves neither is ever called. +- TelegramClient.watchdog_last_ok_age(): None when never probed; a positive float afterwards. +""" +import os +import sys +import time + +import pytest + +# Add project root to sys.path and mock the config module (same pattern as the other tests). +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..'))) +sys.modules['config'] = __import__('tests.mock_config', fromlist=['get_settings']) + +from fastapi.testclient import TestClient + +import api_server +from telegram_client import TelegramClient + + +# --------------------------------------------------------------------------- # +# Fakes +# --------------------------------------------------------------------------- # +class _FakeKurigramClient: + """Stands in for TelegramClient.client — exposes is_connected and RPC spies.""" + def __init__(self, is_connected=True): + self.is_connected = is_connected + self.get_me_calls = 0 + + async def get_me(self): + # If /ping ever touches this, the whole point of the endpoint is defeated. + self.get_me_calls += 1 + raise AssertionError("/ping must never call get_me()") + + +class _FakeTelegramClient: + """Stands in for api_server.client with a controllable probe age + RPC spies.""" + def __init__(self, age, is_connected=True): + self._age = age + self.client = _FakeKurigramClient(is_connected=is_connected) + self.safe_get_messages_calls = 0 + + def watchdog_last_ok_age(self): + return self._age + + async def safe_get_messages(self, *args, **kwargs): + self.safe_get_messages_calls += 1 + raise AssertionError("/ping must never call safe_get_messages()") + + +@pytest.fixture +def patch_client(monkeypatch): + """Return a factory that installs a fake api_server.client and yields a TestClient.""" + def _install(age, is_connected=True): + fake = _FakeTelegramClient(age=age, is_connected=is_connected) + monkeypatch.setattr(api_server, "client", fake) + return fake, TestClient(api_server.app) + return _install + + +THRESHOLD = api_server.Config["tg_ping_unhealthy_after"] # 250 in mock_config + + +# --------------------------------------------------------------------------- # +# /ping endpoint behavior +# --------------------------------------------------------------------------- # +def test_ping_healthy_connected_recent(patch_client): + fake, tc = patch_client(age=THRESHOLD - 10, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert body["connected"] is True + assert body["last_probe_age_s"] == round(THRESHOLD - 10, 1) + assert body["threshold_s"] == THRESHOLD + assert fake.client.get_me_calls == 0 + + +def test_ping_degraded_stale_probe(patch_client): + fake, tc = patch_client(age=THRESHOLD + 100, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is True + assert fake.client.get_me_calls == 0 + + +def test_ping_degraded_disconnected(patch_client): + # Even with a fresh probe age, a disconnected client is unhealthy. + fake, tc = patch_client(age=1.0, is_connected=False) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is False + + +def test_ping_degraded_disconnected_even_when_age_none(patch_client): + fake, tc = patch_client(age=None, is_connected=False) + r = tc.get("/ping") + assert r.status_code == 503 + assert r.json()["status"] == "degraded" + + +def test_ping_fresh_boot_age_none_is_healthy(patch_client): + # Right after boot the watchdog hasn't probed yet (age None); connected => healthy. + fake, tc = patch_client(age=None, is_connected=True) + r = tc.get("/ping") + assert r.status_code == 200 + body = r.json() + assert body["status"] == "ok" + assert body["last_probe_age_s"] is None + assert body["threshold_s"] == THRESHOLD + + +def test_ping_pre_start_connected_none_is_degraded_bool(patch_client): + # Before client.start(), Kurigram's is_connected is None. /ping must not 500: it coerces + # to a bool, so "connected" is false (never null) and the endpoint reports 503 degraded. + fake, tc = patch_client(age=None, is_connected=None) + r = tc.get("/ping") + assert r.status_code == 503 + body = r.json() + assert body["status"] == "degraded" + assert body["connected"] is False # bool, not null + assert fake.client.get_me_calls == 0 + + +def test_ping_issues_no_tg_rpc(patch_client): + """The critical invariant: /ping never issues any Telegram RPC in any branch.""" + for age, connected in [(1.0, True), (THRESHOLD + 500, True), (None, True), (1.0, False)]: + fake, tc = patch_client(age=age, is_connected=connected) + tc.get("/ping") + assert fake.client.get_me_calls == 0, f"get_me called (age={age}, connected={connected})" + assert fake.safe_get_messages_calls == 0, f"safe_get_messages called (age={age}, connected={connected})" + + +def test_ping_route_needs_no_token(patch_client): + # /ping is unauthenticated by design (no token path variant); it just works. + fake, tc = patch_client(age=1.0, is_connected=True) + assert tc.get("/ping").status_code == 200 + + +# --------------------------------------------------------------------------- # +# TelegramClient.watchdog_last_ok_age accessor +# --------------------------------------------------------------------------- # +def test_watchdog_last_ok_age_none_when_never_probed(): + tgc = TelegramClient() + assert tgc._wd_last_ok_monotonic is None + assert tgc.watchdog_last_ok_age() is None + + +def test_watchdog_last_ok_age_positive_after_probe(): + tgc = TelegramClient() + tgc._wd_last_ok_monotonic = time.monotonic() - 5 + age = tgc.watchdog_last_ok_age() + assert age is not None + assert age >= 5.0 + # Sanity: a plausible upper bound so we didn't accidentally read the wrong field. + assert age < 60.0