rmq-healer 0.4.3

Abstract, extensible RabbitMQ error-queue remediation tool

Abstract, extensible RabbitMQ error-queue remediation.
Classify → remediate → audit — fully configurable, no code required for common cases.

---

What Is It?

RMQ Healer watches RabbitMQ error queues and automatically remediates known failure patterns — classify, remediate, audit — fully configurable, no code required for common cases.

When you probably don't need this tool

In a well-designed RabbitMQ system, error queues don't accumulate indefinitely. A properly configured setup handles failure natively:

• Dead-letter exchanges (DLX) — messages that are rejected, expired, or exceed queue length limits are automatically routed to a designated dead-letter queue
• Message TTL — messages that sit too long are expired and dead-lettered rather than silently dropped
• Consumer retry logic — consumers catch transient errors, requeue with backoff, and reject permanently after N attempts
• Dedicated error consumers — purpose-built services subscribe to error queues and handle known failure modes inline

If you control the system and have the bandwidth to implement these properly, you should — they are the right long-term answer.

When RMQ Healer is the right tool

RMQ Healer is built for situations where reconfiguring the messaging system is not an option:

• Legacy processors you can't modify — the consumer is a vendor library, a third-party service, or a system outside your control
• No technical bandwidth for a proper fix — reconfiguring RabbitMQ topology, updating consumers, and testing safely takes time you don't have right now
• Temporary remediation during migration — you're moving toward a better architecture but need something working in the meantime
• Operational recovery — a bad deployment filled an error queue and you need to safely triage, replay, and audit the affected messages

In all of these cases, RMQ Healer gives you a safe, auditable, externally-configurable remediation layer — without modifying the application that produced the messages.

Features

Zero-loss processing, declarative YAML rules, a rich built-in action library (log, mutate_json, set_headers, python_script, republish, quarantine, route_unknown, drop), Python script hooks for arbitrary logic, Prometheus metrics, dry-run with diffs, and an interactive terminal TUI for config editing, rule management, live runs, and queue inspection.

→ Full feature list

---

Architecture

Classify → remediate → audit. Messages are acked only after a terminal action succeeds — if anything fails, the message stays in the queue untouched.

flowchart TD
    EQ[("Error Queue")]
    CL["Classifier\n(first match wins)"]
    WF["WorkflowRunner"]
    AU["AuditLogger\n(JSON-lines)"]

    EQ -->|basic_get| CL
    CL -->|matched rule| WF
    CL -->|no match| UQ[("Unknown Queue")]
    WF --> A1["log / set_headers\nmutate_json / python_script"]
    A1 --> T{"Terminal\naction"}
    T -->|republish| RQ[("Repaired Queue")]
    T -->|quarantine| Q[("Quarantine Queue")]
    T -->|route_unknown| UQ
    T -->|drop| DONE["Ack + discard"]
    WF --> AU
    CL --> AU

→ Architecture overview and design decisions

---

Installation

Recommended: isolated CLI install

Use pipx when you want the rmq-healer command to be available on your shell PATH without manually managing a virtual environment.

python -m pip install --user pipx
python -m pipx ensurepath
pipx install rmq-healer

Restart your shell after pipx ensurepath, then verify:

rmq-healer --version

Alternative: install with pip

python -m pip install rmq-healer

If the install succeeds but rmq-healer is not found, your Python scripts directory is probably not on PATH. Inspect where the command was installed:

python -m pip show -f rmq-healer
python - <<'PY'
import sysconfig
print(sysconfig.get_path("scripts"))
PY

If you use pyenv, refresh shims after installation:

pyenv rehash

Development install

git clone https://github.com/grimnexo/RMQ-Healer.git
cd RMQ-Healer
python -m pip install -e ".[dev]"

---

Quick Start

# 1. Set broker URL
export RMQ_URL="amqp://guest:guest@localhost/"

# 2. Generate config interactively
rmq-healer configure --output config/queues.yml

# 3. Validate
rmq-healer validate --config config/queues.yml

# 4. Test a rule against a sample message
rmq-healer test-rule --config config/queues.yml \
  --queue order-service.error --message examples/retry-timeout/sample.json

# 5. Dry-run with mutation diffs — nothing acked or published
rmq-healer dry-run --config config/queues.yml --all --pretty

# 6. Live run
rmq-healer run --config config/queues.yml --all --pretty

# 7. Peek without consuming
rmq-healer peek --config config/queues.yml --queue order-service.error --count 10

# 8. Visualise a workflow as a Mermaid diagram
rmq-healer workflow render --config config/queues.yml --queue order-service.error

# 9. Search audit logs
rmq-healer audit-search --log audit.jsonl --outcome quarantine

---

Example Configuration

rabbitmq:
  url: "${RMQ_URL}"
  prefetch: 10

idempotency:
  enabled: true
  max_handler_attempts: 3

queues:
  - name: order-service.error
    enabled: true
    rules_file: rules/order-errors.yml
    repaired_exchange: order.events
    repaired_routing_key: order.retry
    quarantine_exchange: ops.errors
    quarantine_routing_key: order.quarantine

# rules/order-errors.yml
rules:
  - id: transient_timeout
    match:
      any:
        - body_contains: "ETIMEDOUT"
        - json_path: $.error.code
          equals: "TIMEOUT"
    workflow:
      - action: set_headers
        params:
          headers:
            x-rmq-healer-rule: transient_timeout
      - action: republish
        params:
          exchange: "{{ queue.repaired_exchange }}"
          routing_key: "{{ queue.repaired_routing_key }}"

  - id: missing_customer_id
    match:
      all:
        - json_path: $.customer_id
          equals: null
    workflow:
      - action: python_script
        params:
          script: scripts/lookup_customer.py
          timeout: 30
          output:
            save_as: customer
      - action: mutate_json
        params:
          set:
            "$.customer_id": "{{ customer.id }}"
      - action: republish
        params:
          exchange: "{{ queue.repaired_exchange }}"
          routing_key: "{{ queue.repaired_routing_key }}"

See the examples/ directory for complete, runnable scenarios.

---

Documentation

Topic	Guide
Installation & setup	docs/installation.md
Configuration reference	docs/configuration.md
Writing rules	docs/rules.md
Python script actions	docs/python-scripts.md
CLI reference (all commands)	docs/cli-reference.md
Operations guide	docs/operations.md
Docker deployment	docs/docker.md
Interactive TUI	docs/developer-gui.md
Plugin architecture	docs/plugins.md
Metrics & observability	docs/metrics.md
Contributing & tooling	docs/contributing.md
GitHub workflow	docs/github-workflow.md
Troubleshooting	docs/troubleshooting.md

Browse all docs →

---

Developers

Quick start for contributors.

# Windows
.\developer\scripts\setup.ps1      # start RabbitMQ + seed fixtures
.\developer\scripts\run-test.ps1   # dry-run against local broker
.\developer\scripts\teardown.ps1   # stop

# Linux / macOS
bash developer/scripts/setup.sh
bash developer/scripts/run-test.sh
bash developer/scripts/teardown.sh

# Python (cross-platform)
python developer/scripts/dev_setup.py
python developer/scripts/run_test.py
python developer/scripts/dev_teardown.py

→ Full contributor guide  ·  GitHub workflow

---

License

MIT — see LICENSE.