Building a Reliable Message Queue for High-Volume Bot Traffic

Running bots at any meaningful scale forces you to think carefully about message dispatch architecture. It is easy to build something that works fine at low volume and falls apart completely when traffic increases. We went through exactly that transition with the group automation bots we run, and the lessons from rebuilding the dispatch layer apply to pretty much any system that needs to send messages reliably under variable load.

This post covers the architecture we ended up with, the reasoning behind each decision, and what the implementation looks like. The specific platform does not really matter for most of what follows: the problems of rate limiting, prioritization, and reliable delivery under load are common to any high-volume messaging system.

Why the original architecture failed

Our initial bot architecture was simple. One persistent connection per group, messages dispatched directly when the bot needed to send something, a basic retry on failure. This is the obvious first implementation and it works fine when message volume is low and traffic is predictable.

The problems appeared during traffic spikes. Groups would have periods of high activity where many users were interacting simultaneously, generating bursts of outgoing messages from the bot. The direct dispatch approach handled this poorly for a few reasons.

Bursts would trigger rate limit responses from the messaging API. Our retry logic would then cause multiple concurrent retries, which would hit more rate limits, creating a feedback loop that made things worse rather than better.

There was no prioritization. An automated welcome message for a new user would sit in the same position as an admin command response. Admin commands should feel instant. Welcome messages can wait a few seconds.

Failure handling was primitive. When a message failed to send, we either dropped it or retried indefinitely. Neither was the right behavior for all situations. Some failures should retry, some should be dropped, and some should alert and drop.

The system needed a proper queued dispatch layer with rate limiting, prioritization, and sensible failure semantics.

Token bucket rate limiting

The first component we rebuilt was the rate limiting layer. A token bucket is the right abstraction for this problem. The concept is straightforward: there is a bucket that holds tokens up to a maximum capacity. Tokens refill at a constant rate. Each message dispatch consumes one token. If the bucket is empty, dispatch waits until a token is available.

This naturally smooths out burst traffic. A sudden spike of outgoing messages drains the token bucket, and subsequent messages queue up and drain at the refill rate. The maximum burst is controlled by bucket capacity, and sustained throughput is controlled by the refill rate.

    import asyncio
    import time

    class TokenBucket:
        """
        Token bucket rate limiter.
        
        rate: tokens added per second
        capacity: maximum tokens the bucket can hold
        """

        def __init__(self, rate: float, capacity: int):
            self.rate = rate
            self.capacity = capacity
            self.tokens = float(capacity)
            self.last_refill = time.monotonic()
            self._lock = asyncio.Lock()

        async def acquire(self, tokens: float = 1.0) -> None:
            async with self._lock:
                while True:
                    self._refill()
                    if self.tokens >= tokens:
                        self.tokens -= tokens
                        return
                    deficit = tokens - self.tokens
                    wait_time = deficit / self.rate
                    await asyncio.sleep(wait_time)

        def _refill(self) -> None:
            now = time.monotonic()
            elapsed = now - self.last_refill
            self.tokens = min(float(self.capacity), self.tokens + elapsed * self.rate)
            self.last_refill = now

        @property
        def available_tokens(self) -> float:
            self._refill()
            return self.tokens

One detail worth emphasizing: we set our rate to around 75 percent of the actual API limit rather than the maximum. This headroom accounts for timing imprecision in Python's sleep functions and means we almost never hit a rate limit response in practice. The cost is slightly lower peak throughput, which is a good tradeoff for the stability it provides.

Priority queue

The second component was message prioritization. We implemented a two-tier system: high priority and normal priority. The dispatcher always drains high-priority messages first, with one exception: if high-priority messages have been draining continuously for more than a few seconds, we interleave a normal-priority message to prevent starvation.

    import asyncio
    import time
    from dataclasses import dataclass, field
    from enum import IntEnum
    from typing import Any

    class Priority(IntEnum):
        HIGH = 0
        NORMAL = 1

    @dataclass(order=True)
    class QueuedMessage:
        priority: Priority
        sequence: int = field(compare=True)
        payload: Any = field(compare=False)
        enqueued_at: float = field(
            compare=False,
            default_factory=time.monotonic,
        )

    class PriorityMessageQueue:
        def __init__(self):
            self._queue: asyncio.PriorityQueue = asyncio.PriorityQueue()
            self._counter = 0

        async def put(self, payload: Any, priority: Priority = Priority.NORMAL) -> None:
            self._counter += 1
            msg = QueuedMessage(
                priority=priority,
                sequence=self._counter,
                payload=payload,
            )
            await self._queue.put(msg)

        async def get(self) -> QueuedMessage:
            return await self._queue.get()

        def task_done(self) -> None:
            self._queue.task_done()

        @property
        def size(self) -> int:
            return self._queue.qsize()

The sequence counter as a tiebreaker within the same priority level is important. Without it, Python's heap comparison might fall through to comparing the payload field, which might not be comparable. The counter guarantees a stable ordering: first by priority, then by insertion order.

The dispatcher

The dispatcher connects the queue and rate limiter. It runs as a background coroutine that continuously pulls messages from the queue, acquires a rate limit token, and calls the actual send function.

    import asyncio
    import logging
    from typing import Callable, Awaitable

    logger = logging.getLogger(__name__)

    class MessageDispatcher:
        def __init__(
            self,
            queue: PriorityMessageQueue,
            rate_limiter: TokenBucket,
            send_fn: Callable[[Any], Awaitable[bool]],
            max_retries: int = 3,
        ):
            self.queue = queue
            self.rate_limiter = rate_limiter
            self.send_fn = send_fn
            self.max_retries = max_retries
            self._running = False
            self._dispatched = 0
            self._failed = 0

        async def run(self) -> None:
            self._running = True
            logger.info("Dispatcher started")

            while self._running:
                try:
                    message = await asyncio.wait_for(
                        self.queue.get(),
                        timeout=1.0,
                    )
                except asyncio.TimeoutError:
                    continue

                try:
                    await self._dispatch(message)
                finally:
                    self.queue.task_done()

        async def _dispatch(self, message: QueuedMessage) -> None:
            await self.rate_limiter.acquire()

            backoff = 0.5
            for attempt in range(self.max_retries):
                try:
                    success = await self.send_fn(message.payload)
                    if success:
                        self._dispatched += 1
                        return
                except Exception as e:
                    logger.warning(
                        f"Send attempt {attempt + 1}/{self.max_retries} failed: {e}"
                    )

                if attempt < self.max_retries - 1:
                    await asyncio.sleep(backoff)
                    backoff *= 2

            self._failed += 1
            logger.error(
                f"Message failed after {self.max_retries} attempts: {message.payload!r}"
            )

        def stop(self) -> None:
            self._running = False

        @property
        def stats(self) -> dict:
            return {
                "dispatched": self._dispatched,
                "failed": self._failed,
                "queue_depth": self.queue.size,
                "rate_limiter_tokens": self.rate_limiter.available_tokens,
            }

The retry logic uses exponential backoff: half a second, then one second, then two seconds. Messages that exhaust all retries are logged and dropped. This is intentional: a persistently failing message should not block all subsequent traffic. Depending on your use case you might want to put failed messages in a dead letter queue for inspection rather than dropping them.

Putting it together

Here is how the full system looks in practice:

    import asyncio

    async def main():
        queue = PriorityMessageQueue()

        # Set rate to 75% of actual API limit for headroom
        rate_limiter = TokenBucket(rate=7.5, capacity=15)

        async def send_message(payload: dict) -> bool:
            # Your actual send logic here
            # Return True on success, raise or return False on failure
            response = await bot_client.send(
                group_id=payload["group_id"],
                text=payload["text"],
            )
            return response.ok

        dispatcher = MessageDispatcher(
            queue=queue,
            rate_limiter=rate_limiter,
            send_fn=send_message,
            max_retries=3,
        )

        # Run dispatcher in background
        dispatcher_task = asyncio.create_task(dispatcher.run())

        # Your bot logic enqueues messages like this
        await queue.put(
            {"group_id": "abc123", "text": "Welcome to the group!"},
            priority=Priority.NORMAL,
        )
        await queue.put(
            {"group_id": "abc123", "text": "Admin command executed."},
            priority=Priority.HIGH,
        )

        # Keep running until stopped
        await dispatcher_task

Observability

After the initial deployment, we added lightweight metrics to the dispatcher. The stats property in the dispatcher class above exposes the basics: messages dispatched, messages failed, current queue depth, and available rate limiter tokens.

We expose these via a simple HTTP endpoint that our monitoring scrapes every minute:

    from aiohttp import web

    async def health_handler(request: web.Request) -> web.Response:
        dispatcher = request.app["dispatcher"]
        return web.json_response({
            "status": "ok",
            **dispatcher.stats,
        })

    app = web.Application()
    app.router.add_get("/health", health_handler)

Queue depth is the most actionable metric. If it grows consistently over time, your dispatch rate is lower than your enqueue rate. That means either you need to reduce message volume, increase the rate limit if there is headroom, or scale out to multiple dispatcher instances.

The failure rate is a signal that something is wrong in the send function itself. A spike in failures usually means either network issues or the downstream API is having problems.

Results

After deploying this architecture, the incidents we were previously having during traffic spikes essentially stopped. The token bucket absorbed bursts cleanly without triggering rate limit responses, and the priority queue meant admin commands always went through promptly regardless of queue depth.

The most surprising benefit was how much easier debugging became. With the old direct dispatch approach, failures were scattered and hard to correlate. With the queued approach, everything flows through one place and the stats endpoint gives a clear picture of system health at any moment.

What we would add next

Persistence is the obvious next step. If the process restarts, messages in the queue are currently lost. For our use case the messages are low enough value that this is acceptable, but for any higher-stakes application you would want to back the queue with Redis or a database.

A dead letter queue would also be valuable. Right now failed messages are logged and dropped. A dead letter queue would let you inspect and replay them without searching through logs, which is much more operationally friendly.

Both additions are straightforward given this architecture. The core queue and rate limiting logic does not change, you just add a persistence layer underneath and a dead letter queue alongside.