Queue Topology

Queue Topology

Mantis uses a structured RabbitMQ topology for command dispatch and AI log analysis. Mandible declares the entire topology on startup; Thorax and the analysis workers are pure consumers.

Results and events are not queue-based

RabbitMQ carries command dispatch (an optional push nudge for listen-mode agents) and AI log analysis only. Execution results flow back to Mandible over gRPC (store_command_result), not a queue — there is no mantis.results exchange. Deployment lifecycle events are delivered to clients via Server-Sent Events (the /sse endpoints), not a mantis.events exchange.

Overview

Exchanges

Exchange	Type	Purpose
mantis.commands	Direct	Route commands to Thorax by mode and priority
mantis.analysis	Direct	Route AI log-analysis jobs to analysis workers
mantis.dlx	Fanout	Dead-letter exchange for failed messages

Command Exchange

The mantis.commands exchange routes commands based on target mode and priority:

mantis.commands (direct) — EVERY routing key binds to mantis.commands.thorax (the
primary consumer queue); priority keys ALSO bind to their priority queue:
├── command.thorax        -> mantis.commands.thorax
├── command.poll.high     -> mantis.commands.thorax + mantis.commands.priority.high
├── command.poll.normal   -> mantis.commands.thorax + mantis.commands.priority.normal
├── command.poll.low      -> mantis.commands.thorax + mantis.commands.priority.low
├── command.listen.high   -> mantis.commands.thorax + mantis.commands.priority.high
├── command.listen.normal -> mantis.commands.thorax + mantis.commands.priority.normal
└── command.listen.low    -> mantis.commands.thorax + mantis.commands.priority.low

Analysis Exchange

The mantis.analysis exchange routes AI log-analysis jobs to the analysis worker queue (mantis.analysis).

Results (gRPC, not a queue)

Execution results are not published to RabbitMQ. Thorax reports each command’s outcome to Mandible directly over gRPC (store_command_result), so there is no mantis.results exchange or result queue to monitor.

Events (SSE, not a queue)

Deployment lifecycle events (deployment.*, step.*, target.*) are streamed to clients over Server-Sent Events via Mandible’s /sse endpoints, not a mantis.events exchange.

Queues

Command Queues

Queue	Purpose	Consumers
mantis.commands.thorax	All commands for processing	Thorax instances
mantis.commands.priority.high	High-priority commands	Unused (reserved for future priority-based routing)
mantis.commands.priority.normal	Normal-priority commands	Unused (reserved for future priority-based routing)
mantis.commands.priority.low	Low-priority commands	Unused (reserved for future priority-based routing)

All current Thorax consumption happens via mantis.commands.thorax. The three priority queues receive bindings from the same routing keys but have no active consumers — they are reserved for future priority-based consumption.

Analysis Queue

Queue	Purpose	Consumers
mantis.analysis	AI log-analysis jobs	Analysis workers

Dead-Letter Queue

Queue	Purpose	Consumers
mantis.dlq	Failed/expired messages	Manual review

Routing Keys

Command Routing

Routing Key	Target Queue	Description
command.thorax	mantis.commands.thorax	All commands
command.poll.high	thorax + priority.high	High-priority poll commands
command.poll.normal	thorax + priority.normal	Normal poll commands
command.poll.low	thorax + priority.low	Low-priority poll commands
command.listen.high	thorax + priority.high	High-priority listen commands
command.listen.normal	thorax + priority.normal	Normal listen commands
command.listen.low	thorax + priority.low	Low-priority listen commands

Queue Configuration

Queue Arguments

Default queue configuration:

# Queue durability
durable = true
auto_delete = false
exclusive = false

# Dead-lettering
x-dead-letter-exchange = mantis.dlx

# Message TTL — set unconditionally by Mandible, and it varies per queue:
#   priority.high: 1,800,000 (30 min)   commands.thorax / priority.normal: 3,600,000 (1 hr)
#   priority.low: 7,200,000 (2 hr)      analysis: 14,400,000 (4 hr)   mantis.dlq: 604,800,000 (7 days)
x-message-ttl = 3600000  # 1 hour (commands.thorax / priority.normal)

Priority Queues

Priority queues use max-priority argument:

# Priority queue arguments
x-max-priority = 10

Quorum Queues

For high availability, use quorum queues:

# Quorum queue arguments
x-queue-type = quorum
x-quorum-initial-group-size = 3

Message Flow

Command Flow

Event Flow

Deployment lifecycle events are delivered to clients over Server-Sent Events, not RabbitMQ:

Dead-Letter Handling

Dead-Letter Reasons

Messages are dead-lettered when:

Reason	Description
rejected	Message explicitly rejected
expired	Message TTL exceeded
maxlen	Queue length limit reached

Retry Configuration

Mantis implements retry with exponential backoff:

1. On arrival the x-death count is checked first; if it already meets max_retries the
   message is NACKed straight to the DLQ (no processing attempt).
2. Otherwise the message is processed.
3. First processing failure (not redelivered): NACK requeue=true — it re-enters the same
   queue immediately.
4. Second failure (redelivered): NACK requeue=false — RabbitMQ routes it through the
   dead-letter exchange back to the main queue, incrementing x-death.
5. The cycle repeats until x-death reaches max_retries, then the message goes to the DLQ.

DLQ Management

Monitor dead-letter queue:

# List messages in DLQ
sudo rabbitmqctl list_queues name messages \
    | grep mantis.dlq

# Inspect DLQ messages
sudo rabbitmqadmin get queue=mantis.dlq count=10

# Purge DLQ (use with caution)
sudo rabbitmqctl purge_queue mantis.dlq

Reprocessing Failed Messages

Requeue messages from DLQ:

# Move messages back to original queue
# Using shovel plugin
sudo rabbitmqctl set_parameter shovel reprocess \
    '{"src-uri": "amqp://", "src-queue": "mantis.dlq", \
      "dest-uri": "amqp://", "dest-queue": "mantis.commands.thorax", \
      "delete-after": "queue-length"}'

Topology Declaration

Automatic Setup

Mandible declares the full topology on startup. Configure the RabbitMQ connection in mandible.toml:

mandible.toml

[queue]
enabled = true
host = "localhost"
port = 5671
username = "mantis"
password = "password"

Topology declaration is idempotent — Mandible re-declares exchanges, queues, and bindings on every startup, so no manual setup is needed. Thorax is a pure consumer and assumes the topology already exists (it retries its consumer connection until the queues appear).

Manual Setup

Declare topology manually:

# Create exchanges
rabbitmqadmin declare exchange name=mantis.commands type=direct durable=true
rabbitmqadmin declare exchange name=mantis.analysis type=direct durable=true
rabbitmqadmin declare exchange name=mantis.dlx type=fanout durable=true

# Create queues
rabbitmqadmin declare queue name=mantis.commands.thorax durable=true \
    arguments='{"x-dead-letter-exchange": "mantis.dlx"}'

rabbitmqadmin declare queue name=mantis.commands.priority.high durable=true \
    arguments='{"x-dead-letter-exchange": "mantis.dlx", "x-max-priority": 10}'

rabbitmqadmin declare queue name=mantis.commands.priority.normal durable=true \
    arguments='{"x-dead-letter-exchange": "mantis.dlx", "x-max-priority": 10}'

rabbitmqadmin declare queue name=mantis.commands.priority.low durable=true \
    arguments='{"x-dead-letter-exchange": "mantis.dlx", "x-max-priority": 10}'

rabbitmqadmin declare queue name=mantis.analysis durable=true \
    arguments='{"x-dead-letter-exchange": "mantis.dlx"}'

rabbitmqadmin declare queue name=mantis.dlq durable=true

# Create bindings
rabbitmqadmin declare binding source=mantis.commands destination=mantis.commands.thorax \
    routing_key=command.thorax

rabbitmqadmin declare binding source=mantis.commands destination=mantis.commands.priority.high \
    routing_key=command.poll.high

rabbitmqadmin declare binding source=mantis.commands destination=mantis.commands.priority.high \
    routing_key=command.listen.high

# ... additional bindings

Consumer Configuration

Prefetch Count

Configure consumer prefetch:

thorax.toml

[queue]
prefetch_count = 10  # Messages to prefetch per consumer

Consumer Tags

Thorax generates unique consumer tags:

thorax-{instance_id}-{uuid}

Acknowledgment

Mantis uses manual acknowledgment:

Action	When
ack	Message processed successfully
nack requeue=true	Temporary failure, retry immediately
nack requeue=false	Permanent failure, route to DLX

High Availability

Mirrored Queues

For classic mirrored queues:

# Set HA policy for all mantis queues
sudo rabbitmqctl set_policy ha-mantis "^mantis\." \
    '{"ha-mode":"all","ha-sync-mode":"automatic"}'

Quorum Queues

For quorum queues (recommended):

# Set policy for quorum queues
sudo rabbitmqctl set_policy quorum-mantis "^mantis\." \
    '{"queue-type":"quorum"}' --apply-to queues

Consumer Failover

When a Thorax instance fails:

1. Messages are redelivered to other consumers
2. In-flight messages timeout and retry
3. Health monitor switches to polling if needed

Scaling Considerations

Multiple Consumers

Scale horizontally with competing consumers:

Queue Sharding

For very high throughput:

mantis.commands.thorax.{shard_id}

Instance Affinity

Listen-mode targets use instance affinity:

# Command message includes assigned instance
assigned_instance_id = "thorax-node-1"

Troubleshooting

Queue Not Receiving Messages

1. Check binding exists:

   rabbitmqadmin list bindings source=mantis.commands

2. Verify routing key matches:

   # Publish test message
   rabbitmqadmin publish exchange=mantis.commands \
       routing_key=command.thorax payload='test'

3. Check queue is not paused:

   rabbitmqctl list_queues name state

Messages Stuck in DLQ

1. Inspect message headers:

   rabbitmqadmin get queue=mantis.dlq count=1

2. Check x-death header for failure reason

3. Review application logs for processing errors

Consumer Not Receiving

1. Check consumer is registered:

   rabbitmqctl list_consumers queue=mantis.commands.thorax

2. Verify prefetch is not exhausted:

   rabbitmqctl list_channels name prefetch_count messages_unacknowledged

Next Steps

• Monitoring - RabbitMQ monitoring
• TLS Configuration - Secure connections
• RabbitMQ Setup - Initial setup