Skip to main content

RabbitMQ Queue Topology & Retry / Dead-letter Mechanics

Overview​

The internal/mq layer is the heart of the entire worker service. It declares the RabbitMQ topology, publishes jobs, consumes jobs, and decides whether a failed message should be retried or sent to a dead-letter queue (DLQ).

The retry model is deliberately not the broker's x-death mechanism but an application-level republish, ported from the original NestJS rabbitmq.decorator.ts so that behaviour matches the legacy system exactly.

The topology declares 38 queues in total, and every queue always has a matching DLQ named <queue>.dlq.

Business Flow​

Declaring the topology​

This runs at boot for every profile that uses RabbitMQ.

  1. Declare the main exchange line_exchange (type direct, durable) and the DLQ exchange line_exchange_dlq (direct, durable).
  2. Iterate over every queue in the list and call QueueDeclare with these arguments:
    • x-dead-letter-exchange = line_exchange_dlq
    • x-dead-letter-routing-key set to the queue's own name
    • x-max-priority = 10 (configurable via RABBITMQ_MAX_PRIORITY)
  3. Bind the queue to line_exchange using a routing key that is always identical to the queue name.
  4. Declare <queue>.dlq as durable and bind it to line_exchange_dlq using the source queue's name as the routing key.

Publishing​

  • Every producer calls Publish(ctx, queue, body, priority), which publishes to line_exchange using the queue name as the routing key.
  • The message body is the bare JSON payload with no envelope wrapper, keeping it compatible with the TypeScript-side producers.
  • Publishing uses publisher confirms together with DeliveryMode: Persistent, and a mutex guards the channel because AMQP channels are not concurrency-safe.

Consuming, retrying, and dead-lettering​

  1. Open a fresh channel per queue, set Qos(prefetch) (RABBITMQ_PREFETCH, default 10), and use manual acks.
  2. Deliveries are processed concurrently up to the prefetch value through a goroutine pool, so one slow message does not block the whole queue. Ack, nack, and republish operations are serialised by the channel mutex.
  3. When the handler returns nil, the message is acked.
  4. When the handler returns an error, classify(err) decides what happens next:
    • Transient — network errors such as ECONNRESET, ETIMEDOUT, ECONNREFUSED, EPIPE, EAI_AGAIN, or HTTP status 429 and 500 or above, and x-retry-count is still below 3. The message is delayed on a quadratic schedule (1s, 4s, 9s), republished to the default exchange with x-retry-count incremented, and the original delivery is acked.
    • Permanent, or transient after 3 retries — Nack(requeue=false), which drops the message into <queue>.dlq.
  5. A panic inside a handler is recovered and treated as a permanent error (dead-lettered); the process stays up.
  6. If the republish itself fails, the message is nacked with requeue=false rather than acked, so nothing disappears silently.

Key Files & Functions​

  • internal/mq/topology.go — AllQueues() (38 queues), MonitoredDLQQueues(), AssertTopology()
  • internal/mq/client.go — Client, New(), Connect(), the reconnect supervisor, and Name() returning "rabbitmq"
  • internal/mq/publisher.go — Publish(), PublishJSON()
  • internal/mq/consumer.go — RunConsumer(), drain(), handle(), the maxRetryCount = 3 constant, and retryCountOf()
  • internal/mq/registrar.go — Registrar.Register(queue, handler), Run(ctx)
  • internal/mq/errclass.go — IsTransientError(), a direct port of error-classifier.ts
  • internal/mq/errors.go — mq.Permanent(err) and mq.Transient(err) for forcing a classification
  • internal/mq/json.go — marshalling that byte-matches V8's JSON.stringify

Full queue list​

Every queue's routing key equals its name.

email, create_audience, update_audience, delete_audience, line_forward_webhook, line_broadcast_rich_message, line_multicast_rich_message, line_webhook, process_campaign, line_change_richmenu, line_sync_follower_user, tracking_log, line_user_last_activity, calculate_campaign_stat, calculate_rich_menu_stat, calculate_rich_menu_stat_item, line_auto_response, line_import_csv_user, import_mapping_job, mookept_audience, attribute_change, attribute_change_realtime, audience_membership_trigger, friend_track_event_trigger, friend_track_ref_import, audience_refresh, action_execute, scheduled_trigger_fire, scheduled_action_fire, form_submitted_trigger, campaign_click_trigger, web_request_execute, message_received_trigger, knowledge_index, mbox_callback, mbox_handoff, booking_notification, booking_event_trigger

note

import_mapping_job is new to the Go implementation and has no equivalent in the legacy NestJS system. message_received_trigger is asserted in the topology but its DLQ is deliberately not monitored — see the Dead-letter Queue Monitor page.

Connections to Other Services​

  • External producers
    • line-management-webhook-go — publishes to line_webhook, tracking_log, and line_forward_webhook
    • line-management-cms-api-go — publishes campaign, audience, import, knowledge index, and rich menu jobs
    • line-management-client-api-go — publishes tracking, form, and booking jobs
  • Internal producers — the pg-listener profile (translating pg_notify events), the scanners in the cron-scheduler profile, and consumers that chain further work, such as a trigger publishing to action_execute
  • Related configuration — RABBITMQ_URL, RABBITMQ_PREFETCH, RABBITMQ_MAX_PRIORITY, RABBITMQ_EXCHANGE_NAME, RABBITMQ_DLQ_EXCHANGE_NAME, and the RABBITMQ_QUEUE_* group