Dead Letter Queue (DLQ) en GCP Pub/Sub: Manejo de Errores y Reintentos

terraform {
  required_providers {
    google = {
      source  = "hashicorp/google"
      version = "~> 7.45"
    }
  }
}

variable "project_id" {
  type = string
}

variable "project_number" {
  type = string
}

provider "google" {
  project = var.project_id
}

# ------------------------------------------------------------
# 1. Topic principal
# ------------------------------------------------------------

resource "google_pubsub_topic" "orders" {
  name = "orders"
}

# ------------------------------------------------------------
# 2. Dead Letter Topic
# ------------------------------------------------------------

resource "google_pubsub_topic" "orders_dlq" {
  name = "orders-dlq"
}

# ------------------------------------------------------------
# 3. Subscription para inspeccionar la DLQ
# ------------------------------------------------------------

resource "google_pubsub_subscription" "orders_dlq" {
  name  = "orders-dlq-sub"
  topic = google_pubsub_topic.orders_dlq.id

  message_retention_duration = "604800s"
}

# ------------------------------------------------------------
# 4. Subscription principal
# ------------------------------------------------------------

resource "google_pubsub_subscription" "orders" {
  name  = "orders-sub"
  topic = google_pubsub_topic.orders.id

  ack_deadline_seconds = 30

  # Backoff entre entregas consecutivas
  retry_policy {
    minimum_backoff = "10s"
    maximum_backoff = "300s"
  }

  # Después de aproximadamente N intentos,
  # Pub/Sub puede reenviar el mensaje a la DLQ.
  dead_letter_policy {
    dead_letter_topic     = google_pubsub_topic.orders_dlq.id
    max_delivery_attempts = 10
  }
}

# ------------------------------------------------------------
# 5. IAM: Pub/Sub puede publicar en la DLQ
# ------------------------------------------------------------

resource "google_pubsub_topic_iam_member" "dlq_publisher" {
  topic  = google_pubsub_topic.orders_dlq.name
  role   = "roles/pubsub.publisher"
  member = "serviceAccount:service-${var.project_number}@gcp-sa-pubsub.iam.gserviceaccount.com"
}

# ------------------------------------------------------------
# 6. IAM: Pub/Sub puede ACKear mensajes reenviados
# ------------------------------------------------------------

resource "google_pubsub_subscription_iam_member" "source_subscriber" {
  subscription = google_pubsub_subscription.orders.name
  role         = "roles/pubsub.subscriber"
  member       = "serviceAccount:service-${var.project_number}@gcp-sa-pubsub.iam.gserviceaccount.com"
}

import json
import logging

from google.cloud import pubsub_v1
from google.api_core.exceptions import DeadlineExceeded, ServiceUnavailable

PROJECT_ID = "my-project"
SUBSCRIPTION_ID = "orders-sub"

logging.basicConfig(level=logging.INFO)


def process_order(payload: dict) -> None:
    """
    Ejecuta la lógica de negocio.

    Los errores transitorios deben propagarse para permitir
    que Pub/Sub vuelva a entregar el mensaje.
    """

    # Ejemplo de lógica de negocio.
    order_id = payload["order_id"]

    logging.info("Processing order=%s", order_id)

    # Llamada a una dependencia externa...
    # external_service.process(payload)


def callback(message: pubsub_v1.subscriber.message.Message) -> None:
    try:
        payload = json.loads(message.data.decode("utf-8"))

        process_order(payload)

        # ACK únicamente después de completar
        # correctamente la operación.
        message.ack()

    except (DeadlineExceeded, ServiceUnavailable) as exc:
        # Error potencialmente transitorio.
        # NACK permite que Pub/Sub vuelva a entregar.
        logging.warning(
            "Transient error. Retrying message=%s: %s",
            message.message_id,
            exc,
        )
        message.nack()

    except (json.JSONDecodeError, KeyError, ValueError) as exc:
        # Error probablemente permanente.
        #
        # No debemos mantener indefinidamente un mensaje
        # estructuralmente inválido en el flujo normal.
        #
        # En una arquitectura real puede registrarse el
        # contexto y ACKearse si el proceso de remediación
        # se realiza fuera de Pub/Sub.
        logging.error(
            "Invalid message=%s: %s",
            message.message_id,
            exc,
        )
        message.ack()

    except Exception:
        # Error desconocido:
        # conservar el mensaje para permitir reintento.
        logging.exception(
            "Unexpected error for message=%s",
            message.message_id,
        )
        message.nack()


def main() -> None:
    subscriber = pubsub_v1.SubscriberClient()

    subscription_path = subscriber.subscription_path(
        PROJECT_ID,
        SUBSCRIPTION_ID,
    )

    streaming_pull_future = subscriber.subscribe(
        subscription_path,
        callback=callback,
    )

    logging.info(
        "Listening on %s",
        subscription_path,
    )

    try:
        streaming_pull_future.result()
    except KeyboardInterrupt:
        streaming_pull_future.cancel()
        streaming_pull_future.result()


if __name__ == "__main__":
    main()