Expiration Policy

Overview

An expiration policy puts a hard time limit on how long a queue message may sit unconsumed. It solves a different problem than a dead-letter policy — this isn't about messages that fail processing, it's about messages that go stale: a price quote, a one-time code, a cache-invalidation signal, where late delivery is actively wrong, not just delayed. Instead of every consumer re-checking timestamps itself, the deadline lives on the message and the broker enforces it.

At the API level, expiration_in_seconds=5 attaches a per-message TTL when you build the QueueMessage, and the clock starts the moment the broker accepts it via send_queue_message, not when a consumer picks it up. Let the TTL elapse unconsumed and the broker silently removes it — a later poll just comes back empty, no error, no trace.

Gotchas: expiration is silent — no DLQ routing, no event, just a message that vanishes — so pair it with monitoring if you need visibility into how much work is being dropped. The timer starts at send time, not when a consumer picks up the work, so a message can expire mid-backlog even while a consumer is actively polling. And setting the TTL too short for your real consumer lag just turns ordinary slowness into silent data loss.

Prerequisites

• KubeMQ server running on localhost:50000
• Python SDK installed (pip install kubemq)

Code

"""Example: Expiration policy — send a message that expires after a timeout."""

from __future__ import annotations

import asyncio

from kubemq import AsyncQueuesClient
from kubemq import QueueMessage


async def main() -> None:
    async with AsyncQueuesClient(
        address="localhost:50000",
        client_id="python-queues-stream-expiration-policy-client",
    ) as client:
        # Send a message that expires in 5 seconds
        result = await client.send_queue_message(
            QueueMessage(
                channel="python-queues-stream.expiration-policy",
                body=b"message with expiration",
                expiration_in_seconds=5,
            )
        )
        print(f"Sent message with 5s expiration: {result}")

        # Wait for the message to expire
        print("Waiting 6 seconds for expiration...")
        await asyncio.sleep(6)

        # Try to receive — message should have expired
        response = await client.receive_queue_messages(
            channel="python-queues-stream.expiration-policy",
            max_messages=1,
            wait_timeout_seconds=2,
            auto_ack=True,
        )
        print(f"Received {len(response.messages)} messages (expected 0 — message expired)")


if __name__ == "__main__":
    asyncio.run(main())

How It Works

• QueueMessage.expiration_in_seconds=5 sets a TTL on the message; the broker automatically discards it if it has not been consumed within that window.
• The sample waits 6 seconds (past the TTL), then polls the queue and receives zero messages, confirming the expiration policy worked.
• Expiration is useful for time-sensitive events that lose relevance if not processed promptly (e.g., sensor readings, ephemeral tokens).
• Unlike delay, expiration removes the message from the queue entirely — there is no DLQ routing unless max_receive_queue is also set.

Related

• Pattern overview
• Python SDK Reference
• Send & Receive
• Ack All