diff --git a/gateway/delivery_ledger.py b/gateway/delivery_ledger.py index af3352754a611..2d06cd6c71d6e 100644 --- a/gateway/delivery_ledger.py +++ b/gateway/delivery_ledger.py @@ -74,10 +74,18 @@ # merely suggests retrying is never mistaken for a flood. _RAW_FLOOD_RE = re.compile(r"flood control exceeded.*?retry in\s+(\d+(?:\.\d+)?)", re.IGNORECASE) +# Some adapters describe their own throttling without either the ``flood_control:`` prefix or PTB's +# wording — e.g. Weixin's circuit breaker fails a send closed with "...cooldown active for 12.3s" +# (``gateway/platforms/weixin.py``). The cooldown length is still embedded in the text, so it is worth +# extracting precisely rather than falling back to the generic default every time. +_COOLDOWN_WAIT_RE = re.compile(r"cooldown active for\s+(\d+(?:\.\d+)?)s", re.IGNORECASE) + def _raw_flood_wait(text: str) -> Optional[float]: - """Seconds asked for by a flood error still carrying the platform's own wording, else ``None``.""" - match = _RAW_FLOOD_RE.search(text or "") + """Seconds asked for by a flood error still carrying the platform's own wording (PTB's "flood + control exceeded... retry in N seconds", or an adapter's own "cooldown active for Ns"), else + ``None``.""" + match = _RAW_FLOOD_RE.search(text or "") or _COOLDOWN_WAIT_RE.search(text or "") if not match: return None try: @@ -86,11 +94,34 @@ def _raw_flood_wait(text: str) -> Optional[float]: return None +def _classified_rate_limited(text: str) -> bool: + """Whether ``text`` matches the platform-neutral "rate_limited" send-error classification + (``gateway.platforms.base.classify_send_error``), so any adapter's rate-limit wording is + recognised here too and not just Telegram's. Imported locally (this module is loaded well before + ``gateway.platforms.base`` in some paths, e.g. CLI commands that only touch the ledger) and + best-effort like everything else in this module.""" + if not text: + return False + try: + from gateway.platforms.base import classify_send_error + except Exception: + return False + try: + return classify_send_error(None, text) == "rate_limited" + except Exception: + return False + + def is_flood_error(error: Any) -> bool: - """True for a flood refusal: the adapters' fail-closed ``flood_control:`` result, or a - row still carrying the platform's own flood wording (see ``_RAW_FLOOD_RE``).""" + """True for a flood refusal: the adapters' fail-closed ``flood_control:`` result, a row + still carrying the platform's own flood/cooldown wording (see ``_RAW_FLOOD_RE``), or any other + error text the platform-neutral classifier recognises as a rate limit (see + ``_classified_rate_limited``) — so an adapter that surfaces a rate limit without either of the + two specific shapes above (e.g. Weixin's bare ``RuntimeError``) still gets a timed redelivery + instead of being treated as an ordinary failure and retried immediately inside its own cooldown.""" text = str(error or "").strip().lower() - return text.startswith(FLOOD_ERROR_PREFIX) or _raw_flood_wait(text) is not None + return (text.startswith(FLOOD_ERROR_PREFIX) or _raw_flood_wait(text) is not None + or _classified_rate_limited(text)) def flood_wait_seconds(error: Any, default: float = FLOOD_RETRY_DEFAULT_SECONDS) -> float: diff --git a/tests/gateway/test_delivery_flood_invariants.py b/tests/gateway/test_delivery_flood_invariants.py index fb0e2a25cf13e..4857e0fee744e 100644 --- a/tests/gateway/test_delivery_flood_invariants.py +++ b/tests/gateway/test_delivery_flood_invariants.py @@ -5,11 +5,17 @@ from gateway import delivery_ledger as dl +# Weixin's circuit breaker fails a send closed with this shape (gateway/platforms/weixin.py) — a bare +# error string with neither the ``flood_control:`` prefix adapters normally use nor PTB's raw +# "Flood control exceeded... retry in N seconds" wording, but the same platform-neutral "rate_limited" +# classification (gateway.platforms.base.classify_send_error) that gates the in-attempt send backoff. +WEIXIN_RATE_LIMIT_ERROR = 'iLink sendmessage rate limited; cooldown active for 12.3s' -def record(oid, *, profile=None): - dl.record_obligation(obligation_id=oid, session_key='session-' + oid, platform='telegram', + +def record(oid, *, profile=None, platform='telegram', error='flood_control:185'): + dl.record_obligation(obligation_id=oid, session_key='session-' + oid, platform=platform, chat_id='123', thread_id='77', content='.' * 3000, adapter_profile=profile) - dl.mark_failed(oid, 'flood_control:185') + dl.mark_failed(oid, error) def read(oid): @@ -59,3 +65,66 @@ def test_boot_adopts_waiting_rows_without_spending_or_losing_deadline(): claimed = dl.sweep_failed_for_runtime('telegram', now=original['updated_at'] + 186) assert len(claimed) == 1 and claimed[0]['needs_marker'] assert read('boot')['state'] == 'attempting' and read('boot')['last_error'] is None + + +def test_weixin_rate_limit_text_classifies_as_flood_and_extracts_its_own_wait(): + """``is_flood_error``/``flood_wait_seconds`` must recognise Weixin's own rate-limit wording (no + ``flood_control:`` prefix, no PTB text) via the platform-neutral classifier, and read the cooldown + seconds Weixin itself embeds rather than falling back to the generic default.""" + assert dl.is_flood_error(WEIXIN_RATE_LIMIT_ERROR) + assert dl.flood_wait_seconds(WEIXIN_RATE_LIMIT_ERROR) == 12.3 + # An ordinary Weixin failure (not rate-limited) must still read as an ordinary failure. + assert not dl.is_flood_error('iLink sendmessage error: ret=1 errcode=2 errmsg=unknown error') + + +def test_weixin_rate_limit_arms_timed_redelivery_not_immediate_retry(): + """The ledger must treat a Weixin rate-limit failure exactly like a Telegram flood failure: adopted + without spending an attempt while still inside the wait, then claimable once it has passed — never + claimed as an ordinary failure that spends an attempt inside the still-active cooldown.""" + record('weixin-due', platform='weixin', error=WEIXIN_RATE_LIMIT_ERROR) + stamp = read('weixin-due')['updated_at'] + # Still inside Weixin's own 12.3s cooldown: the runtime sweep must not claim it yet. + assert dl.sweep_failed_for_runtime('weixin', now=stamp + 11) == [] + assert read('weixin-due')['attempts'] == 0 + # Past the cooldown: claimable, carries a marker (rate-limit wording), and clears the stale error. + claimed = dl.sweep_failed_for_runtime('weixin', now=stamp + 13) + assert [r['obligation_id'] for r in claimed] == ['weixin-due'] + assert claimed[0]['needs_marker'] and 'rate limit' in claimed[0]['marker'] + assert read('weixin-due')['last_error'] is None + + +def test_weixin_rate_limit_boot_adoption_does_not_spend_attempt(): + """A dead-owner boot sweep must adopt a still-cooling-down Weixin rate-limit row (no attempt spent, + deadline preserved) exactly as it does for Telegram's ``flood_control:`` rows — not treat it as an + ordinary failed row and immediately spend a redelivery attempt inside the cooldown.""" + record('weixin-boot', platform='weixin', error=WEIXIN_RATE_LIMIT_ERROR) + original = read('weixin-boot') + with sqlite3.connect(dl._db_path()) as conn: + conn.execute("UPDATE delivery_obligations SET owner_pid=NULL, owner_started_at=NULL, adapter_profile=NULL") + claimed = dl.sweep_recoverable(now=original['updated_at'] + 5, + deliverable_targets={('weixin', None)}) + assert len(claimed) == 1 and claimed[0].get('adopted') + adopted = read('weixin-boot') + assert adopted['owner_pid'] == os.getpid() + assert adopted['adapter_profile'] == 'default' + assert adopted['attempts'] == 0 and adopted['updated_at'] == original['updated_at'] + assert adopted['last_error'] == WEIXIN_RATE_LIMIT_ERROR + # A second boot-sweep pass finds this process still the live owner: nothing more to adopt. + assert dl.sweep_recoverable(now=original['updated_at'] + 6) == [] + # Still inside the cooldown: the runtime sweep (this process claiming its own adopted row) must wait. + assert dl.sweep_failed_for_runtime('weixin', now=original['updated_at'] + 11) == [] + # Past the deadline: claimed, marker set, error cleared. + claimed = dl.sweep_failed_for_runtime('weixin', now=original['updated_at'] + 13) + assert len(claimed) == 1 and claimed[0]['needs_marker'] + assert read('weixin-boot')['state'] == 'attempting' and read('weixin-boot')['last_error'] is None + + +def test_telegram_flood_detection_unaffected_by_broader_classifier(): + """Widening ``is_flood_error`` to also consult the platform-neutral classifier must not change + Telegram's own, already-correct behavior for its two established shapes.""" + assert dl.is_flood_error('flood_control:185') + assert dl.is_flood_error('Flood control exceeded. Retry in 185 seconds') + assert dl.flood_wait_seconds('flood_control:185') == 185.0 + assert dl.flood_wait_seconds('Flood control exceeded. Retry in 185 seconds') == 185.0 + # A permanent Telegram rejection must never be mistaken for a flood. + assert not dl.is_flood_error('Forbidden: bot was blocked by the user')