diff --git a/README.rabbitmq.md b/README.rabbitmq.md index b783390..c01c71c 100644 --- a/README.rabbitmq.md +++ b/README.rabbitmq.md @@ -5,8 +5,9 @@ The RabbitMQ provider uses [`github.com/rabbitmq/amqp091-go`](https://github.com/rabbitmq/amqp091-go). It creates separate producer and consumer channels, declares quorum queues, applies -consumer prefetch, reconnects closed connections, and reports unroutable -publishes. +consumer prefetch, and reconnects closed connections. Services can explicitly +opt into persistent publisher confirms and delayed source acknowledgements when +their delivery guarantees justify the additional broker round trip. ## Quick Start @@ -62,6 +63,7 @@ methods or channels is required. | `AnalysisQueue` | No | Optional analysis destination | | `PrefetchCount` | No | Maximum unacknowledged deliveries; defaults to `5` | | `MaxRetries` | No | Retry limit before dead-lettering; defaults to `10` | +| `ConfirmedDelivery` | No | Provisions the confirmed producer used by explicit confirmed methods; defaults to `false` | | `Exchange` | No | Compatibility field; the default exchange is used | | `Uri` | No | Compatibility field; connection uses host and credentials | | `TLS` | No | Enables an `amqps://` connection | @@ -70,7 +72,7 @@ methods or channels is required. The builder provides matching setters, including `SetConsumerQueue`, `SetDeadletterQueue`, `SetRouterQueue`, `SetHost`, `SetUsername`, `SetPassword`, -`SetPrefetchCount`, `SetMaxRetries`, and the TLS setters. +`SetPrefetchCount`, `SetMaxRetries`, `SetConfirmedDelivery`, and the TLS setters. ### Environment Example @@ -93,12 +95,47 @@ queues. The router, analysis, and arbitrary publish destinations must already exist or be declared by their owning consumers. Publishing uses the default exchange with the destination queue as its routing -key. Messages are marked mandatory. If no queue is bound to the routing key, -RabbitMQ returns the message instead of silently dropping it. +key. `Publish` retains the legacy asynchronous behavior. Mandatory returns are +reported through `ReturnHandler`, but they can arrive after `Publish` returns. + +For stronger delivery guarantees, enable and call the explicit confirmed API: + +```go +options := queue.NewRabbitOptions(). + SetConsumerQueue("kcloud-export-queue"). + SetDeadletterQueue("dead-letter-queue"). + SetHost("rabbitmq.example.com:5672"). + SetUsername("username"). + SetPassword("password"). + SetConfirmedDelivery(true). + Build() + +rabbit, err := queue.NewRabbitMQ(options) +if err != nil { + log.Fatal(err) +} +if err := rabbit.Connect(); err != nil { + log.Fatal(err) +} +defer rabbit.Close() + +if err := rabbit.PublishConfirmed("target-queue", payload); err != nil { + log.Fatal(err) +} +``` + +`PublishConfirmed` marks messages persistent and waits up to five seconds for +RabbitMQ to confirm acceptance. It returns an error for a negative +acknowledgement, confirmation timeout, closed channel, or mandatory return. +Confirmed publishes on one client are serialized so returns correlate with +their confirmation, and reconnect cannot replace that producer mid-publish. +This synchronous broker round trip can reduce throughput, so enable it per +service rather than as a blanket dependency upgrade. Use `SetReturnHandler` on the concrete `RabbitMQ` client to report returned -messages to metrics or structured logging. Without a handler, returns are logged -through the standard logger. +messages to metrics or structured logging. The callback runs asynchronously. +`PublishConfirmed` also returns the unroutable error synchronously. Without a +handler, returns are logged through the standard logger. ```go rabbit, err := queue.NewRabbitMQ(options) @@ -131,9 +168,10 @@ Use `SetTLSInsecureSkipVerify(true)` only in controlled development environments ## Consumption Semantics -RabbitMQ deliveries use manual acknowledgements. `ReadMessages` decodes each -payload as `models.PipelineEvent`, invokes the handler, performs the selected -pipeline action, and acknowledges the original delivery. +RabbitMQ deliveries use manual acknowledgements. `ReadMessages` retains the +legacy behavior. `ReadMessagesConfirmed` uses persistent confirmed publishes +and acknowledges a source delivery only after its transfer is durable. It +requires `SetConfirmedDelivery(true)`. - Invalid JSON is sent to the dead-letter queue. - `PipelineForward` removes the completed stage and publishes to `RouterQueue` @@ -145,6 +183,13 @@ pipeline action, and acknowledges the original delivery. original delivery unacknowledged so reconnect/redelivery can recover it. - `PipelineCancel` only acknowledges the delivery. +Under `ReadMessagesConfirmed`, any failed forward, retry, or dead-letter +transfer is offered to the configured `DisasterRecoveryHandler`. The original +delivery is acknowledged only after the destination publish or disaster +recovery succeeds. If both fail (or no recovery handler is configured), the +library negatively acknowledges with requeue, closes the stale connection, and +returns the error so a fresh consumer can redeliver it. + RabbitMQ retries carry an `x-retry-count` message header. After `MaxRetries`, the payload is parked on `DeadletterQueue` rather than requeued indefinitely. @@ -158,6 +203,10 @@ The concrete `RabbitMQ` client also supports queues whose payload is not a - `ReadRawMessages` runs a long-lived raw consumer. - `ReadOneRaw` reads at most one message for verification and drain tools. +- `ReadRawMessagesConfirmed` and `ReadOneRawConfirmed` provide the corresponding + confirmed settlement behavior when explicitly enabled. + +`RouteMessagesConfirmed` is the confirmed counterpart to `RouteMessages`. `RawMessageHandler` receives the original bytes and returns a pipeline action, an optional forwarding payload, and retry backoff seconds. @@ -200,5 +249,8 @@ Run RabbitMQ-focused unit tests: go test ./pkg/queue -run 'TestRabbit|TestRetry' -v ``` -The integration test runs only when RabbitMQ connection environment variables -are provided; otherwise it is skipped. +The integration suite runs only when `RABBITMQ_HOST`, `RABBITMQ_USERNAME`, and +`RABBITMQ_PASSWORD` are provided; otherwise it is skipped. In addition to basic +connectivity, it verifies that confirmed publishing rejects an unroutable queue +and that an unacknowledged persistent message is redelivered after a forced AMQP +connection interruption and reconnect. diff --git a/pkg/queue/rabbitmq.go b/pkg/queue/rabbitmq.go index ebfb46c..7548408 100644 --- a/pkg/queue/rabbitmq.go +++ b/pkg/queue/rabbitmq.go @@ -25,21 +25,22 @@ type DisasterRecoveryHandler func([]byte) error // exchange the routing key is the target queue name, so a return means the // producer is publishing to a queue that does not exist — typically a // producer/consumer queue-name drift that is otherwise completely silent. The -// handler must be cheap and non-blocking; it runs on the return-watcher goroutine. +// handler runs asynchronously for both legacy and confirmed publishes. type ReturnHandler func(amqp.Return) // RabbitOptions holds the configuration for RabbitMQ type RabbitOptions struct { - ConsumerQueue string `validate:"required"` // Queue from which to consume messages, one consumer per queue - RouterQueue string // Router queue; only used — and only needs to be set — when a handler returns the Forward action. Stage workers that never forward may leave it unset. - DeadletterQueue string `validate:"required"` // When something goes wrong, messages are sent here - AnalysisQueue string // Queue for analysis messages - Uri string - Host string `validate:"required"` - Username string `validate:"required"` - Password string `validate:"required"` - PrefetchCount int - Exchange string + ConsumerQueue string `validate:"required"` // Queue from which to consume messages, one consumer per queue + RouterQueue string // Router queue; only used — and only needs to be set — when a handler returns the Forward action. Stage workers that never forward may leave it unset. + DeadletterQueue string `validate:"required"` // When something goes wrong, messages are sent here + AnalysisQueue string // Queue for analysis messages + Uri string + Host string `validate:"required"` + Username string `validate:"required"` + Password string `validate:"required"` + PrefetchCount int + Exchange string + ConfirmedDelivery bool // MaxRetries caps how many times a PipelineRetry re-queues a message before // it is parked on the deadletter queue; zero selects defaultMaxRetries. It is @@ -158,6 +159,13 @@ func (b *RabbitOptionsBuilder) SetMaxRetries(maxRetries int) *RabbitOptionsBuild return b } +// SetConfirmedDelivery provisions the dedicated publisher-confirm channel used +// by PublishConfirmed and the confirmed consumer methods. +func (b *RabbitOptionsBuilder) SetConfirmedDelivery(enabled bool) *RabbitOptionsBuilder { + b.options.ConfirmedDelivery = enabled + return b +} + // SetTLS enables TLS for the connection func (b *RabbitOptionsBuilder) SetTLS(enabled bool) *RabbitOptionsBuilder { b.options.TLS = enabled @@ -184,13 +192,16 @@ func (b *RabbitOptionsBuilder) Build() *RabbitOptions { // RabbitMQ wraps rabbitmq.Client to implement the Queue interface type RabbitMQ struct { options *RabbitOptions - connectionString string // e.g., amqp://user:pass@host:port/ - Connection *amqp.Connection // The underlying RabbitMQ connection - Consumer *amqp.Channel // Channel for consuming messages - Producer *amqp.Channel // Channel for producing messages + connectionString string // e.g., amqp://user:pass@host:port/ + Connection *amqp.Connection // The underlying RabbitMQ connection + Consumer *amqp.Channel // Channel for consuming messages + Producer *amqp.Channel // Channel for producing messages + confirmedProducer *amqp.Channel + confirmedReturns <-chan amqp.Return disasterRecoveryHandler DisasterRecoveryHandler // Optional handler for failed messages returnHandler ReturnHandler // Optional handler for unroutable (mandatory-returned) messages mu sync.Mutex + publishMu sync.Mutex } // NewRabbitMQ creates a new RabbitMQ with the provided RabbitMQ settings @@ -240,6 +251,9 @@ func (r *RabbitMQ) SetReturnHandler(handler ReturnHandler) { // Connect establishes the RabbitMQ connection and channels func (r *RabbitMQ) Connect() error { + r.publishMu.Lock() + defer r.publishMu.Unlock() + r.mu.Lock() defer r.mu.Unlock() @@ -273,10 +287,33 @@ func (r *RabbitMQ) connectLocked() error { _ = connection.Close() return err } + returns := producer.NotifyReturn(make(chan amqp.Return, 16)) + go r.watchReturns(returns) + + var confirmedProducer *amqp.Channel + var confirmedReturns <-chan amqp.Return + if r.options.ConfirmedDelivery { + confirmedProducer, err = connection.Channel() + if err != nil { + _ = producer.Close() + _ = connection.Close() + return err + } + confirmedReturns = confirmedProducer.NotifyReturn(make(chan amqp.Return, 1)) + if err := confirmedProducer.Confirm(false); err != nil { + _ = confirmedProducer.Close() + _ = producer.Close() + _ = connection.Close() + return err + } + } // Create channel for consuming, receiving messages. consumer, err := connection.Channel() if err != nil { + if confirmedProducer != nil { + _ = confirmedProducer.Close() + } _ = producer.Close() _ = connection.Close() return err @@ -285,6 +322,9 @@ func (r *RabbitMQ) connectLocked() error { err = consumer.Qos(prefetchCount, 0, false) if err != nil { _ = consumer.Close() + if confirmedProducer != nil { + _ = confirmedProducer.Close() + } _ = producer.Close() _ = connection.Close() return err @@ -294,29 +334,26 @@ func (r *RabbitMQ) connectLocked() error { err = r.declareQueue(consumer) if err != nil { _ = consumer.Close() + if confirmedProducer != nil { + _ = confirmedProducer.Close() + } _ = producer.Close() _ = connection.Close() return err } - // Surface unroutable publishes. With the default exchange a message published - // mandatory is returned when no queue is bound to its routing key (the target - // queue name) — a producer/consumer queue-name drift that would otherwise drop - // silently. Register the watcher on the new producer channel; it ends when that - // channel closes (closeResources below on the next reconnect), so every fresh - // producer channel gets its own watcher. - returns := producer.NotifyReturn(make(chan amqp.Return, 16)) - go r.watchReturns(returns) - oldConnection := r.Connection oldConsumer := r.Consumer oldProducer := r.Producer + oldConfirmedProducer := r.confirmedProducer r.Connection = connection r.Consumer = consumer r.Producer = producer + r.confirmedProducer = confirmedProducer + r.confirmedReturns = confirmedReturns - r.closeResources(oldConsumer, oldProducer, oldConnection) + r.closeResources(oldConsumer, oldProducer, oldConfirmedProducer, oldConnection) return nil } @@ -390,6 +427,16 @@ func (r *RabbitMQ) buildTLSConfig() (*tls.Config, error) { } func (r *RabbitMQ) ensureConnected() error { + r.mu.Lock() + if !r.needsReconnectLocked() { + r.mu.Unlock() + return nil + } + r.mu.Unlock() + + r.publishMu.Lock() + defer r.publishMu.Unlock() + r.mu.Lock() defer r.mu.Unlock() @@ -419,6 +466,9 @@ func (r *RabbitMQ) needsReconnectLocked() bool { if r.Producer == nil || r.Producer.IsClosed() { return true } + if r.options.ConfirmedDelivery && (r.confirmedProducer == nil || r.confirmedProducer.IsClosed()) { + return true + } return false } @@ -437,32 +487,40 @@ func (r *RabbitMQ) currentProducer() *amqp.Channel { return r.Producer } -func (r *RabbitMQ) closeResources(consumer, producer *amqp.Channel, connection *amqp.Connection) { +func (r *RabbitMQ) currentConfirmedPublisher() (*amqp.Channel, <-chan amqp.Return) { + r.mu.Lock() + defer r.mu.Unlock() + + return r.confirmedProducer, r.confirmedReturns +} + +func (r *RabbitMQ) closeResources(consumer, producer, confirmedProducer *amqp.Channel, connection *amqp.Connection) { if consumer != nil { _ = consumer.Close() } if producer != nil { _ = producer.Close() } + if confirmedProducer != nil { + _ = confirmedProducer.Close() + } if connection != nil { _ = connection.Close() } } -// watchReturns drains the producer channel's mandatory-return notifications until -// the channel closes (on a reconnect). Each amqp.Return is a message the broker -// could not route to any queue; with the default exchange that means no queue is -// bound to the routing key (the target queue name), i.e. a publish target with no -// consumer queue — a queue-name drift. It is delegated to the configured -// ReturnHandler, or logged via the standard logger so the condition is never silent. +func (r *RabbitMQ) reportReturn(ret amqp.Return) { + if r.returnHandler != nil { + r.returnHandler(ret) + return + } + log.Printf("queue: message returned as unroutable (no queue bound to routing key %q on exchange %q): replyCode=%d replyText=%q — likely a producer/consumer queue-name drift", + ret.RoutingKey, ret.Exchange, ret.ReplyCode, ret.ReplyText) +} + func (r *RabbitMQ) watchReturns(returns <-chan amqp.Return) { for ret := range returns { - if r.returnHandler != nil { - r.returnHandler(ret) - continue - } - log.Printf("queue: message returned as unroutable (no queue bound to routing key %q on exchange %q): replyCode=%d replyText=%q — likely a producer/consumer queue-name drift", - ret.RoutingKey, ret.Exchange, ret.ReplyCode, ret.ReplyText) + r.reportReturn(ret) } } @@ -475,6 +533,45 @@ func isClosedError(err error) bool { return strings.Contains(message, "channel/connection is not open") || strings.Contains(message, "connection is not open") } +func (r *RabbitMQ) settleTransferredDelivery(delivery amqp.Delivery, transfer func() error) error { + if err := transfer(); err != nil { + nackErr := delivery.Nack(false, true) + r.Close() + if nackErr != nil { + return fmt.Errorf("transfer failed: %w; requeue failed: %v", err, nackErr) + } + return err + } + + if err := delivery.Ack(false); err != nil { + r.Close() + return err + } + + return nil +} + +func (r *RabbitMQ) requireConfirmedDelivery() error { + if !r.options.ConfirmedDelivery { + return fmt.Errorf("confirmed RabbitMQ delivery is not enabled") + } + return nil +} + +func (r *RabbitMQ) deadletterAndSettle(delivery amqp.Delivery, payload []byte, confirmed bool) error { + if confirmed { + return r.settleTransferredDelivery(delivery, func() error { + return r.deadletterOrRecover(payload) + }) + } + + if err := r.AddToDeadletter(payload); err != nil { + _ = r.DisasterRecovery(payload) + } + _ = delivery.Ack(false) + return nil +} + // ReadMessages reads messages from the RabbitMQ queue, processes them using the provided handler, // and reports metrics using the provided Prometheus handler. It will then take action based on the handler's result. // Forwards, cancels, retries or sends to deadletter as needed. @@ -484,6 +581,24 @@ func isClosedError(err error) bool { // - handlePrometheus: function to handle metrics reporting // - args: additional arguments to pass to the message handler func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + return r.readMessages(false, handleMessage, handlePrometheus, args...) +} + +// ReadMessagesConfirmed processes PipelineEvents with confirmed forwarding and +// only acknowledges each source delivery after its transfer is durable. +func (r *RabbitMQ) ReadMessagesConfirmed(handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + if err := r.requireConfirmedDelivery(); err != nil { + return err + } + return r.readMessages(true, handleMessage, handlePrometheus, args...) +} + +func (r *RabbitMQ) readMessages(confirmed bool, handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + publish := r.publishLegacyWithRecovery + if confirmed { + publish = r.publishConfirmedWithReconnect + } + for { if err := r.ensureConnected(); err != nil { return err @@ -522,13 +637,10 @@ func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handleProme var pipelineEvent models.PipelineEvent err = json.Unmarshal(payload, &pipelineEvent) if err != nil { - // Failed to unmarshal - send to deadletter and ack to remove from queue - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - // Deadletter failed - use disaster recovery - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) // Ack even on error to prevent infinite redelivery - continue // Continue processing next message + continue } // We will override payload with the new payload @@ -552,42 +664,38 @@ func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handleProme // Marshal updated event pipelineEventPayload, err := json.Marshal(pipelineEvent) if err != nil { - // Marshal failed - send original payload to deadletter - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } topic := r.options.RouterQueue - err = r.Publish(topic, pipelineEventPayload) + err = publish(topic, pipelineEventPayload, nil) if err != nil { - // Publish failed - send to deadletter - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } case models.PipelineError: // Send to deadletter queue pipelineEventPayload, err := json.Marshal(pipelineEvent) if err != nil { - // Marshal failed - send original payload to deadletter - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } - topic := r.options.DeadletterQueue - err = r.Publish(topic, pipelineEventPayload) - if err != nil { - // Publish to deadletter failed - try with original payload - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if confirmed { + if err := r.deadletterAndSettle(d, pipelineEventPayload, true); err != nil { + return err + } + continue + } + if err := publish(r.options.DeadletterQueue, pipelineEventPayload, nil); err != nil { + if err := r.deadletterAndSettle(d, payload, false); err != nil { + return err } - d.Ack(false) continue } case models.PipelineCancel: @@ -597,7 +705,15 @@ func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handleProme // the attempt count on the message and dead-letters it once the cap is // hit, so a permanently failing payload can never loop the consumer // queue forever. - if err := r.retryOrDeadletter(d.Headers, payload, 5*time.Second, r.publishWithReconnect); err != nil { + if confirmed { + if err := r.settleTransferredDelivery(d, func() error { + return r.retryOrDeadletter(d.Headers, payload, 5*time.Second, publish) + }); err != nil { + return err + } + continue + } + if err := r.retryOrDeadletter(d.Headers, payload, 5*time.Second, publish); err != nil { return err } } @@ -605,9 +721,11 @@ func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handleProme // Always acknowledge messages regardless of sync mode err = d.Ack(false) if err != nil { - // Ack failed - try disaster recovery but continue processing - // The message may be redelivered, but we shouldn't stop the consumer - r.DisasterRecovery(payload) + if confirmed { + r.Close() + return err + } + _ = r.DisasterRecovery(payload) continue } @@ -642,6 +760,18 @@ func (r *RabbitMQ) ReadMessages(handleMessage models.MessageHandler, handleProme // failure is reported. The handler's replacement payload is ignored; this never // publishes onward. func (r *RabbitMQ) ReadOneRaw(handleMessage RawMessageHandler, args ...any) (bool, error) { + return r.readOneRaw(false, handleMessage, args...) +} + +// ReadOneRawConfirmed is ReadOneRaw with confirmed dead-letter settlement. +func (r *RabbitMQ) ReadOneRawConfirmed(handleMessage RawMessageHandler, args ...any) (bool, error) { + if err := r.requireConfirmedDelivery(); err != nil { + return false, err + } + return r.readOneRaw(true, handleMessage, args...) +} + +func (r *RabbitMQ) readOneRaw(confirmed bool, handleMessage RawMessageHandler, args ...any) (bool, error) { if err := r.ensureConnected(); err != nil { return false, err } @@ -663,9 +793,15 @@ func (r *RabbitMQ) ReadOneRaw(handleMessage RawMessageHandler, args ...any) (boo case models.PipelineCancel: return true, d.Ack(false) case models.PipelineError: - if pubErr := r.Publish(r.options.DeadletterQueue, d.Body); pubErr != nil { + if confirmed { + if err := r.deadletterAndSettle(d, d.Body, true); err != nil { + return true, err + } + return true, nil + } + if err := r.Publish(r.options.DeadletterQueue, d.Body); err != nil { if dlErr := r.AddToDeadletter(d.Body); dlErr != nil { - r.DisasterRecovery(d.Body) + _ = r.DisasterRecovery(d.Body) } } return true, d.Ack(false) @@ -693,6 +829,24 @@ type RawMessageHandler func(payload []byte, args ...any) (models.PipelineAction, // cancels). The connection, ack, deadletter and reconnect handling mirror // ReadMessages so behaviour is identical apart from the message shape. func (r *RabbitMQ) ReadRawMessages(handleMessage RawMessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + return r.readRawMessages(false, handleMessage, handlePrometheus, args...) +} + +// ReadRawMessagesConfirmed processes raw deliveries with confirmed forwarding +// and only acknowledges each source delivery after its transfer is durable. +func (r *RabbitMQ) ReadRawMessagesConfirmed(handleMessage RawMessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + if err := r.requireConfirmedDelivery(); err != nil { + return err + } + return r.readRawMessages(true, handleMessage, handlePrometheus, args...) +} + +func (r *RabbitMQ) readRawMessages(confirmed bool, handleMessage RawMessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + publish := r.publishLegacyWithRecovery + if confirmed { + publish = r.publishConfirmedWithReconnect + } + for { if err := r.ensureConnected(); err != nil { return err @@ -732,33 +886,45 @@ func (r *RabbitMQ) ReadRawMessages(handleMessage RawMessageHandler, handlePromet // with no router queue as a misconfiguration and dead-letter the // message rather than publishing to an empty queue name. if r.options.RouterQueue == "" { - if err := r.Publish(r.options.DeadletterQueue, payload); err != nil { - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if !confirmed { + if err := r.Publish(r.options.DeadletterQueue, payload); err != nil { + if dlErr := r.AddToDeadletter(payload); dlErr != nil { + _ = r.DisasterRecovery(payload) + } } + _ = d.Ack(false) + continue + } + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } forwardPayload := outPayload if forwardPayload == nil { forwardPayload = payload } - if err := r.Publish(r.options.RouterQueue, forwardPayload); err != nil { - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if err := publish(r.options.RouterQueue, forwardPayload, nil); err != nil { + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } case models.PipelineError: - if err := r.Publish(r.options.DeadletterQueue, payload); err != nil { - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if !confirmed { + if err := r.Publish(r.options.DeadletterQueue, payload); err != nil { + if dlErr := r.AddToDeadletter(payload); dlErr != nil { + _ = r.DisasterRecovery(payload) + } + _ = d.Ack(false) + continue } - d.Ack(false) - continue + break } + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err + } + continue case models.PipelineCancel: // Nothing to do, just acknowledge below. case models.PipelineRetry: @@ -767,13 +933,25 @@ func (r *RabbitMQ) ReadRawMessages(handleMessage RawMessageHandler, handlePromet // hit, so a permanently failing payload can never loop the consumer // queue forever (the workflow engine's marker-ingest retry that backed // this queue up). - if err := r.retryOrDeadletter(d.Headers, payload, time.Duration(backoff)*time.Second, r.publishWithReconnect); err != nil { + if confirmed { + if err := r.settleTransferredDelivery(d, func() error { + return r.retryOrDeadletter(d.Headers, payload, time.Duration(backoff)*time.Second, publish) + }); err != nil { + return err + } + continue + } + if err := r.retryOrDeadletter(d.Headers, payload, time.Duration(backoff)*time.Second, publish); err != nil { return err } } if err := d.Ack(false); err != nil { - r.DisasterRecovery(payload) + if confirmed { + r.Close() + return err + } + _ = r.DisasterRecovery(payload) continue } @@ -794,6 +972,24 @@ func (r *RabbitMQ) ReadRawMessages(handleMessage RawMessageHandler, handlePromet } func (r *RabbitMQ) RouteMessages(handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + return r.routeMessages(false, handleMessage, handlePrometheus, args...) +} + +// RouteMessagesConfirmed routes PipelineEvents with confirmed publishing and +// only acknowledges each source delivery after its transfer is durable. +func (r *RabbitMQ) RouteMessagesConfirmed(handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + if err := r.requireConfirmedDelivery(); err != nil { + return err + } + return r.routeMessages(true, handleMessage, handlePrometheus, args...) +} + +func (r *RabbitMQ) routeMessages(confirmed bool, handleMessage models.MessageHandler, handlePrometheus models.PrometheusHandler, args ...any) error { + publish := r.publishLegacyWithRecovery + if confirmed { + publish = r.publishConfirmedWithReconnect + } + for { if err := r.ensureConnected(); err != nil { return err @@ -832,25 +1028,20 @@ func (r *RabbitMQ) RouteMessages(handleMessage models.MessageHandler, handleProm var pipelineEvent models.PipelineEvent err = json.Unmarshal(payload, &pipelineEvent) if err != nil { - // Failed to unmarshal - send to deadletter and ack to remove from queue - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - // Deadletter failed - use disaster recovery - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) // Ack even on error to prevent infinite redelivery - continue // Continue processing next message + continue } if len(pipelineEvent.Stages) > 0 { nextQueue := pipelineEvent.Stages[0] nextQueue = r.formatQueueName(nextQueue) // Apply legacy naming convention, we will remove this later - err = r.Publish(nextQueue, payload) + err = publish(nextQueue, payload, nil) if err != nil { - // Publish failed - send to deadletter - if dlErr := r.AddToDeadletter(payload); dlErr != nil { - r.DisasterRecovery(payload) + if err := r.deadletterAndSettle(d, payload, confirmed); err != nil { + return err } - d.Ack(false) continue } } @@ -858,9 +1049,11 @@ func (r *RabbitMQ) RouteMessages(handleMessage models.MessageHandler, handleProm // Always acknowledge messages regardless of sync mode err = d.Ack(false) if err != nil { - // Ack failed - try disaster recovery but continue processing - // The message may be redelivered, but we shouldn't stop the consumer - r.DisasterRecovery(payload) + if confirmed { + r.Close() + return err + } + _ = r.DisasterRecovery(payload) continue } @@ -886,16 +1079,22 @@ func (r *RabbitMQ) RouteMessages(handleMessage models.MessageHandler, handleProm } func (r *RabbitMQ) Close() { + r.publishMu.Lock() + defer r.publishMu.Unlock() + r.mu.Lock() consumer := r.Consumer producer := r.Producer + confirmedProducer := r.confirmedProducer connection := r.Connection r.Consumer = nil r.Producer = nil + r.confirmedProducer = nil + r.confirmedReturns = nil r.Connection = nil r.mu.Unlock() - r.closeResources(consumer, producer, connection) + r.closeResources(consumer, producer, confirmedProducer, connection) } // formatQueueName applies legacy naming convention to queue names @@ -906,32 +1105,48 @@ func (r *RabbitMQ) formatQueueName(queueName string) string { // Publish sends a message immediately to the specified RabbitMQ queue func (r *RabbitMQ) Publish(queueName string, payload []byte) error { - return r.publishWithReconnect(queueName, payload, nil) + return r.publishLegacyWithRecovery(queueName, payload, nil) } -// publishWithReconnect publishes payload (optionally carrying headers) and, on a -// closed/stale channel, reconnects once and retries. It is the shared body -// behind Publish and the delayed retry path: headers is nil for an ordinary -// publish and carries the retry counter when a message is re-queued. -func (r *RabbitMQ) publishWithReconnect(queueName string, payload []byte, headers amqp.Table) error { +// PublishConfirmed publishes a persistent message and waits for RabbitMQ to +// confirm that it accepted the message into the destination queue. +func (r *RabbitMQ) PublishConfirmed(queueName string, payload []byte) error { + if err := r.requireConfirmedDelivery(); err != nil { + return err + } + err := r.publishConfirmedWithReconnect(queueName, payload, nil) + if err == nil { + return nil + } + if recoveryErr := r.DisasterRecovery(payload); recoveryErr != nil { + return fmt.Errorf("publish failed: %w; disaster recovery failed: %v", err, recoveryErr) + } + return err +} + +func (r *RabbitMQ) publishLegacyWithReconnect(queueName string, payload []byte, headers amqp.Table) error { if err := r.ensureConnected(); err != nil { - r.DisasterRecovery(payload) return err } - err := r.publish(queueName, payload, headers) + err := r.publishLegacy(queueName, payload, headers) if err != nil && (r.needsReconnect() || isClosedError(err)) { if reconnectErr := r.Reconnect(); reconnectErr == nil { - err = r.publish(queueName, payload, headers) + err = r.publishLegacy(queueName, payload, headers) } } + return err +} + +func (r *RabbitMQ) publishLegacyWithRecovery(queueName string, payload []byte, headers amqp.Table) error { + err := r.publishLegacyWithReconnect(queueName, payload, headers) if err != nil { - r.DisasterRecovery(payload) + _ = r.DisasterRecovery(payload) } return err } -func (r *RabbitMQ) publish(queueName string, payload []byte, headers amqp.Table) error { +func (r *RabbitMQ) publishLegacy(queueName string, payload []byte, headers amqp.Table) error { producer := r.currentProducer() if producer == nil { return fmt.Errorf("RabbitMQ producer channel is not initialized") @@ -940,17 +1155,79 @@ func (r *RabbitMQ) publish(queueName string, payload []byte, headers amqp.Table) ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) defer cancel() - return producer.PublishWithContext(ctx, + return producer.PublishWithContext(ctx, "", queueName, true, false, amqp.Publishing{ + ContentType: "application/json", + Headers: headers, + Body: payload, + }) +} + +// publishConfirmedWithReconnect publishes payload (optionally carrying headers) and, on a +// closed/stale channel, reconnects once and retries. It is the shared body +// behind confirmed publishing and confirmed consumer transfers. +func (r *RabbitMQ) publishConfirmedWithReconnect(queueName string, payload []byte, headers amqp.Table) error { + if err := r.ensureConnected(); err != nil { + return err + } + + err := r.publishConfirmed(queueName, payload, headers) + if err != nil && (r.needsReconnect() || isClosedError(err)) { + if reconnectErr := r.Reconnect(); reconnectErr == nil { + err = r.publishConfirmed(queueName, payload, headers) + } + } + return err +} + +func (r *RabbitMQ) publishConfirmed(queueName string, payload []byte, headers amqp.Table) error { + r.publishMu.Lock() + defer r.publishMu.Unlock() + + producer, returns := r.currentConfirmedPublisher() + if producer == nil { + return fmt.Errorf("RabbitMQ producer channel is not initialized") + } + + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + + confirmation, err := producer.PublishWithDeferredConfirmWithContext(ctx, "", // exchange queueName, // routing key (queue name) true, // mandatory: return (don't silently drop) a message no queue is bound to false, // immediate amqp.Publishing{ - ContentType: "application/json", - Headers: headers, - Body: payload, + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Headers: headers, + Body: payload, }, ) + if err != nil { + return err + } + if confirmation == nil { + return fmt.Errorf("RabbitMQ producer channel is not in confirm mode") + } + + acknowledged, err := confirmation.WaitContext(ctx) + if err != nil { + return fmt.Errorf("waiting for RabbitMQ publish confirmation: %w", err) + } + if !acknowledged { + return fmt.Errorf("RabbitMQ broker negatively acknowledged publish to %q", queueName) + } + + select { + case ret, ok := <-returns: + if !ok { + return fmt.Errorf("RabbitMQ return channel closed while publishing to %q", queueName) + } + go r.reportReturn(ret) + return fmt.Errorf("RabbitMQ returned publish to %q as unroutable: %s", ret.RoutingKey, ret.ReplyText) + default: + return nil + } } // PublishWithDelay sends a message to the specified RabbitMQ queue after a delay @@ -966,7 +1243,7 @@ func (r *RabbitMQ) PublishWithDelay(queueName string, payload []byte, backoff in func (r *RabbitMQ) publishWithDelayHeaders(queueName string, payload []byte, backoff int, headers amqp.Table) { go func() { time.Sleep(time.Duration(backoff) * time.Second) - _ = r.publishWithReconnect(queueName, payload, headers) + _ = r.publishLegacyWithRecovery(queueName, payload, headers) }() } @@ -1037,6 +1314,23 @@ func (r *RabbitMQ) AddToDeadletter(payload []byte) error { return r.Publish(topic, payload) } +func (r *RabbitMQ) addToDeadletterConfirmed(payload []byte) error { + return r.publishConfirmedWithReconnect(r.options.DeadletterQueue, payload, nil) +} + +func (r *RabbitMQ) deadletterOrRecover(payload []byte) error { + if err := r.addToDeadletterConfirmed(payload); err != nil { + if r.disasterRecoveryHandler == nil { + return fmt.Errorf("dead-letter publish failed and no disaster recovery handler is configured: %w", err) + } + if recoveryErr := r.DisasterRecovery(payload); recoveryErr != nil { + return fmt.Errorf("dead-letter publish failed: %w; disaster recovery failed: %v", err, recoveryErr) + } + } + + return nil +} + // DisasterRecovery handles messages that failed to publish // Uses the injected disaster recovery handler if provided, otherwise does nothing func (r *RabbitMQ) DisasterRecovery(payload []byte) error { diff --git a/pkg/queue/rabbitmq_test.go b/pkg/queue/rabbitmq_test.go index b894265..87e6265 100644 --- a/pkg/queue/rabbitmq_test.go +++ b/pkg/queue/rabbitmq_test.go @@ -6,8 +6,33 @@ import ( "os" "testing" "time" + + amqp "github.com/rabbitmq/amqp091-go" ) +type recordingAcknowledger struct { + acks int + nacks int + requeue bool + ackError error + nackError error +} + +func (a *recordingAcknowledger) Ack(_ uint64, _ bool) error { + a.acks++ + return a.ackError +} + +func (a *recordingAcknowledger) Nack(_ uint64, _ bool, requeue bool) error { + a.nacks++ + a.requeue = requeue + return a.nackError +} + +func (a *recordingAcknowledger) Reject(_ uint64, _ bool) error { + return nil +} + // TestRabbitOptionsValidation tests the validation of RabbitMQ options func TestRabbitOptionsValidation(t *testing.T) { tests := []struct { @@ -294,6 +319,18 @@ func TestRabbitOptionsBuilder(t *testing.T) { t.Errorf("expected Exchange to be empty, got '%s'", opts.Exchange) } }) + + t.Run("ConfirmedDeliveryIsOptIn", func(t *testing.T) { + disabled := NewRabbitOptions().Build() + if disabled.ConfirmedDelivery { + t.Fatal("expected confirmed delivery to be disabled by default") + } + + enabled := NewRabbitOptions().SetConfirmedDelivery(true).Build() + if !enabled.ConfirmedDelivery { + t.Fatal("expected SetConfirmedDelivery to enable confirmed delivery") + } + }) } // TestWorkflowsQueueAliases verifies the workflow-oriented builder aliases set @@ -447,6 +484,118 @@ func TestRabbitMQIntegration(t *testing.T) { queueName = "test-integration-queue" } + t.Run("ConfirmedPublishRejectsUnroutableMessage", func(t *testing.T) { + testQueue := fmt.Sprintf("%s-confirm-%d", queueName, time.Now().UnixNano()) + opts := NewRabbitOptions(). + SetConsumerQueue(testQueue). + SetDeadletterQueue(testQueue + "-dlq"). + SetHost(host). + SetUsername(username). + SetPassword(password). + SetConfirmedDelivery(true). + Build() + + rabbit, err := NewRabbitMQ(opts) + if err != nil { + t.Fatalf("create RabbitMQ client: %v", err) + } + if err := rabbit.Connect(); err != nil { + t.Fatalf("connect to RabbitMQ: %v", err) + } + defer rabbit.Close() + + returned := make(chan amqp.Return, 1) + rabbit.SetReturnHandler(func(ret amqp.Return) { + returned <- ret + }) + + missingQueue := testQueue + "-missing" + if err := rabbit.PublishConfirmed(missingQueue, []byte(`{"message":"unroutable"}`)); err == nil { + t.Fatal("expected an unroutable confirmed publish to fail") + } + select { + case ret := <-returned: + if ret.RoutingKey != missingQueue { + t.Fatalf("returned routing key = %q, want %q", ret.RoutingKey, missingQueue) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for the mandatory return handler") + } + }) + + t.Run("ConnectionInterruptionReconnectsAndRedelivers", func(t *testing.T) { + testQueue := fmt.Sprintf("%s-reconnect-%d", queueName, time.Now().UnixNano()) + opts := NewRabbitOptions(). + SetConsumerQueue(testQueue). + SetDeadletterQueue(testQueue + "-dlq"). + SetHost(host). + SetUsername(username). + SetPassword(password). + SetConfirmedDelivery(true). + Build() + + rabbit, err := NewRabbitMQ(opts) + if err != nil { + t.Fatalf("create RabbitMQ client: %v", err) + } + if err := rabbit.Connect(); err != nil { + t.Fatalf("connect to RabbitMQ: %v", err) + } + defer rabbit.Close() + + payload := []byte(`{"message":"redeliver"}`) + if err := rabbit.PublishConfirmed(testQueue, payload); err != nil { + t.Fatalf("publish test message: %v", err) + } + + delivery, ok, err := rabbit.currentConsumer().Get(testQueue, false) + if err != nil { + t.Fatalf("get test message: %v", err) + } + if !ok { + t.Fatal("expected the published message") + } + if delivery.DeliveryMode != amqp.Persistent { + t.Fatalf("delivery mode = %d, want persistent (%d)", delivery.DeliveryMode, amqp.Persistent) + } + + oldConnection := rabbit.Connection + if err := oldConnection.Close(); err != nil { + t.Fatalf("interrupt RabbitMQ connection: %v", err) + } + if err := rabbit.ensureConnected(); err != nil { + t.Fatalf("reconnect after interruption: %v", err) + } + if rabbit.Connection == oldConnection { + t.Fatal("expected reconnect to replace the interrupted connection") + } + + deadline := time.Now().Add(5 * time.Second) + for { + delivery, ok, err = rabbit.currentConsumer().Get(testQueue, false) + if err != nil { + t.Fatalf("get redelivered message: %v", err) + } + if ok { + break + } + if time.Now().After(deadline) { + t.Fatal("timed out waiting for unacked message redelivery") + } + time.Sleep(25 * time.Millisecond) + } + + if string(delivery.Body) != string(payload) { + t.Fatalf("redelivered body = %q, want %q", delivery.Body, payload) + } + if !delivery.Redelivered { + t.Fatal("expected interrupted unacked delivery to be marked redelivered") + } + if err := delivery.Ack(false); err != nil { + t.Fatalf("ack redelivered message: %v", err) + } + }) + t.Run("ConnectToRealRabbitMQ", func(t *testing.T) { // Build RabbitMQ options from environment variables opts := NewRabbitOptions(). @@ -682,6 +831,69 @@ func TestRabbitMQNeedsReconnectWithoutResources(t *testing.T) { } } +func TestConfirmedDeliveryRequiresOptIn(t *testing.T) { + rabbit := &RabbitMQ{options: &RabbitOptions{}} + + if err := rabbit.PublishConfirmed("target", []byte("payload")); err == nil { + t.Fatal("expected PublishConfirmed to require confirmed delivery") + } + if err := rabbit.ReadMessagesConfirmed(nil, nil); err == nil { + t.Fatal("expected ReadMessagesConfirmed to require confirmed delivery") + } + if err := rabbit.ReadRawMessagesConfirmed(nil, nil); err == nil { + t.Fatal("expected ReadRawMessagesConfirmed to require confirmed delivery") + } + if err := rabbit.RouteMessagesConfirmed(nil, nil); err == nil { + t.Fatal("expected RouteMessagesConfirmed to require confirmed delivery") + } + if _, err := rabbit.ReadOneRawConfirmed(nil); err == nil { + t.Fatal("expected ReadOneRawConfirmed to require confirmed delivery") + } +} + +func TestSettleTransferredDelivery(t *testing.T) { + t.Run("AcknowledgesAfterSuccessfulTransfer", func(t *testing.T) { + acknowledger := &recordingAcknowledger{} + delivery := amqp.Delivery{Acknowledger: acknowledger, DeliveryTag: 1} + + err := (&RabbitMQ{}).settleTransferredDelivery(delivery, func() error { return nil }) + if err != nil { + t.Fatalf("settle successful transfer: %v", err) + } + if acknowledger.acks != 1 || acknowledger.nacks != 0 { + t.Fatalf("acks = %d, nacks = %d; want one ack and no nacks", acknowledger.acks, acknowledger.nacks) + } + }) + + t.Run("RequeuesAfterFailedTransfer", func(t *testing.T) { + acknowledger := &recordingAcknowledger{} + delivery := amqp.Delivery{Acknowledger: acknowledger, DeliveryTag: 1} + transferError := errors.New("destination unavailable") + + err := (&RabbitMQ{}).settleTransferredDelivery(delivery, func() error { return transferError }) + if !errors.Is(err, transferError) { + t.Fatalf("settle error = %v, want %v", err, transferError) + } + if acknowledger.acks != 0 || acknowledger.nacks != 1 || !acknowledger.requeue { + t.Fatalf("acks = %d, nacks = %d, requeue = %t; want no ack and one requeue nack", acknowledger.acks, acknowledger.nacks, acknowledger.requeue) + } + }) + + t.Run("ReturnsAcknowledgementFailure", func(t *testing.T) { + ackError := errors.New("consumer channel closed") + acknowledger := &recordingAcknowledger{ackError: ackError} + delivery := amqp.Delivery{Acknowledger: acknowledger, DeliveryTag: 1} + + err := (&RabbitMQ{}).settleTransferredDelivery(delivery, func() error { return nil }) + if !errors.Is(err, ackError) { + t.Fatalf("settle error = %v, want %v", err, ackError) + } + if acknowledger.acks != 1 || acknowledger.nacks != 0 { + t.Fatalf("acks = %d, nacks = %d; want one attempted ack and no nacks", acknowledger.acks, acknowledger.nacks) + } + }) +} + // TestDisasterRecovery tests the DisasterRecovery function func TestDisasterRecovery(t *testing.T) { t.Run("WithHandler", func(t *testing.T) {