feat(publishmq): reject invalid messages and cap redeliveries

[alexluong]

Stops publish queue messages from looping forever. publishmq nacks every handler error, and RabbitMQ requeues on nack, so a malformed message is redelivered in a tight loop and never reaches the queue's dead-letter exchange.

Two changes:

• Invalid messages (invalid JSON, data that is not a JSON object, missing topic when TOPICS is set) are rejected on the first failure.
• PUBLISH_MAX_REDELIVERIES (opt-in) caps redeliveries of any failed message.

PUBLISH_MAX_REDELIVERIES=5   # -1 (default) = unlimited, 0 = never redeliver

Behavior

Broker	Invalid message	Failure past the cap
RabbitMQ	nack(requeue=false): dead-lettered if the queue has a DLX, discarded otherwise	same
Azure Service Bus	dead-lettered	same
SQS, Pub/Sub	no reject exists; counted like any failure	acked (deleted)

• With the cap unset, failures are redelivered without limit, as today. Only the invalid-message reject on RabbitMQ and Azure changes by default.
• A topic missing from TOPICS stays a normal failure, since adding the topic fixes it (e.g. mid rolling deploy).
• On SQS and Pub/Sub, operators keep capped messages by setting maxReceiveCount / max_delivery_attempts no higher than the cap, so the broker dead-letters first. A broker-side limit also applies on RabbitMQ (quorum x-delivery-limit) and Azure (MaxDeliveryCount); the lower limit fires first.
• The cap counts attempts. Brokers redeliver immediately unless the queue has backoff, so a low cap can be used up during a short outage.

Counter

Failures only: one pipelined INCR + EXPIRE (24h) on {deployment_id}:publishmq:failures:{message id}. When the message has no ID, the key uses a hash of the body. If Redis errors, the message is nacked, so a Redis outage never drops messages. The count is not reset on success, to keep Redis off the success path.

Heads-up

• gocloud's rabbitpubsub Nack always requeues, so Reject nacks the raw amqp091.Delivery via As and clears gocloud's "never acked" finalizer (internal/mqs/queue_rabbitmq.go). Azure dead-letters through the same receiver gocloud settles with.
• The RabbitMQ key uses the delivery's MessageId, since gocloud's LoggableID falls back to the delivery tag, which changes on every redelivery.

Tests

• TestIntegrationMQ_RabbitMQReject: a rejected message lands in the DLX queue and is not redelivered.
• TestMessageHandler_MaxRedeliveries*: rejected (or acked on SQS/Pub/Sub) on the Nth redelivery, and still nacked when the counter errors.

🤖 Generated with Claude Code