Queen MQ - Python Client

---

What is Queen MQ?

Queen MQ is a PostgreSQL-backed message queue system with a powerful feature set:

• FIFO Partitions - Unlimited ordered partitions within queues
• Consumer Groups - Kafka-style consumer groups for scalability
• Flexible Semantics - Exactly-once, at-least-once, and at-most-once delivery
• Transactions - Atomic operations across push and ack
• High Performance — 104K msg/s push, 165K msg/s fan-out with consumer groups on a single 32-core node (benchmarks)
• Subscription Modes - Process from beginning, new messages only, or from timestamp
• Dead Letter Queue - Automatic failure handling and monitoring
• Message Tracing - Debug distributed workflows with trace timelines
• Client-Side Buffering - 10x-100x throughput boost for high-volume pushes
• Real-time Streaming - Windowed aggregation and processing

This client provides a fluent, async/await API for Python applications.

---

Installation

pip install queen-mq

Requirements: Python 3.8+

---

Quick Start

import asyncio
from queen import Queen

async def process(message):
    print('Processing:', message['data'])
    # Auto-ack on success, auto-retry on error

async def main():
    # Connect to Queen server
    async with Queen('http://localhost:6632') as queen:
        # Create a queue
        await queen.queue('tasks').create()

        # Push messages
        await queen.queue('tasks').push([
            {'data': {'task': 'send-email', 'to': 'alice@example.com'}}
        ])

        # Consume messages (handler is a regular async function)
        await queen.queue('tasks').consume(process)

asyncio.run(main())

---

Core Concepts

Queues

Logical containers for messages with configurable settings:

await queen.queue('orders').config({
    'leaseTime': 300,          # 5 minutes
    'retryLimit': 3,
    'priority': 5,
    'encryptionEnabled': False
}).create()

Partitions

Ordered lanes within a queue:

# All messages for user-123 are processed in order
await queen.queue('user-events').partition('user-123').push([
    {'data': {'event': 'login'}},
    {'data': {'event': 'view-page'}},
    {'data': {'event': 'logout'}}
])

Consumer Groups

Multiple consumers sharing work:

async def send_handler(msg):
    await send_email(msg['data'])

async def analytics_handler(msg):
    await log_metrics(msg['data'])

# Worker 1 & 2 share the load
await queen.queue('emails').group('processors').consume(send_handler)

# Separate group processes same messages independently
await queen.queue('emails').group('analytics').consume(analytics_handler)

Subscription Modes

Control whether consumer groups process historical messages:

async def handler(msg):
    ...

# Default: Process ALL messages (including backlog)
await queen.queue('events').group('batch-analytics').consume(handler)

# Skip history, only new messages
await queen.queue('events').group('realtime-monitor').subscription_mode('new').consume(handler)

# Start from specific timestamp
await queen.queue('events').group('replay').subscription_from('2025-10-28T10:00:00.000Z').consume(handler)

---

Connection Options

Single Server

queen = Queen('http://localhost:6632')

Multiple Servers (High Availability)

queen = Queen(['http://server1:6632', 'http://server2:6632'])

Full Configuration

queen = Queen({
    'urls': ['http://server1:6632', 'http://server2:6632'],
    'timeout_millis': 30000,
    'retry_attempts': 3,
    'load_balancing_strategy': 'affinity',  # or 'round-robin', 'session'
    'enable_failover': True
})

---

Basic Usage Patterns

Push Messages

# Simple push
await queen.queue('tasks').push([
    {'data': {'job': 'resize-image', 'imageId': 123}}
])

# With partition
await queen.queue('tasks').partition('tenant-456').push([
    {'data': {'action': 'process'}}
])

# With custom transaction ID (for exactly-once)
await queen.queue('tasks').push([
    {'transactionId': 'unique-id-123', 'data': {'value': 42}}
])

Consume Messages (Long-Running Workers)

# Single message processing (batch=1, default)
# Handler receives a single message
async def single_handler(message):
    await process_task(message['data'])
    # Auto-ack on success, auto-retry on error

await queen.queue('tasks').concurrency(10).consume(single_handler)

# Batch processing (batch>1)
# Handler receives an array of messages
async def batch_handler(messages):
    for message in messages:
        await process_task(message['data'])

await queen.queue('tasks').batch(20).concurrency(5).consume(batch_handler)

# Process with limit and stop
await queen.queue('tasks').limit(100).consume(single_handler)

Pop Messages (On-Demand Processing)

# Grab messages manually
messages = await queen.queue('tasks').batch(10).wait(True).pop()

# Manual acknowledgment
for message in messages:
    try:
        await process_message(message['data'])
        await queen.ack(message, True)  # Success
    except Exception as error:
        await queen.ack(message, False)  # Retry

Multi-Partition Pop (Drain Many Partitions Per Call)

# One round-trip drains up to 200 messages spread across up to 50 partitions.
# batch(200) is the GLOBAL cap on total messages; partitions(50) is the
# hard cap on partitions claimed. All claimed partitions share one leaseId
# — a single renew() call extends every partition's lease atomically.
messages = await (queen.queue('events')
                  .batch(200)
                  .partitions(50)
                  .wait(True)
                  .pop())

# Each message carries its own partition info (per-message partitionId,
# partition name, leaseId, consumerGroup) — ACK and renew always work
# message-by-message regardless of how many partitions the batch spans.
for m in messages:
    print(f"from {m['partition']}:", m['data'])

# Same builder works on .consume() for long-running workers
async def handler(msgs):
    for m in msgs:
        await process(m['data'])

await (queen.queue('events')
       .batch(100)
       .partitions(8)
       .consume(handler))

When to use: queues with many partitions where each partition only has a handful of new messages per polling interval (per-customer event streams, per-tenant work queues, per-device telemetry). Reduces network round-trips from O(P) to O(P / N) while preserving per-partition FIFO ordering.

When not to use: few partitions, or each one busy enough to fill batch(B) on its own. Leaving .partitions() unset hands the sweep width to the broker (see Pop Autopilot below); .partitions(1) pins the legacy single-partition behaviour.

.partitions(N) only applies to wildcard pops; specifying .partition('name') ignores the cap.

Pop Autopilot (Let the Broker Size the Pop)

Since 1.2, batch and partitions that you do not set are chosen by the broker, per pop, from state the client cannot see: how many partitions of the group are ready, how old their oldest ready message is, how fast messages are arriving. The knobs you do set are never touched.

# Both knobs are the broker's: it picks the sweep width and the budget.
await (queen.queue('events').group('workers')
       .consume(handler))

# One knob pinned, one delegated: this consumer stays on one partition
# forever, and the broker sizes the batch for it.
await (queen.queue('events').group('workers').partitions(1)
       .consume(handler))

The request carries autopilot=true and simply omits the delegated knobs. Setting both leaves nothing to decide, so nothing changes on the wire at all.

Two ways to switch it off, both restoring the previous client-side defaults (batch 1, partitions 1) byte for byte:

await queen.queue('events').autopilot(False).consume(handler)

QUEEN_SDK_POP_AUTOPILOT=off   # whole process, read once at client creation

What the broker chose rides back on the response and is there for the reading, along with an optional pacing hint the consume loop honours in place of its own delay between empty polls:

res = await queen.queue('events').group('workers').pop_result()
if res.autopilot:
    print(f'{res.autopilot.partitions} partitions, batch {res.autopilot.batch}, '
          f'poll again in {res.autopilot.wait_millis}ms')

Requires broker >= 1.2. An older broker ignores the parameter, so the omitted knobs take its defaults (batch 200, partitions 1) instead of the old client-side ones. That is a sizing difference and nothing else — no message is lost, reordered or duplicated — so unlike conflation it degrades silently and on purpose. Pin the values explicitly, or turn autopilot off, if you need the old numbers against an old broker.

Transactions (Atomic Operations)

# Pop from queue A
messages = await queen.queue('input').pop()

# Atomically: ack input AND push output
await (queen.transaction()
    .ack(messages[0])
    .queue('output')
    .push([{'data': processed_result}])
    .commit())

Client-Side Buffering (High Throughput)

# Buffer messages locally, batch to server
for i in range(10000):
    await queen.queue('events').buffer({'message_count': 500, 'time_millis': 1000}).push([
        {'data': {'id': i}}
    ])

# Flush remaining buffered messages
await queen.flush_all_buffers()

# Result: 10x-100x faster than individual pushes

Buffer options:

Option	Default	Meaning
message_count	100	Flush once this many messages are buffered (also the batch size of one POST)
time_millis	1000	Flush this long after the first buffered message
max_size	4 * message_count	Backpressure bound: push() BLOCKS while this many messages are waiting
retry_delay_millis	250	Wait before retrying a batch whose POST failed

The buffer is bounded and push() blocks at the bound, so a producer that outruns the flush pipeline is slowed down to the drain rate instead of growing the process until it dies with every unflushed message inside it. max_size of 0 means the default bound, not unbounded. A buffered push() therefore either returns {'buffered': True, ...} or raises: cancel it (asyncio.wait_for) if your producer needs a deadline, and treat the exception as "not buffered".

A batch whose POST fails goes back to the front of the buffer and is retried after retry_delay_millis, in order, until it lands. Nothing is dropped, so a broker outage shows up as blocked producers rather than as missing messages. flush_all_buffers() keeps retrying while the broker is down (cancel it if you need a bounded wait); close() already bounds it at 30 seconds and logs how many messages were left unsent, so a shutdown ends loudly instead of hanging.

Dead Letter Queue

# Enable DLQ on queue
await queen.queue('risky').config({'retryLimit': 3, 'dlqAfterMaxRetries': True}).create()

# Query failed messages
dlq = await queen.queue('risky').dlq().limit(10).get()

print(f"Found {dlq['total']} failed messages")
for msg in dlq['messages']:
    print('Error:', msg.get('errorMessage'))

Message Tracing

async def order_handler(msg):
    order_id = msg['data']['orderId']

    # Record trace with name for cross-service correlation
    await msg['trace']({
        'traceName': f"order-{order_id}",
        'eventType': 'info',
        'data': {'text': 'Order processing started'}
    })

    await process_order(msg['data'])

    await msg['trace']({
        'traceName': f"order-{order_id}",
        'eventType': 'processing',
        'data': {'text': 'Order completed', 'total': msg['data']['total']}
    })

await queen.queue('orders').consume(order_handler)

# View traces in webapp: Traces → Search "order-12345"

---

Key/Value State and Timers

Both surfaces are always there. There is no flag to turn on before using them and nothing to probe first: any broker you can push to can also hold state and schedule timers.

An operator can still pause either one during an incident, in which case a call raises KvError / TimerError with status == 503, code kv_disabled or timers_disabled, and a Retry-After in retry_after_seconds. Inside a transaction the same pause is a 403 instead, so a bundle fails fast rather than spinning with messages in hand. Timer cancels are never paused.

Key/Value

from datetime import timedelta

# An expiry is MANDATORY on every write: exactly one of ttl_seconds, ttl,
# until or forever=True. A put never inherits the previous expiry: that is
# the fastest way to make a marker immortal.
await queen.kv.put('orders', 'order:9f1', {'state': 'held'}, ttl_seconds=60)
await queen.kv.put('orders', 'order:9f1', {'state': 'held'}, ttl=timedelta(minutes=1))

got = await queen.kv.get('orders', 'order:9f1')
if got:                       # follows `found`, not the value
    print(got['value'], got['version'])

# "Did I win?", in one call, one boolean. This is the idempotency marker.
if await queen.kv.once('dedup', f"evt:{event_id}", ttl_seconds=86400):
    await do_the_external_effect()

# Optimistic lock. expect=0 means "must not exist"; expect=N is a pure update
# that creates nothing when it matches no row.
res = await queen.kv.put('orders', 'order:9f1', {'state': 'shipped'},
                         ttl_seconds=60, expect=got['version'])
if not res:
    print(res['reason'])      # 'version' | 'absent' | 'exists' | 'limit' | 'type'

# Rate limiting without a CAS loop. With max, `applied` IS the admission
# decision: nothing saturates, nothing truncates, a refusal spends no budget.
allowed = await queen.kv.incr('quota', f"{customer}:{hour}", delta=1, max=1000, ttl_seconds=3600)
if not allowed:
    raise TooManyRequests()

rows = await queen.kv.list_all('saga', 'order:9f1:')   # follows nextAfter

A write that did not apply is not an error. applied: false answers HTTP 200 with the current value and version, so the loser needs no second round trip. Results are falsy when they did not apply, so if await queen.kv.delete(...) reads the verdict rather than "an object came back".

Read-modify-write across two calls is safe only when the KV key derives from the partition key. Otherwise the lanes do not serialise it for you: use incr, or carry expect.

put_if_absent plus a TTL is not a distributed lock. A lock that expires is not revoked: the old holder keeps working, it simply no longer has the row. Carry your version as expect on every later write so a lapsed holder fails with reason: "version" instead of overwriting the new one.

Timers

res = await queen.timers.schedule('orders', 'order:9f1:expire',
                                  {'orderId': '9f1'}, delay_ms=30_000)
res['txn'], res['messageId'], res['deliverAt']

await (queen.timer('orders')
            .key('order:9f1:expire')
            .payload({'orderId': '9f1'})
            .after(timedelta(minutes=30))
            .schedule())

cancelled = await queen.timers.cancel('orders', 'order:9f1:expire', txn=res['txn'])

Durations that can be sub-second are in milliseconds (delay_ms), the ones that cannot are in seconds (ttl_seconds). Only relative delays exist: there is one clock and it is the database's. A delay in the past is legal and fires on the first cycle.

deliverAt is "not before", never "exactly at".

absent means "no longer pending" and may mean ALREADY DELIVERED. There is no tombstone: a delivered timer has no row. The response echoes the txn back so the authority, the log, can be consulted without a second API call. A saga that cancels a compensation timer must therefore have its compensation consumer re-check the saga state before compensating, because the cancel may have arrived 5 ms after the fire.

Use queen.timers.cancel(...), not a cancel op inside a batch: the DELETE route it takes is the one that is never blocked by a quota. A tenant that cannot cancel keeps producing messages it cannot stop.

Inside a transaction

The transaction is the primary fence; expect is only the secondary assertion. A state write that shares the transaction with its ack is undone when an expired lease makes the ack fail, which a CAS cannot do.

result = await (queen.transaction()
    .once('dedup', f"evt:{event_id}", ttl_seconds=86400)   # the gate, first
    .queue('orders').push([{'data': {...}}])
    .ack(message)
    .timer('orders').key(f"order:{oid}:expire").payload({'oid': oid}).after_ms(30_000).schedule()
    .commit())

if not result:
    # RETURNED, not raised: a lost gate is the expected outcome of a legitimate
    # redelivery, so it stays out of your retry policy and your error metrics.
    assert result['reason'] == 'kv_precondition'
    print(result['failedIndex'], result['kvReason'], result['version'], result['value'])

Everything else still raises. get_prefix is not available inside a transaction: its cost is not bounded by the caller.

---

API Reference

Queue Operations

# Create
await queen.queue('my-queue').create()
await queen.queue('my-queue').config({'priority': 5}).create()

# Delete
await queen.queue('my-queue').delete()

Push

await queen.queue('q').push([{'data': {'value': 1}}])
await queen.queue('q').partition('p1').push([{'data': {'value': 1}}])
await queen.queue('q').buffer({'message_count': 100, 'time_millis': 1000}).push([...])

Pop

msgs = await queen.queue('q').pop()                            # broker-sized (see Pop Autopilot)
msgs = await queen.queue('q').batch(10).pop()
msgs = await queen.queue('q').batch(10).wait(True).pop()
msgs = await queen.queue('q').batch(200).partitions(50).pop()  # multi-partition pop
res = await queen.queue('q').pop_result()                      # + what the broker chose

Consume

# batch=1 (default): handler receives single message
async def single_handler(msg):
    ...

await queen.queue('q').consume(single_handler)

# batch>1: handler receives array of messages
async def batch_handler(msgs):
    ...

await queen.queue('q').batch(10).consume(batch_handler)

# Other options
await queen.queue('q').limit(10).consume(single_handler)
await queen.queue('q').concurrency(5).consume(single_handler)
await queen.queue('q').group('my-group').consume(single_handler)

Acknowledgment

await queen.ack(message, True)   # Success
await queen.ack(message, False)  # Retry
await queen.ack(message, False, {'error': 'reason'})
await queen.ack([msg1, msg2], True)  # Batch ack

Transactions

await (queen.transaction()
    .ack(message)
    .queue('output')
    .push([{'data': {'result': 'processed'}}])
    .commit())

Lease Renewal

await queen.renew(message)
await queen.renew([msg1, msg2, msg3])

async def handler(msg):
    ...
await queen.queue('q').renew_lease(True, 60000).consume(handler)

Buffering

await queen.flush_all_buffers()
await queen.queue('q').flush_buffer()
stats = queen.get_buffer_stats()

Dead Letter Queue

dlq = await queen.queue('q').dlq().limit(10).get()
dlq = await queen.queue('q').dlq('consumer-group').limit(10).get()

Shutdown

await queen.close()  # Flush buffers and close connections

---

Configuration Defaults

Client Defaults

{
    'timeout_millis': 30000,
    'retry_attempts': 3,
    'retry_delay_millis': 1000,
    'load_balancing_strategy': 'affinity',
    'enable_failover': True
}

Queue Defaults

{
    'leaseTime': 300,           # 5 minutes
    'retryLimit': 3,
    'priority': 0,
    'delayedProcessing': 0,
    'windowBuffer': 0,
    'maxSize': 0,              # Unlimited
    'retentionSeconds': 0,     # Keep forever
    'encryptionEnabled': False
}

Consume Defaults

{
    'concurrency': 1,
    'batch': 1,                # autopilot OFF only -- unset means the broker sizes it
    'max_partitions': 1,       # autopilot OFF only -- unset means the broker sizes it
    'auto_ack': True,
    'wait': True,              # Long polling
    'timeout_millis': 30000,
    'limit': None,             # Run forever
    'renew_lease': False
}

batch and max_partitions are the autopilot-off defaults: with autopilot on (the default) a knob you never set is not defaulted at all, it is delegated to the broker. These values are what comes back with .autopilot(False) or QUEEN_SDK_POP_AUTOPILOT=off.

---

Logging

Enable detailed logging for debugging:

export QUEEN_CLIENT_LOG=true
python your_app.py

Example output:

[2025-10-28T10:30:45.123Z] [INFO] [Queen.constructor] {"status":"initialized","urls":1}
[2025-10-28T10:30:45.234Z] [INFO] [QueueBuilder.push] {"queue":"tasks","partition":"Default","count":5}

---

Type Hints

Full type hints included for IDE support:

from queen import Queen, Message
from typing import Dict, Any

queen: Queen = Queen('http://localhost:6632')

async def handler(message: Message) -> None:
    data: Dict[str, Any] = message['data']
    print(data)

await queen.queue('orders').consume(handler)

---

Best Practices

1. ✅ Use consume() for workers - Simpler API, handles retries automatically
2. ✅ Use pop() for control - When you need precise control over acking
3. ✅ Buffer for speed - Always use buffering when pushing many messages
4. ✅ Partitions for order - Use partitions when message order matters
5. ✅ Consumer groups for scale - Run multiple workers in the same group
6. ✅ Transactions for consistency - Use transactions for atomic operations
7. ✅ Enable DLQ - Always enable DLQ in production
8. ✅ Renew long leases - Use auto-renewal for long-running tasks
9. ✅ Graceful shutdown - Use async context manager or call queen.close()
10. ✅ Monitor DLQ - Regularly check for failed messages

📝 Important Notes

Handler Signatures:

• When batch=1 (default), handler receives a single message: async def handler(message): ...
• When batch>1, handler receives an array of messages: async def handler(messages): ...
• When each=True, always receives single messages regardless of batch size

---

Documentation

• Node.js Client — Node.js client documentation
• HTTP API Reference — raw HTTP endpoints
• Server Guide — server setup and configuration
• Architecture & internals — published architecture overview
• libqueen design notes — adaptive engine deep-dive

---

License

Apache 2.0 - See LICENSE

---

Support

• GitHub: queen-mq/queen
• Issues: GitHub Issues
• LinkedIn: Smartness