Log RabbitMQ dead-letter decisions before publication

README.md

@@ -77,6 +77,12 @@ testing can enable confirmed delivery and call `ReadMessagesToDeadletter`. The
 method preserves the original body and acknowledges it only after the broker
 confirms the dead-letter envelope.
 
+RabbitMQ can log one structured warning before each new dead-letter publication
+using `RabbitOptions.Logger` / `SetLogger`. The warning records the existing
+terminal reason and available correlation identifiers, without parsing or logging
+payloads. It records a request, not successful delivery.
+See [dead-letter request logging](README.rabbitmq.md#dead-letter-request-logging).
+
 RabbitMQ also offers an optional `RabbitOptions.DeadLetterObserver` callback for
 payload-free final DLQ publish outcomes, including failures and confirmation
 status. See [dead-letter publish outcomes](README.rabbitmq.md#dead-letter-publish-outcomes).

README.rabbitmq.md

@@ -65,6 +65,7 @@ methods or channels is required.
 | `MaxRetries` | No | Retry limit before dead-lettering; defaults to `10` |
 | `ConfirmedDelivery` | No | Provisions the confirmed producer used by explicit confirmed methods; defaults to `false` |
 | `DeadLetterObserver` | No | Optional `func(DeadLetterPublishEvent)` observing final DLQ publish outcomes |
+| `Logger` | No | Optional `*logrus.Logger` warning before a new DLQ publication is attempted |
 | `Exchange` | No | Compatibility field; the default exchange is used |
 | `Uri` | No | Compatibility field; connection uses host and credentials |
 | `TLS` | No | Enables an `amqps://` connection |
@@ -148,6 +149,66 @@ rabbit.SetReturnHandler(func(returned amqp.Return) {
 })
 ```
 
+### Dead-letter request logging
+
+Inject your application's logger to log the reason immediately before a logical
+new dead-letter publication is attempted:
+
+```go
+logger := logrus.New()
+logger.SetFormatter(&logrus.JSONFormatter{})
+
+options := queue.NewRabbitOptions().
+    SetConsumerQueue("kcloud-sequence-queue").
+    SetDeadletterQueue("dead-letter-queue").
+    SetHost("rabbitmq.example.com:5672").
+    SetUsername("username").
+    SetPassword("password").
+    SetLogger(logger).
+    Build()
+```
+
+Continue using the normal `ReadMessages` or `ReadMessagesConfirmed` handlers.
+The single WARN uses the shared `models/pkg/api` envelope:
+
+- `applicationStatusCode`: `pipeline_warning`
+- `entityStatusCode`: `dead_letter_requested`
+- `metadata.data`: `sourceQueue`, `deadLetterQueue`, `reason`, `attempts`
+- `metadata.traceId` and `metadata.mediaFileName`: copied only from an already
+  decoded pipeline event, when available. The media identifier prefers
+  `Payload.Key`, falling back to `FileName`.
+
+The existing `DeadLetterReason` catalog supplies the terminal reason:
+
+| Reason | Meaning |
+| --- | --- |
+| `unspecified` | Direct/manual dead-letter request without a more specific reason |
+| `malformed` | Invalid message JSON or an event that cannot be encoded |
+| `handler_error` | The handler returned `PipelineError` |
+| `retry_exhausted` | The retry count reached `MaxRetries` |
+| `publish_failed` | Publishing to the next stage failed |
+
+`attempts` is the retry count for exhaustion and otherwise the value known by the
+publication path (normally zero). Forced transfers may supply their own reason.
+Correlate the terminal request with the consumer's preceding safe error logs for
+the underlying application failure.
+
+This logs a **request, not success or durable delivery**. A publish failure still
+has a request log; reconnect attempts do not emit duplicate request logs.
+Redelivery can create another logical request. Payloads, whole events, signed
+URLs, credentials, and raw exception text are not logged.
+
+Logging does not parse payloads or envelopes. Raw/direct/forced transfers have no
+correlation fields unless the caller already supplied an event through the
+internal pipeline path. Successful forwards, cancellations, and retries below
+the limit do not log a DLQ request. Inspection, scanning, replay, discard, and
+restoration do not count as new requests.
+
+`Logger` is nil by default, with no log construction when disabled. Configure it
+before starting the client. No handler API, payload, headers, settlement behavior,
+or background worker changes are needed. `DeadLetterObserver` below remains a
+separate, optional mechanism; request logging does not enable it.
+
 ### Dead-letter publish outcomes
 
 Set `RabbitOptions.DeadLetterObserver` or use `SetDeadLetterObserver` before

go.mod

@@ -10,6 +10,7 @@ require (
 	github.com/confluentinc/confluent-kafka-go/v2 v2.14.1
 	github.com/go-playground/validator/v10 v10.30.2
 	github.com/rabbitmq/amqp091-go v1.10.0
+	github.com/sirupsen/logrus v1.9.3
 	github.com/uug-ai/models v1.2.26
 )
 

go.sum

@@ -74,6 +74,8 @@ github.com/containerd/typeurl/v2 v2.1.1 h1:3Q4Pt7i8nYwy2KmQWIw2+1hTvwTE/6w9Fqctt
 github.com/containerd/typeurl/v2 v2.1.1/go.mod h1:IDp2JFvbwZ31H8dQbEIY7sDl2L3o3HZj1hsSQlywkQ0=
 github.com/cpuguy83/dockercfg v0.3.1 h1:/FpZ+JaygUR/lZP2NlFI2DVfrOEMAIKP5wWEJdoYe9E=
 github.com/cpuguy83/dockercfg v0.3.1/go.mod h1:sugsbF4//dDlL/i+S+rtpIWp+5h0BHJHfjj5/jFyUJc=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc h1:U9qPSI2PIWSS1VwoXQT9A3Wy9MM3WgvqSxFWenqJduM=
 github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/distribution/reference v0.6.0 h1:0IXCQ5g4/QMHHkarYzh5l+u8T3t73zM5QvfrDyIgxBk=
@@ -244,6 +246,7 @@ github.com/pelletier/go-toml v1.9.5 h1:4yBQzkHv+7BHq2PQUZF3Mx0IYxG7LsP222s7Agd3v
 github.com/pelletier/go-toml v1.9.5/go.mod h1:u1nR/EPcESfeI/szUZKdtJ0xRNbUoANCkoOuaOx1Y+c=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2 h1:Jamvg5psRIccs7FGNTlIRMkT8wgtp5eCXdBlqhYGL6U=
 github.com/pmezard/go-difflib v1.0.1-0.20181226105442-5d4384ee4fb2/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw=
@@ -280,6 +283,8 @@ github.com/spf13/cobra v1.8.1 h1:e5/vxKd/rZsfSJMUX1agtjeTDf+qv1/JdBF8gg5k9ZM=
 github.com/spf13/cobra v1.8.1/go.mod h1:wHxEcudfqmLYa8iTfL+OuZPbBZkmvliBWKIezN3kD9Y=
 github.com/spf13/pflag v1.0.5 h1:iy+VFUOCP1a+8yFto/drg2CJ5u0yRoB7fZw3DKv/JXA=
 github.com/spf13/pflag v1.0.5/go.mod h1:McXfInJRrz4CZXVZOBLb0bTZqETkiAhM9Iw0y3An2Bg=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
 github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
 github.com/testcontainers/testcontainers-go v0.33.0 h1:zJS9PfXYT5O0ZFXM2xxXfk4J5UMw/kRiISng037Gxdw=
@@ -356,6 +361,7 @@ golang.org/x/oauth2 v0.18.0 h1:09qnuIAgzdx1XplqJvW6CQqMCtGZykZWcXzPMPUusvI=
 golang.org/x/oauth2 v0.18.0/go.mod h1:Wf7knwG0MPoWIMMBgFlEaSUDaKskp0dCfrlJRJXbBi8=
 golang.org/x/sync v0.21.0 h1:HLII4xRRTtCRkxYp4HNFF0Js/Og6q2i++KXbg0gHCwM=
 golang.org/x/sync v0.21.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.0.0-20220715151400-c0bba94af5f8/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.44.0 h1:ildZl3J4uzeKP07r2F++Op7E9B29JRUy+a27EibtBTQ=
 golang.org/x/sys v0.44.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/term v0.43.0 h1:S4RLU2sB31O/NCl+zFN9Aru9A/Cq2aqKpTZJ6B+DwT4=
@@ -378,10 +384,12 @@ google.golang.org/protobuf v1.33.0 h1:uNO2rsAINq/JlFpSdYEKIZ0uKD/R9cpdv0T+yoGwGm
 google.golang.org/protobuf v1.33.0/go.mod h1:c6P6GXX6sHbq/GpV6MGZEdwhWPcYBgnhAHhKbcUYpos=
 gopkg.in/cenkalti/backoff.v1 v1.1.0 h1:Arh75ttbsvlpVA7WtVpH4u9h6Zl46xuptxqLxPiSo4Y=
 gopkg.in/cenkalti/backoff.v1 v1.1.0/go.mod h1:J6Vskwqd+OMVJl8C33mmtxTBs2gyzfv7UDAkHu8BrjI=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
 gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc=
 gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw=
 gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
 gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 k8s.io/api v0.29.2 h1:hBC7B9+MU+ptchxEqTNW2DkUosJpp1P+Wn6YncZ474A=

pkg/queue/deadletter_log.go

@@ -0,0 +1,38 @@
+package queue
+
+import (
+	"github.com/uug-ai/models/pkg/api"
+	"github.com/uug-ai/models/pkg/models"
+)
+
+func (r *RabbitMQ) logDeadLetterRequest(reason DeadLetterReason, attempts int, events ...*models.PipelineEvent) {
+	r.logDeadLetterRequestFromSource(r.options.ConsumerQueue, reason, attempts, events...)
+}
+
+func (r *RabbitMQ) logDeadLetterRequestFromSource(source string, reason DeadLetterReason, attempts int, events ...*models.PipelineEvent) {
+	logger := r.options.Logger
+	if logger == nil {
+		return
+	}
+	if reason == "" {
+		reason = DeadLetterReasonUnspecified
+	}
+	metadata := api.Metadata{
+		Data: map[string]any{
+			"sourceQueue":     source,
+			"deadLetterQueue": r.options.DeadletterQueue,
+			"reason":          reason,
+			"attempts":        attempts,
+		},
+	}
+	if len(events) > 0 && events[0] != nil {
+		event := events[0]
+		metadata.TraceId = event.TraceId
+		metadata.MediaFileName = event.Payload.FileName
+		if metadata.MediaFileName == "" {
+			metadata.MediaFileName = event.FileName
+		}
+	}
+	warning := api.CreateWarning(api.HttpNoStatus, api.PipelineWarning, api.PipelineStatus("dead_letter_requested"), metadata)
+	logger.WithFields(api.CreateWarningLog(logger, warning)).Warn("Dead-letter publication requested")
+}