RabbitMQ DLQ Poison-Pill Quarantine and Safe Replay Gate

id: rabbitmq-dlq-quarantine-and-replay
namespace: company.team
description: |
  Drain dead-lettered messages from a RabbitMQ Dead-Letter Queue (DLQ),
  classify failure root causes into transient vs unrecoverable poison pills,
  quarantine poisoned payloads to Amazon S3 with an audit manifest, and
  safely replay recoverable messages back to the primary exchange behind an
  explicit dry-run gate.

triggers:
  - id: periodic_dlq_drain
    type: io.kestra.plugin.core.trigger.Schedule
    description: Drain and triage dead-lettered messages every 15 minutes. Shipped
      disabled; enable once queue names and secrets are configured.
    cron: "*/15 * * * *"
    disabled: true

inputs:
  - id: dlq_queue
    type: STRING
    displayName: "DLQ Queue Name"
    defaults: "orders.dlq"
    description: Name of the RabbitMQ Dead-Letter Queue to drain.

  - id: target_exchange
    type: STRING
    displayName: "Target Exchange"
    defaults: "orders.exchange"
    description: The exchange to republish recoverable messages back into.

  - id: target_routing_key
    type: STRING
    displayName: "Target Routing Key"
    defaults: "orders.created"
    description: The routing key applied to replayed messages.

  - id: max_records
    type: INT
    displayName: "Max Records"
    defaults: 50
    description: Maximum dead-letter messages to drain and process in a single batch.

  - id: max_retry_attempts
    type: INT
    displayName: "Max Retry Attempts"
    defaults: 3
    description: Maximum death count from RabbitMQ x-death headers before a message
      is classified as retry-exhausted and permanently quarantined.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: When true, simulate triage and skip re-publishing to the exchange.
      Set to false to execute live replay.

  - id: s3_quarantine_bucket
    type: STRING
    displayName: "S3 Quarantine Bucket"
    defaults: "company-dlq-quarantine-archive"
    description: Amazon S3 bucket used for cold-storage quarantine archives.

  - id: s3_region
    type: STRING
    displayName: "AWS Region"
    defaults: "us-east-1"
    description: AWS region where the quarantine S3 bucket is hosted.

tasks:
  - id: drain_dlq
    type: io.kestra.plugin.amqp.Consume
    description: Consume up to max_records from the Dead-Letter Queue into Kestra
      internal storage.
    host: "{{ secret('RABBITMQ_HOST') }}"
    port: "{{ secret('RABBITMQ_PORT') }}"
    username: "{{ secret('RABBITMQ_USERNAME') }}"
    password: "{{ secret('RABBITMQ_PASSWORD') }}"
    virtualHost: "{{ secret('RABBITMQ_VHOST') }}"
    queue: "{{ inputs.dlq_queue }}"
    serdeType: JSON
    maxRecords: "{{ inputs.max_records }}"
    maxDuration: PT30S
    autoAck: false

  - id: evaluate_drained_messages
    type: io.kestra.plugin.core.flow.If
    description: Evaluate whether any dead-letter messages were retrieved during the
      poll cycle.
    condition: "{{ (outputs.drain_dlq.count ?? 0) > 0 }}"
    then:
      - id: classify_and_triage
        type: io.kestra.plugin.scripts.python.Script
        description: Classify error signatures, separate transient failures from toxic
          poison pills, and prepare quarantine and replay artifacts.
        containerImage: python:3.11-slim
        taskRunner:
          type: io.kestra.plugin.scripts.runner.docker.Docker
        inputFiles:
          dlq_data.ion: "{{ outputs.drain_dlq.uri }}"
        outputFiles:
          - quarantine_manifest.json
          - replay_batch.json
          - triage_summary.json
        script: |
          import json
          import os
          from kestra import Kestra

          max_retries = int("{{ inputs.max_retry_attempts }}")
          raw_records = []

          with open("dlq_data.ion", "r", encoding="utf-8", errors="ignore") as f:
              for line in f:
                  line = line.strip()
                  if not line:
                      continue
                  try:
                      if line.startswith("{") and line.endswith("}"):
                          parsed = json.loads(line)
                          raw_records.append(parsed)
                      else:
                          raw_records.append({"data": line, "raw": True})
                  except Exception:
                      raw_records.append({"data": line, "raw": True, "parse_error": True})

          replay_candidates = []
          quarantined_records = []
          poison_count = 0
          exhausted_count = 0
          transient_count = 0

          for item in raw_records:
              payload = item.get("data", item)
              headers = item.get("headers", {})

              retry_count = 0
              x_death = headers.get("x-death")
              if isinstance(x_death, list) and len(x_death) > 0:
                  retry_count = x_death[0].get("count", 0)
              elif "retry_count" in headers:
                  try:
                      retry_count = int(headers["retry_count"])
                  except Exception:
                      retry_count = 0
              elif isinstance(payload, dict) and "retry_count" in payload:
                  try:
                      retry_count = int(payload["retry_count"])
                  except Exception:
                      retry_count = 0

              raw_str = json.dumps(item) if isinstance(item, (dict, list)) else str(item)
              is_transient = any(kw in raw_str.upper() for kw in ["TIMEOUT", "503", "504", "429", "TEMPORARY", "CONNECTION_RESET", "LOCK_WAIT"])

              if item.get("parse_error") or not isinstance(payload, dict):
                  classification = "POISON_CORRUPT_PAYLOAD"
                  poison_count += 1
                  quarantined_records.append({
                      "classification": classification,
                      "retry_count": retry_count,
                      "error_reason": "Payload unparseable or non-dictionary format",
                      "record": item
                  })
              elif retry_count >= max_retries:
                  classification = "RETRY_BUDGET_EXHAUSTED"
                  exhausted_count += 1
                  quarantined_records.append({
                      "classification": classification,
                      "retry_count": retry_count,
                      "error_reason": f"Retry count {retry_count} reached threshold {max_retries}",
                      "record": item
                  })
              elif is_transient:
                  classification = "TRANSIENT_RECOVERABLE"
                  transient_count += 1
                  updated_headers = dict(headers) if isinstance(headers, dict) else {}
                  updated_headers["x-replayed-by"] = "kestra-rabbitmq-dlq-pipeline"
                  updated_headers["retry_count"] = retry_count + 1
                  replay_candidates.append({
                      "data": payload,
                      "headers": updated_headers
                  })
              else:
                  classification = "POISON_UNHANDLED_ERROR"
                  poison_count += 1
                  quarantined_records.append({
                      "classification": classification,
                      "retry_count": retry_count,
                      "error_reason": "Business rejection or unhandled application exception",
                      "record": item
                  })

          summary = {
              "total_drained": len(raw_records),
              "replay_count": len(replay_candidates),
              "quarantine_count": len(quarantined_records),
              "transient_count": transient_count,
              "poison_count": poison_count,
              "exhausted_count": exhausted_count
          }

          manifest = {
              "execution_id": "{{ execution.id }}",
              "dlq_queue": "{{ inputs.dlq_queue }}",
              "target_exchange": "{{ inputs.target_exchange }}",
              "target_routing_key": "{{ inputs.target_routing_key }}",
              "summary": summary,
              "quarantined_items": quarantined_records
          }

          with open("quarantine_manifest.json", "w", encoding="utf-8") as f:
              json.dump(manifest, f, indent=2)

          with open("replay_batch.json", "w", encoding="utf-8") as f:
              json.dump(replay_candidates, f, indent=2)

          with open("triage_summary.json", "w", encoding="utf-8") as f:
              json.dump(summary, f, indent=2)

          Kestra.outputs({
              "total_drained": len(raw_records),
              "replay_count": len(replay_candidates),
              "quarantine_count": len(quarantined_records),
              "transient_count": transient_count,
              "poison_count": poison_count,
              "exhausted_count": exhausted_count,
              "replay_records": replay_candidates
          })

      - id: evaluate_replay
        type: io.kestra.plugin.core.flow.If
        description: Republish recoverable transient messages back to the primary
          exchange when dry run is disabled.
        condition: "{{ outputs.classify_and_triage.vars.replay_count > 0 and
          inputs.dry_run == 'false' }}"
        then:
          - id: replay_to_exchange
            type: io.kestra.plugin.amqp.Publish
            description: Safely publish recoverable messages to the primary target exchange.
            host: "{{ secret('RABBITMQ_HOST') }}"
            port: "{{ secret('RABBITMQ_PORT') }}"
            username: "{{ secret('RABBITMQ_USERNAME') }}"
            password: "{{ secret('RABBITMQ_PASSWORD') }}"
            virtualHost: "{{ secret('RABBITMQ_VHOST') }}"
            exchange: "{{ inputs.target_exchange }}"
            routingKey: "{{ inputs.target_routing_key }}"
            serdeType: JSON
            from: "{{ outputs.classify_and_triage.vars.replay_records }}"

          - id: log_replay_executed
            type: io.kestra.plugin.core.log.Log
            message: "Replay executed: Published {{
              outputs.classify_and_triage.vars.replay_count }} recoverable
              message(s) to exchange '{{ inputs.target_exchange }}' with routing
              key '{{ inputs.target_routing_key }}'."
        else:
          - id: log_replay_skipped
            type: io.kestra.plugin.core.log.Log
            message: "Replay skipped: dry_run={{ inputs.dry_run }}, replay_candidates={{
              outputs.classify_and_triage.vars.replay_count }}."

      - id: evaluate_quarantine
        type: io.kestra.plugin.core.flow.If
        description: Archive toxic or retry-exhausted messages to Amazon S3 cold storage
          and notify on-call engineers.
        condition: "{{ outputs.classify_and_triage.vars.quarantine_count > 0 }}"
        then:
          - id: upload_quarantine_to_s3
            type: io.kestra.plugin.aws.s3.Upload
            description: Upload structured forensic quarantine manifest to Amazon S3.
            accessKeyId: "{{ secret('AWS_ACCESS_KEY_ID') }}"
            secretKeyId: "{{ secret('AWS_SECRET_ACCESS_KEY') }}"
            region: "{{ inputs.s3_region }}"
            bucket: "{{ inputs.s3_quarantine_bucket }}"
            from: "{{ outputs.classify_and_triage.outputFiles['quarantine_manifest.json']
              }}"
            key: "quarantine/rabbitmq/{{ now() | date('yyyy/MM/dd') }}/quarantine-{{
              execution.id }}.json"

          - id: alert_quarantined_poison
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: Dispatch Slack alert detailing quarantined poison pills and S3
              storage location.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "RabbitMQ DLQ Quarantine Alert: {{ outputs.classify_and_triage.vars.quarantine_count }} poisoned or retry-exhausted message(s) quarantined from queue `{{ inputs.dlq_queue }}`. Forensic manifest archived at `s3://{{ inputs.s3_quarantine_bucket }}/quarantine/rabbitmq/{{ now() | date('yyyy/MM/dd') }}/quarantine-{{ execution.id }}.json`. Execution: {{ execution.id }}."
              }

      - id: notify_triage_summary
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Deliver end-of-batch summary report to the engineering operations
          Slack channel.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "RabbitMQ DLQ Triage Summary: Queue `{{ inputs.dlq_queue }}` drained {{ outputs.classify_and_triage.vars.total_drained }} message(s). Replayable: {{ outputs.classify_and_triage.vars.replay_count }} (dry_run: {{ inputs.dry_run }}). Quarantined: {{ outputs.classify_and_triage.vars.quarantine_count }} (Poison: {{ outputs.classify_and_triage.vars.poison_count }}, Exhausted: {{ outputs.classify_and_triage.vars.exhausted_count }}). Execution: {{ execution.id }}."
          }
    else:
      - id: log_clean_dlq
        type: io.kestra.plugin.core.log.Log
        message: "DLQ '{{ inputs.dlq_queue }}' is clear. Drained 0 messages."

errors:
  - id: alert_pipeline_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert engineering team if the triage pipeline encounters an
      unhandled broker or storage error.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "RabbitMQ DLQ Triage Pipeline FAILED in flow {{ flow.id }} (execution {{ execution.id }}) for queue {{ inputs.dlq_queue }}. Verify RabbitMQ broker connectivity, credentials, and S3 permissions."
      }

outputs:
  - id: total_drained
    type: INT
    description: Total number of dead-lettered messages drained from the queue.
    value: "{{ outputs.drain_dlq.count ?? 0 }}"

  - id: replay_count
    type: INT
    description: Total number of recoverable transient messages identified for replay.
    value: "{{ outputs.classify_and_triage.vars.replay_count ?? 0 }}"

  - id: quarantine_count
    type: INT
    description: Total number of poison pills or exhausted messages quarantined to S3.
    value: "{{ outputs.classify_and_triage.vars.quarantine_count ?? 0 }}"