Rabbitmq Triage Dlq

Triaging RabbitMQ DLQ Messages

Overview

Systematic diagnosis of why messages end up in a Dead Letter Queue. Analyzes message payloads, headers, subscriber DTOs, handler logic, and database state to identify root causes and recommend actions. After triage, use rabbitmq-reprocess-dlq to act on findings.

Infrastructure Context

See infra-portforward skill for service names, ports, and credential retrieval. See rabbitmq-reprocess-dlq for queue naming conventions.

Key facts for triage:

• Consumer pods are in namespace plataformadato
• App code is in /app/dist/src/ inside pods
• Subscriber pattern: /app/dist/src/application/amqp/<entity>-subscriber/
• Config pattern: /app/dist/src/config/production.config.js
• DB access via PgBouncer (no psql in pods, use node -e with pg module)
• PgBouncer does NOT support parameterized queries ($1::uuid[]). Use inline SQL.

Triage Process

1. Survey queues           → How many messages? Active consumers?
2. Fetch messages           → Via Management API (needs port-forward)
3. Analyze message patterns → Schema, headers, failure classification
4. Inspect subscriber code  → DTO validation + handler logic from pod
5. Verify external state    → DB lookups for referenced entities
6. Check consumer logs      → Trigger 1-message reprocess if needed
7. Present findings         → Root cause + recommendation

Step 1: Survey queues

kubectl --context=<env> get pods -n shared-infra | grep rabbit

# List ALL DLQs with messages
kubectl --context=<env> exec -n shared-infra <pod> -- \
  rabbitmqctl list_queues -p data_platform name messages consumers state \
  | grep "dead_letter" | awk '$2 > 0'

# Check specific service (main + DLQ)
kubectl --context=<env> exec -n shared-infra <pod> -- \
  rabbitmqctl list_queues -p data_platform name messages consumers state \
  | grep "<service>"

If main queue has 0 consumers, the service may be down — check pods in plataformadato first.

Step 2: Fetch messages

Requires port-forward + credentials (see infra-portforward skill for setup). Never hardcode credentials.

# Fetch messages (non-destructive peek)
curl -s -u <user>:<password> \
  -X POST -H "Content-Type: application/json" \
  'http://localhost:15672/api/queues/data_platform/<url-encoded-queue>/get' \
  -d '{"count": <N>, "ackmode": "ack_requeue_true", "encoding": "auto"}' \
  > /tmp/dlq_messages.json

URL-encode queue names: replace : with %3A.

Step 3: Analyze message patterns

Run the analysis script from this skill's scripts/ directory:

python3 ~/.claude/skills/rabbitmq-triage-dlq/scripts/analyze_messages.py /tmp/dlq_messages.json
python3 ~/.claude/skills/rabbitmq-triage-dlq/scripts/analyze_messages.py /tmp/dlq_messages.json --fields clientId,family,category --detail 10

The script covers: header analysis (immediateNack, x-death counts), schema variations, field distributions, and message detail. See scripts/analyze_messages.py for full source.

Classification:

Step 4: Inspect subscriber code

Read the DTO and handler directly from the running pod:

# Find consumer pods
kubectl --context=<env> get pods -n plataformadato | grep <service>

# Find subscriber files
kubectl --context=<env> exec -n plataformadato <pod> -- \
  find /app/dist -type f -name "*.js" | grep -i "<entity>.*subscriber"

# Read the DTO (most critical file)
kubectl --context=<env> exec -n plataformadato <pod> -- \
  cat /app/dist/src/application/amqp/<entity>-subscriber/dto/<entity>-dto.js

# Read the handler
kubectl --context=<env> exec -n plataformadato <pod> -- \
  cat /app/dist/src/application/amqp/<entity>-subscriber/<entity>.subscriber.js

DTO analysis checklist:

• Compare every @IsNotEmpty() field against actual message payloads
• Check nested DTOs (jiraRefs, erpRef, sourceSystems) — they have their own validations
• @ValidateNested() on arrays means each element must pass validation
• Field in DTO as @IsNotEmpty() but missing in messages? That's the bug.

Step 5: Verify external state

If the handler does DB lookups (getClientById, etc.), check whether referenced entities exist.

# Read DB config
kubectl --context=<env> exec -n plataformadato <pod> -- \
  cat /app/dist/src/config/production.config.js

# Query via node.js (no psql available in pods)
kubectl --context=<env> exec -n plataformadato <pod> -- sh -c '
node -e "
const { Client } = require(\"pg\");
const c = new Client({
  host: \"<pgbouncer-host>\", port: 6432,
  database: \"<database>\", user: \"postgres\",
  password: process.env.POSTGRES_PASSWORD
});
const ids = [\"uuid1\", \"uuid2\"];
const inClause = ids.map(id => \"\\x27\" + id + \"\\x27\").join(\",\");
c.connect()
  .then(() => c.query(\"SELECT id FROM <table> WHERE id IN (\" + inClause + \")\"))
  .then(r => {
    const found = new Set(r.rows.map(x => x.id));
    const missing = ids.filter(id => !found.has(id));
    console.log(\"Found: \" + found.size + \"/\" + ids.length);
    if (missing.length > 0) console.log(\"Missing: \" + JSON.stringify(missing));
    c.end();
  }).catch(e => { console.error(e.message); c.end(); });
"
'

Step 6: Check consumer logs

kubectl --context=<env> logs <pod> -n plataformadato --since=30m 2>&1 \
  | grep -i "error\|exception\|reject\|nack\|<entity>" | tail -40

If no errors appear: the failure is DTO validation (happens before the handler, produces no log output).

If you need live errors: trigger a 1-message reprocess and watch logs simultaneously:

# Terminal 1: watch logs
kubectl --context=<env> logs -f <pod> -n plataformadato | grep -i "error\|<entity>"

# Terminal 2: reprocess 1 message via shovel
kubectl --context=<env> exec -n shared-infra <rabbitmq-pod> -- \
  rabbitmqctl set_parameter -p data_platform shovel "test-1msg" \
  '{"src-protocol":"amqp091","src-uri":"amqp:///data_platform","src-queue":"<dlq-queue>","src-prefetch-count":1,"dest-protocol":"amqp091","dest-uri":"amqp:///data_platform","dest-queue":"<main-queue>","src-delete-after":"1"}'

This moves exactly 1 message so you can observe the error in real-time.

Diagnosis aids

See references/known-patterns.md for:

• Root cause decision tree
• Known failure patterns (jiraRefs, client not found, schema drift)
• Output template format