feat(telegram): calibrate the gate's history category from production logs (#1418)

src/collection_queue.py

@@ -8,6 +8,8 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any, cast
 
+from telethon_floodgate import TelegramRateLimitedError
+
 from src.database import Database, DatabaseBusyError
 from src.database.bundles import ChannelBundle
 from src.live_runtime_pause import LiveRuntimePauseGate
@@ -30,6 +32,10 @@
 # must be matched by BOTH type and message. Messages mirror facade._SQLITE_BUSY_MESSAGES.
 _SQLITE_BUSY_MESSAGES = ("database is locked", "database table is locked", "database is busy")
 
+# Slack added on top of the gate's exact retry_after so the rescheduled run
+# does not land a millisecond before the sliding window actually reopens.
+GATE_RATE_LIMIT_RETRY_BUFFER_SEC = 5.0
+
 
 def _is_transient_busy_error(exc: BaseException) -> bool:
     """True for a transient SQLite lock from either DB path (#1249).
@@ -602,6 +608,33 @@ async def _handle_collection_exception(
                 exc.phone,
             )
             return True, False
+        if isinstance(exc, TelegramRateLimitedError):
+            # The calibrated gate (#1418) legitimately binds on peak collector
+            # minutes (media 50/min vs the 48/min history cap): deferring the
+            # task is the designed outcome, not a failure — mirror the
+            # resolve-rate-limited branch above.
+            run_after = datetime.now(timezone.utc) + timedelta(
+                seconds=exc.retry_after_sec + GATE_RATE_LIMIT_RETRY_BUFFER_SEC
+            )
+            note = (
+                "Отложено: gate "
+                f"{exc.category} rate-limited до {run_after.astimezone(timezone.utc).isoformat()}"
+            )
+            self._retried_tasks.discard(task_id)
+            await self._channels.reschedule_collection_task(task_id, run_after=run_after, note=note)
+            self._schedule_requeue_after_delay(
+                task_id=task_id, channel=channel, force=force, full=full, run_after=run_after
+            )
+            logger.warning(
+                "Rescheduled collection task %d for channel %d until %s: "
+                "gate %s rate-limited on %s",
+                task_id,
+                channel.channel_id,
+                run_after.isoformat(),
+                exc.category,
+                exc.phone,
+            )
+            return True, False
         if isinstance(exc, NoActiveCollectionClientsError):
             run_after = datetime.now(timezone.utc) + timedelta(
                 seconds=self.NO_CLIENTS_RETRY_DELAY_SEC

src/telegram/client_pool.py

@@ -57,7 +57,12 @@
 # #1046 split. The live call sites now live in the mixin modules, which import
 # these names into their own namespaces.
 from telethon.tl.types import ChannelForbidden  # noqa: F401
-from telethon_floodgate import FloodCircuitBreaker, ResolveRateLimiter, TelegramRateLimitGate
+from telethon_floodgate import (
+    FloodCircuitBreaker,
+    RateLimitSpec,
+    ResolveRateLimiter,
+    TelegramRateLimitGate,
+)
 
 from src.config import TelegramRuntimeConfig
 from src.database import Database
@@ -91,6 +96,20 @@
 # duplicated here — nothing reads ``client_pool.<CONST>`` any more (#1046 cleanup).
 
 
+# Phase 2 calibration of the gate's ``history`` category (#1418, epic #1331).
+# The released telethon-floodgate 0.1.0 default (600/min) never bound: the
+# production app.log shows 209 FLOOD_WAITs on messages.getHistory with
+# collector peaks of 115 channel fetches per minute, far below that guard.
+# 24/30s is the empirically measured Telegram boundary (30 requests in ~30s,
+# the 31st returned FLOOD_WAIT_3) with a 20% margin.  Peak collector bursts
+# (p95 74-101 fetches/min) stretch instead of flooding; incremental min_id
+# collection catches deferred channels on the next pass.  Other categories
+# had no flood signal in the logs and keep the package defaults pending a
+# larger sample.  Keep in sync with the package default; drop the override
+# once a released telethon-floodgate ships this value (#1418).
+HISTORY_CALIBRATED_SPEC = RateLimitSpec(max_calls=24, window_sec=30.0)
+
+
 @dataclass(frozen=True)
 class StatsClientAvailability:
     state: str  # "available" | "all_flooded" | "no_connected_active"
@@ -172,10 +191,14 @@ def __init__(
         self._dialog_refresh_tasks: dict[tuple[str, str], asyncio.Task[list[dict]]] = {}
         self._premium_flood_wait_until: dict[str, datetime] = {}
         self._resolve_rate_limiter = ResolveRateLimiter()
-        # Central proactive gate.  Category limits are conservative operating
-        # defaults calibrated from the available production signals; keep the
-        # registry injectable for future recalibration (#1331).
-        self._rate_limit_gate = TelegramRateLimitGate()
+        # Central proactive gate.  ``history`` is calibrated from the
+        # production log sample (#1418); the remaining categories are the
+        # package's conservative operating defaults pending production
+        # signals — keep the registry injectable for future recalibration
+        # (#1331).
+        self._rate_limit_gate = TelegramRateLimitGate(
+            category_limits={"history": HISTORY_CALIBRATED_SPEC},
+        )
         # Reactive counterpart to the gate (#1330/#1368): the gate paces calls
         # against guessed limits, the breaker stops an (operation, phone) pair
         # that Telegram is already flood-waiting instead of hammering on.

tests/test_collection_queue_db_pull.py

@@ -12,6 +12,7 @@
 from datetime import datetime, timedelta, timezone
 
 import pytest
+from telethon_floodgate import TelegramRateLimitedError
 
 from src.collection_queue import CollectionQueue
 from src.database import Database
@@ -77,6 +78,15 @@ async def collect_single_channel(
         raise UsernameResolveRateLimitedError("+7001", 28.2)
 
 
+class _GateRateLimitedCollector(_FakeCollector):
+    async def collect_single_channel(
+        self, channel, *, full=False, progress_callback=None, force=False, cancel_event=None
+    ):
+        self.calls.append(channel.channel_id)
+        self.full_calls.append(full)
+        raise TelegramRateLimitedError("+7001", "history", 17.0)
+
+
 class _BlockingCollector:
     def __init__(self):
         self.calls: list[int] = []
@@ -331,6 +341,54 @@ async def test_username_resolve_rate_limit_keeps_task_pending(tmp_path):
         await db.close()
 
 
+@pytest.mark.anyio
+async def test_gate_rate_limit_keeps_task_pending(tmp_path):
+    """A calibrated-gate deferral reschedules instead of failing (#1418).
+
+    Once the history category binds on peak collector minutes,
+    TelegramRateLimitedError reaches the queue's main path and must take the
+    same reschedule route as the neighbouring resolve-rate-limited branch —
+    not FAILED + logger.exception noise on every peak.
+    """
+    db = Database(str(tmp_path / "queue.db"))
+    await db.initialize()
+    try:
+        await _seed_channel(db)
+
+        collector = _GateRateLimitedCollector()
+        queue = CollectionQueue(collector, db)
+        channel = (await db.get_channels(active_only=True))[0]
+        before = datetime.now(timezone.utc)
+        task_id = await queue.enqueue(channel)
+
+        deadline = asyncio.get_event_loop().time() + 2.0
+        while asyncio.get_event_loop().time() < deadline:
+            task = await db.get_collection_task(task_id)
+            if task.status == "pending" and task.run_after is not None:
+                break
+            await asyncio.sleep(0.05)
+
+        task = await db.get_collection_task(task_id)
+        assert task.status == "pending"
+        assert task.error is None
+        assert task.run_after is not None
+        # retry_after (17s) + the 5s reschedule buffer.
+        assert task.run_after >= before + timedelta(seconds=22)
+        assert "gate history rate-limited" in (task.note or "")
+
+        queue.start_db_pull(interval=0.02)
+        try:
+            await asyncio.sleep(0.12)
+        finally:
+            await queue.stop_db_pull()
+
+        assert task_id in queue._known_task_ids
+        assert len(queue._delayed_requeues) == 1
+    finally:
+        await queue.shutdown()
+        await db.close()
+
+
 @pytest.mark.anyio
 async def test_db_pull_does_not_double_ingest(tmp_path):
     db = Database(str(tmp_path / "queue.db"))

tests/test_rate_limit_gate.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 from types import SimpleNamespace
+from unittest.mock import MagicMock
 
 import pytest
 from telethon import TelegramClient
@@ -16,6 +17,7 @@
 )
 
 from src.telegram.backends import TelegramTransportSession
+from src.telegram.client_pool import HISTORY_CALIBRATED_SPEC
 
 
 class _Clock:
@@ -429,3 +431,41 @@ class Pool:
     assert await session.send_message(12345, "b") == "ok"
     # An entity peer_key cannot read at all yields None and also proceeds.
     assert await session.send_message(object(), "c") == "ok"
+
+
+def test_pool_applies_production_history_calibration() -> None:
+    """The pool's gate must ship the #1418 calibrated history spec.
+
+    The released floodgate default (600/min) never bound: production peaks ran
+    at 115 fetches/min.  The override lives in the pool, so the pool is what
+    this pins; the remaining categories must stay untouched package defaults.
+    """
+    from src.telegram.client_pool import ClientPool
+
+    pool = ClientPool(MagicMock(api_id=1, api_hash="h"), MagicMock())
+    history = pool._rate_limit_gate._limiters["history"]
+    assert history._max_calls == 24
+    assert history._window_sec == 30.0
+    # Untouched categories keep their package defaults.
+    send = pool._rate_limit_gate._limiters["send"]
+    assert (send._max_calls, send._window_sec) == (30, 60.0)
+
+
+def test_history_calibration_stops_a_collector_burst_before_telegram() -> None:
+    """A peak burst (p95 74-101 fetches/min in app.log) defers, not floods.
+
+    24 calls fit the calibrated 30s window; the 25th is refused before any
+    Telegram call, and the budget returns once the window slides.
+    """
+    clock = _Clock()
+    gate = TelegramRateLimitGate(
+        category_limits={"history": HISTORY_CALIBRATED_SPEC},
+        time_func=clock,
+    )
+    for _ in range(HISTORY_CALIBRATED_SPEC.max_calls):
+        assert gate.try_acquire("+7001", "history") == 0.0
+    deferred = gate.try_acquire("+7001", "history")
+    assert deferred > 0
+    # Sliding window: after the window passes, the bucket accepts again.
+    clock.now += HISTORY_CALIBRATED_SPEC.window_sec
+    assert gate.try_acquire("+7001", "history") == 0.0