Bot Rate Limiting

Throttle outbound bot messages to messaging platforms like Telegram, Discord, and Slack

from praisonai.bots._rate_limit import RateLimiter

limiter = RateLimiter.for_platform("telegram")
await limiter.acquire(channel_id="telegram-chat-12345")

Bot Rate Limiting prevents messaging platform 429 errors by throttling outbound messages to channels, respecting platform-specific limits and per-channel delays.

Since PraisonAI #2578, the limiter also covers the proactive path — scheduled and agent-initiated sends now share one bucket per platform with the reply path. DeliveryRouter.deliver reuses adapter._rate_limiter if the adapter already carries one; otherwise it lazily calls RateLimiter.for_platform(platform).

Quick Start

1. Simple Usage

   Use the default rate limiter for general messaging bots.

   from praisonai.bots._rate_limit import RateLimiter
   
   # Default: 1 message/sec, burst 5, 1s per-channel delay
   limiter = RateLimiter()
   
   # Bot adapters call this before sending
   await limiter.acquire(channel_id="telegram-chat-12345")

2. Platform-Specific Configuration

   Use platform presets for optimal compliance with API limits.

   from praisonai.bots._rate_limit import RateLimiter
   
   # Single-process: fast in-memory bucket (unchanged behaviour)
   telegram_limiter = RateLimiter.for_platform("telegram")
   
   # Multi-worker: share ONE bucket across every gateway worker on this bot token
   from praisonai.bots import DeliveryControlStore
   shared = DeliveryControlStore("~/.praisonai/state/delivery_control.sqlite")
   telegram_limiter = RateLimiter.for_platform("telegram", store=shared)
   # scope defaults to the platform name — all "telegram" limiters on this store
   # share one ceiling. Pass scope="telegram:bot-<hash>" for multi-tenant isolation.

3. Custom Configuration

   Fine-tune rate limits for specific platform policies or custom requirements.

   from praisonai.bots._rate_limit import RateLimiter, RateLimitConfig
   
   limiter = RateLimiter(RateLimitConfig(
       messages_per_second=2.0,  # Global rate
       burst_size=10,            # Burst capacity
       per_channel_delay=1.5,    # Min delay per channel
   ))
   
   await limiter.acquire(channel_id="custom-channel-456")

How It Works

The rate limiter uses a two-phase approach:

Configuration Options

RateLimiter constructor kwargs (multi-worker)

Platform Presets

Memory Management

Per-channel state is tracked in an LRU cache capped at 4096 channels. If a bot serves more channels than that, the least-recently-used channels fall out of the cache and their per_channel_delay window resets. This bounds memory for long-running bots.

# Memory usage stays bounded even with many channels
for channel_id in range(10000):  # 10k channels
    await limiter.acquire(f"channel-{channel_id}")
    # Internal cache automatically evicts old entries at 4096 limit

Multi-worker (horizontally-scaled gateway)

A single-process gateway is served by the fast in-memory token bucket — RateLimiter() with no store= — and that’s the right default. But once you run N workers, each with its own RateLimiter, each worker keeps its own bucket and the platform sees N × messages_per_second. That’s the exact scenario a token-bucket limiter exists to prevent, and it produces 429s and temporary bans.

Pass a shared DeliveryControlStore so every worker draws from one SQLite-backed bucket:

from praisonai.bots import DeliveryControlStore
from praisonai.bots._rate_limit import RateLimiter

# One file, shared by every worker on the box (or on the shared volume)
store = DeliveryControlStore("~/.praisonai/state/delivery_control.sqlite")

limiter = RateLimiter.for_platform("telegram", store=store)
# scope defaults to "telegram" — matches every other worker calling for_platform("telegram", store=store)

The token reservation now happens inside a BEGIN IMMEDIATE transaction, so N workers racing on acquire() get atomic wait times summing to the configured ceiling. The sleep still happens outside the transaction — workers block only for their share of the wait, not for the transaction round-trip.

scope — sharing identity

scope is the shared-bucket identity. All limiters that must share one ceiling must use the same scope string. Two common patterns:

• Per platform (default): every worker running for_platform("telegram", store=store) uses scope="telegram" — one ceiling per platform per host.
• Per bot token (multi-tenant): hash the token so tenant A’s Telegram bot doesn’t share a bucket with tenant B’s Telegram bot: RateLimiter.for_platform("telegram", store=store, scope=f"telegram:{token_hash}").

Restart survival

The bucket state (tokens, last_refill, global_penalty_until, per-channel last_send, penalty_until) is persisted on every reservation. A worker restart resumes from the last recorded reservation anchor instead of refilling to burst_size, so a rolling restart doesn’t briefly triple the send rate.

penalise() and reset() are also shared

When any worker records a 429/Retry-After hint via penalise(), every other worker on the same scope immediately respects the same hold-off window. reset() clears the shared row.

DeliveryControlStore

The store is stdlib sqlite3-only (no extra dependency), uses WAL journaling, and can back both the rate limiter and the dead-target registry from one file.

from praisonai.bots import DeliveryControlStore

store = DeliveryControlStore("~/.praisonai/state/delivery_control.sqlite")
# Parent directory is created if missing.

Table shape (informational — you never write SQL yourself):

• rate_limit_state(scope, tokens, last_refill, global_penalty_until) — one row per scope.
• rate_limit_channel(scope, channel_id, last_send, penalty_until) — per-channel rows, capped at 4096 per scope with LRU eviction.
• dead_targets(platform, channel_id, reason, ts) — see Dead-Target Registry.

Concurrency Design

The global lock is held only long enough to reserve a token and update bookkeeping — the actual sleep happens outside the lock. Multiple channels can be rate-limited concurrently without serialising on one mutex.

Before PR #1870 (serialized):

# Old behavior: sleep INSIDE lock - channels wait in line
async with self._lock:
    # check tokens, check channel timing
    await asyncio.sleep(delay)  # BLOCKS other channels

After PR #1870 (concurrent):

# New behavior: sleep OUTSIDE lock - channels sleep in parallel
async with self._lock:
    # check tokens, check channel timing, compute delay
    pass  # lock released immediately
await asyncio.sleep(delay)  # Multiple channels sleep concurrently

Server-provided Retry-After

When a messaging platform explicitly tells the bot how long to wait, that hint takes precedence over the policy backoff.

Precedence order (highest first):

1. Server-mandated wait — extracted from the platform error:
2. Telegram: parameters.retry_after field (or .retry_after attr on RetryAfter exception)
3. HTTP channels (Slack / Discord / WhatsApp): Retry-After header — integer seconds or HTTP-date
4. Text fallback: retry after / retry_after: in the error message body
5. Policy exponential backoff — only used when the server provides no hint

The resilience layer (_resilience.deliver_with_retry and _delivery.deliver_with_retry) reads the server hint via server_retry_after(err) and sleeps for exactly that duration before the next attempt. The OutboundQueue also stores the hint and gates the next drain cycle accordingly.

Penalised Lanes

After a 429 from a messaging platform, RateLimiter.penalise(channel_id, seconds) widens the wait window for that channel — and the global window — so the next sends don’t immediately re-trip the limit.

from praisonai.bots._rate_limit import RateLimiter

limiter = RateLimiter.for_platform("telegram")

# When a 429 arrives, the delivery layer calls:
# limiter.penalise(channel_id="telegram-chat-123", seconds=30)
# — the channel's per_channel_delay grows for 30 s,
#   and the global bucket also absorbs the penalty.

penalise is called automatically by _delivery.deliver_with_retry via its optional rate_limiter= argument when a server-mandated Retry-After hint is detected. You can also call it manually if you observe 429s through other means.

Best Practices

# Good: use tested platform preset
limiter = RateLimiter.for_platform("discord")

# Risky: custom config might hit undocumented limits
limiter = RateLimiter(RateLimitConfig(messages_per_second=10.0))

from praisonai.bots import DeliveryControlStore
from praisonai.bots._rate_limit import RateLimiter

# Multi-worker gateway — every worker constructs its own limiter, but they share
# one bucket via the store on the shared volume.
store = DeliveryControlStore("~/.praisonai/state/delivery_control.sqlite")
telegram_limiter = RateLimiter.for_platform("telegram", store=store)

import logging
logging.getLogger("praisonai.bots._rate_limit").setLevel(logging.DEBUG)

# Log output:
# DEBUG:praisonai.bots._rate_limit:Rate limit: waiting 0.750s for channel telegram-chat-123

# WhatsApp allows rapid bursts then enforces stricter limits
whatsapp_limiter = RateLimiter(RateLimitConfig(
    messages_per_second=10.0,  # Sustained rate
    burst_size=50,             # Initial burst capacity
    per_channel_delay=0.1      # Quick per-channel recovery
))

Related