From 479e4341ed4d396506fa4212b3a53c7bf2706375 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Wed, 16 Sep 2026 14:30:26 +0000 Subject: [PATCH 1/6] feat(queue): add confirmed raw DLQ transfer Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- README.md | 5 ++++ pkg/queue/deadletter_rabbitmq.go | 39 +++++++++++++++++++++++++++ pkg/queue/deadletter_rabbitmq_test.go | 19 +++++++++++++ pkg/queue/rabbitmq.go | 23 +++++----------- pkg/queue/retry_test.go | 29 ++++++++++++++++---- 5 files changed, 94 insertions(+), 21 deletions(-) diff --git a/README.md b/README.md index 3480545..68aab73 100644 --- a/README.md +++ b/README.md @@ -72,6 +72,11 @@ Dead-letter inspection and replay are deliberately exposed through the separate administrative methods, while operational tooling can use one contract across all supported brokers. +RabbitMQ services that need to deliberately park raw deliveries for operational +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. + ### Pipeline Actions Message handlers return a `models.PipelineAction`. The queue client maps that diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index 1efeede..50d6820 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -25,6 +25,45 @@ func (r *RabbitMQ) PublishDeadLetter(ctx context.Context, payload []byte, metada return r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, envelope) } +// ReadMessagesToDeadletter transfers raw consumer deliveries into versioned +// dead-letter envelopes. Each source delivery is acknowledged only after the +// broker confirms the dead-letter publish. +func (r *RabbitMQ) ReadMessagesToDeadletter(reason DeadLetterReason) error { + if err := r.requireConfirmedDelivery(); err != nil { + return err + } + if reason == "" { + reason = DeadLetterReasonUnspecified + } + if err := r.ensureConnected(); err != nil { + return err + } + consumer := r.currentConsumer() + if consumer == nil { + return fmt.Errorf("RabbitMQ consumer channel is not initialized") + } + deliveries, err := consumer.Consume( + r.options.ConsumerQueue, + "", + false, + false, + false, + false, + nil, + ) + if err != nil { + return err + } + for delivery := range deliveries { + if err := r.settleTransferredDelivery(delivery, func() error { + return r.addToDeadletterConfirmed(delivery.Body, reason) + }); err != nil { + return err + } + } + return fmt.Errorf("RabbitMQ consumer channel closed") +} + func (r *RabbitMQ) InspectDeadLetters(ctx context.Context, request DeadLetterInspectRequest) (DeadLetterInspectResult, error) { var result DeadLetterInspectResult if err := validateDeadLetterLimit(request.Limit); err != nil { diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index 27ce696..1068b48 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -3,6 +3,7 @@ package queue import ( "bytes" "context" + "strings" "testing" amqp "github.com/rabbitmq/amqp091-go" @@ -155,6 +156,24 @@ func TestRabbitPublishDeadLetterUsesTrustedAdministrativeMetadata(t *testing.T) } } +func TestRabbitReadMessagesToDeadletterRequiresConfirmedDelivery(t *testing.T) { + client, err := NewRabbitMQ(NewRabbitOptions(). + SetConsumerQueue("events"). + SetDeadletterQueue("deadletter"). + SetHost("rabbitmq:5672"). + SetUsername("guest"). + SetPassword("guest"). + Build()) + if err != nil { + t.Fatalf("NewRabbitMQ: %v", err) + } + + err = client.ReadMessagesToDeadletter(DeadLetterReasonHandlerError) + if err == nil || !strings.Contains(err.Error(), "confirmed RabbitMQ delivery is not enabled") { + t.Fatalf("error = %v, want confirmed delivery requirement", err) + } +} + func TestRabbitDeadLetterEnvelopePreservesFailureReason(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). diff --git a/pkg/queue/rabbitmq.go b/pkg/queue/rabbitmq.go index e8d61b4..7051363 100644 --- a/pkg/queue/rabbitmq.go +++ b/pkg/queue/rabbitmq.go @@ -46,9 +46,8 @@ type RabbitOptions struct { VirtualHost string // MaxRetries caps how many times a PipelineRetry re-queues a message before - // it is parked on the deadletter queue; zero selects defaultMaxRetries. It is - // the hard stop that keeps a permanently failing payload from looping the - // consumer queue forever (see retryOrDeadletter). + // it is parked on the deadletter queue. Zero preserves the historical + // unlimited retry behavior. MaxRetries int // TLS configuration for secure connections (e.g., AWS Amazon MQ) @@ -154,9 +153,7 @@ func (b *RabbitOptionsBuilder) SetPrefetchCount(count int) *RabbitOptionsBuilder } // SetMaxRetries caps how many times a PipelineRetry re-queues a message before -// it is dead-lettered. Zero (the default) selects defaultMaxRetries. Set it to -// bound how long a transient sink failure is retried before the message is -// parked for inspection instead of being requeued forever. +// it is dead-lettered. Zero (the default) preserves unlimited retries. func (b *RabbitOptionsBuilder) SetMaxRetries(maxRetries int) *RabbitOptionsBuilder { b.options.MaxRetries = maxRetries return b @@ -1282,18 +1279,12 @@ func (r *RabbitMQ) publishWithDelayHeaders(queueName string, payload []byte, bac // first delivery (count 0) and set explicitly on every re-queue. const retryCountHeader = "x-retry-count" -// defaultMaxRetries bounds PipelineRetry re-queues when RabbitOptions.MaxRetries -// is unset. With the default 5s backoff this rides out a brief sink outage -// (~50s) before the message is dead-lettered instead of looping forever. +// defaultMaxRetries preserves the bounded defaults used by Kafka and SQS. const defaultMaxRetries = 10 -// maxRetries returns the configured PipelineRetry cap, or defaultMaxRetries when -// unset. +// maxRetries returns the configured PipelineRetry cap. Zero means unlimited. func (r *RabbitMQ) maxRetries() int { - if r.options.MaxRetries > 0 { - return r.options.MaxRetries - } - return defaultMaxRetries + return r.options.MaxRetries } // retryCount reads the x-retry-count header off a delivery, tolerating the @@ -1324,7 +1315,7 @@ func retryCount(headers amqp.Table) int { // after this returns nil, so a failed retry publish cannot silently lose it. func (r *RabbitMQ) retryOrDeadletter(headers amqp.Table, payload []byte, backoff time.Duration, publish func(string, []byte, amqp.Table) error) error { attempts := retryCount(headers) - if attempts >= r.maxRetries() { + if maxRetries := r.maxRetries(); maxRetries > 0 && attempts >= maxRetries { envelope, err := r.deadLetterEnvelope(payload, DeadLetterReasonRetryExhausted, attempts) if err != nil { return err diff --git a/pkg/queue/retry_test.go b/pkg/queue/retry_test.go index 6a85fb2..ac84ee0 100644 --- a/pkg/queue/retry_test.go +++ b/pkg/queue/retry_test.go @@ -31,21 +31,40 @@ func TestSetMaxRetriesBuilder(t *testing.T) { if opts := NewRabbitOptions().SetMaxRetries(7).Build(); opts.MaxRetries != 7 { t.Errorf("MaxRetries = %d, want 7", opts.MaxRetries) } - // Unset leaves the zero value so the client falls back to defaultMaxRetries. + // Unset preserves the historical unlimited retry behavior. if def := NewRabbitOptions().Build(); def.MaxRetries != 0 { - t.Errorf("default MaxRetries = %d, want 0 (selects defaultMaxRetries at runtime)", def.MaxRetries) + t.Errorf("default MaxRetries = %d, want 0", def.MaxRetries) } } -func TestMaxRetriesDefault(t *testing.T) { - if got := newTestRabbit(t, 0).maxRetries(); got != defaultMaxRetries { - t.Errorf("maxRetries() with unset option = %d, want default %d", got, defaultMaxRetries) +func TestMaxRetries(t *testing.T) { + if got := newTestRabbit(t, 0).maxRetries(); got != 0 { + t.Errorf("maxRetries() with unset option = %d, want unlimited (0)", got) } if got := newTestRabbit(t, 3).maxRetries(); got != 3 { t.Errorf("maxRetries() = %d, want 3", got) } } +func TestRetryOrDeadletterRetriesWithoutCap(t *testing.T) { + rabbit := newTestRabbit(t, 0) + var queueName string + + err := rabbit.retryOrDeadletter(amqp.Table{retryCountHeader: int32(100)}, []byte("payload"), time.Nanosecond, func(gotQueue string, _ []byte, headers amqp.Table) error { + queueName = gotQueue + if got := retryCount(headers); got != 101 { + t.Fatalf("retry header = %d, want 101", got) + } + return nil + }) + if err != nil { + t.Fatalf("retryOrDeadletter() error = %v", err) + } + if queueName != "c" { + t.Fatalf("queue = %q, want consumer queue", queueName) + } +} + // TestRetryCount covers the header decode that drives the retry cap: a missing // header means a first attempt (0), and the several integer types AMQP may // decode the counter as are all accepted. From 032f67be4a586042c8cd8e6ac7ab149db24bd6f7 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Wed, 16 Sep 2026 15:16:59 +0000 Subject: [PATCH 2/6] feat(queue): route DLQ replays through configured router Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- README.md | 8 +++ pkg/queue/deadletter.go | 53 ++++++++++++++------ pkg/queue/deadletter_rabbitmq_test.go | 10 +++- pkg/queue/deadletter_test.go | 70 +++++++++++++++++++++++++-- pkg/queue/kafka.go | 13 ++--- pkg/queue/kafka_test.go | 17 +++++++ pkg/queue/rabbitmq.go | 13 ++--- pkg/queue/sqs.go | 13 ++--- pkg/queue/sqs_test.go | 15 ++++++ 9 files changed, 173 insertions(+), 39 deletions(-) diff --git a/README.md b/README.md index 68aab73..cbfe99a 100644 --- a/README.md +++ b/README.md @@ -99,6 +99,8 @@ writes should therefore be idempotent. - The exact original payload, encoded safely even when it is not valid JSON - The source queue or topic - The dead-letter destination +- The replay destination: the configured pipeline router, or the source when no + router is configured - The failure reason, attempt count, and UTC timestamp - Optional service and provider-neutral attributes @@ -113,6 +115,12 @@ Existing raw dead-letter messages remain readable. They are reported with the source `unknown` and require an explicit replay destination because the library cannot safely infer where they came from. +Replay prefers an operator-supplied destination, then the destination recorded +in the envelope, and finally the source queue for envelopes written before +`replayDestination` was introduced. Pipeline workers therefore replay through +their configured router, which resolves the next queue from the event's +remaining stages, while non-pipeline workers preserve source-queue behavior. + ## Inspecting and Replaying Dead-Letter Messages The operational commands are provided by the separate diff --git a/pkg/queue/deadletter.go b/pkg/queue/deadletter.go index 93bde99..8de2293 100644 --- a/pkg/queue/deadletter.go +++ b/pkg/queue/deadletter.go @@ -25,13 +25,14 @@ const ( // DeadLetterMetadata describes where and why a message was parked. type DeadLetterMetadata struct { - Source string `json:"source,omitempty"` - Destination string `json:"destination"` - Service string `json:"service,omitempty"` - Reason DeadLetterReason `json:"reason"` - Attempts int `json:"attempts,omitempty"` - Timestamp time.Time `json:"timestamp"` - Attributes map[string]string `json:"attributes,omitempty"` + Source string `json:"source,omitempty"` + Destination string `json:"destination"` + ReplayDestination string `json:"replayDestination,omitempty"` + Service string `json:"service,omitempty"` + Reason DeadLetterReason `json:"reason"` + Attempts int `json:"attempts,omitempty"` + Timestamp time.Time `json:"timestamp"` + Attributes map[string]string `json:"attributes,omitempty"` } // DeadLetterEnvelope is the provider-independent format stored on dead-letter @@ -79,13 +80,14 @@ type DeadLetterReplayRequest struct { } type DeadLetterReplayResult struct { - Scanned int - Matched int - Planned int - Replayed int - Retained int - Legacy int - Unroutable int + Scanned int + Matched int + Planned int + Replayed int + Retained int + Legacy int + Unroutable int + Destinations map[string]int } // DeadLetterAdmin is intentionally separate from QueueInterface: applications @@ -113,6 +115,20 @@ func encodeDeadLetter(payload []byte, metadata DeadLetterMetadata) ([]byte, erro }) } +func runtimeDeadLetterMetadata(source, deadLetterDestination, routerDestination string, reason DeadLetterReason, attempts int) DeadLetterMetadata { + replayDestination := strings.TrimSpace(routerDestination) + if replayDestination == "" { + replayDestination = strings.TrimSpace(source) + } + return DeadLetterMetadata{ + Source: source, + Destination: deadLetterDestination, + ReplayDestination: replayDestination, + Reason: reason, + Attempts: attempts, + } +} + func encodeDeadLetterForDestination(payload []byte, metadata DeadLetterMetadata, destination string) ([]byte, error) { if metadata.Source == "" { return nil, fmt.Errorf("dead-letter source is required") @@ -237,7 +253,10 @@ func planDeadLetterReplay(result *DeadLetterReplayResult, message DeadLetterMess result.Matched++ destination := strings.TrimSpace(request.Destination) if destination == "" && !message.Legacy { - destination = strings.TrimSpace(message.DeadLetter.Source) + destination = strings.TrimSpace(message.DeadLetter.ReplayDestination) + if destination == "" { + destination = strings.TrimSpace(message.DeadLetter.Source) + } } if destination == "" { result.Unroutable++ @@ -248,6 +267,10 @@ func planDeadLetterReplay(result *DeadLetterReplayResult, message DeadLetterMess return deadLetterReplayPlan{}, fmt.Errorf("refusing to replay dead-letter message %q back to %q", message.ID, deadLetterDestination) } result.Planned++ + if result.Destinations == nil { + result.Destinations = make(map[string]int) + } + result.Destinations[destination]++ if !request.Execute { result.Retained++ } diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index 1068b48..cfcfbeb 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -90,6 +90,7 @@ func TestRabbitDeadLetterReplayPublishesBeforeAck(t *testing.T) { func TestRabbitPublishToDeadLetterQueueAddsEnvelope(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). + SetRouterQueue("router"). SetDeadletterQueue("deadletter"). SetHost("rabbitmq:5672"). SetUsername("guest"). @@ -114,7 +115,10 @@ func TestRabbitPublishToDeadLetterQueueAddsEnvelope(t *testing.T) { if err != nil { t.Fatal(err) } - if message.Legacy || message.DeadLetter.Source != "events" || string(message.Payload) != "payload" { + if message.Legacy || + message.DeadLetter.Source != "events" || + message.DeadLetter.ReplayDestination != "router" || + string(message.Payload) != "payload" { t.Fatalf("dead-letter message = %+v", message) } } @@ -177,6 +181,7 @@ func TestRabbitReadMessagesToDeadletterRequiresConfirmedDelivery(t *testing.T) { func TestRabbitDeadLetterEnvelopePreservesFailureReason(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). + SetRouterQueue("router"). SetDeadletterQueue("deadletter"). SetHost("rabbitmq:5672"). SetUsername("guest"). @@ -203,6 +208,9 @@ func TestRabbitDeadLetterEnvelopePreservesFailureReason(t *testing.T) { if message.DeadLetter.Reason != reason { t.Fatalf("reason = %q, want %q", message.DeadLetter.Reason, reason) } + if message.DeadLetter.ReplayDestination != "router" { + t.Fatalf("replay destination = %q, want router", message.DeadLetter.ReplayDestination) + } }) } } diff --git a/pkg/queue/deadletter_test.go b/pkg/queue/deadletter_test.go index 2fe81fa..81805a3 100644 --- a/pkg/queue/deadletter_test.go +++ b/pkg/queue/deadletter_test.go @@ -10,10 +10,11 @@ func TestDeadLetterEnvelopeRoundTripPreservesMalformedPayload(t *testing.T) { payload := []byte{0xff, 0x00, '{'} timestamp := time.Date(2026, 9, 16, 9, 48, 15, 0, time.UTC) encoded, err := encodeDeadLetter(payload, DeadLetterMetadata{ - Source: "events", - Destination: "deadletter", - Reason: DeadLetterReasonMalformed, - Timestamp: timestamp, + Source: "events", + Destination: "deadletter", + ReplayDestination: "router", + Reason: DeadLetterReasonMalformed, + Timestamp: timestamp, }) if err != nil { t.Fatalf("encodeDeadLetter: %v", err) @@ -25,7 +26,9 @@ func TestDeadLetterEnvelopeRoundTripPreservesMalformedPayload(t *testing.T) { if message.Legacy || !bytes.Equal(message.Payload, payload) { t.Fatalf("decoded message = %+v", message) } - if message.DeadLetter.Source != "events" || message.DeadLetter.Timestamp != timestamp { + if message.DeadLetter.Source != "events" || + message.DeadLetter.ReplayDestination != "router" || + message.DeadLetter.Timestamp != timestamp { t.Fatalf("decoded metadata = %+v", message.DeadLetter) } } @@ -104,3 +107,60 @@ func TestReplayPlanRequiresDestinationForLegacyMessage(t *testing.T) { t.Fatalf("plan = %+v, result = %+v", plan, result) } } + +func TestReplayPlanDestinationPrecedence(t *testing.T) { + message := DeadLetterMessage{ + ID: "message-1", + DeadLetter: DeadLetterMetadata{ + Source: "sequence", + ReplayDestination: "event", + }, + } + tests := []struct { + name string + request DeadLetterReplayRequest + destination string + }{ + { + name: "recorded router", + request: DeadLetterReplayRequest{Limit: 1}, + destination: "event", + }, + { + name: "explicit override", + request: DeadLetterReplayRequest{Limit: 1, Destination: "manual"}, + destination: "manual", + }, + { + name: "older envelope fallback", + request: DeadLetterReplayRequest{Limit: 1}, + destination: "sequence", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + replayMessage := message + if test.name == "older envelope fallback" { + replayMessage.DeadLetter.ReplayDestination = "" + } + var result DeadLetterReplayResult + plan, err := planDeadLetterReplay(&result, replayMessage, test.request, "deadletter") + if err != nil { + t.Fatalf("planDeadLetterReplay: %v", err) + } + if plan.destination != test.destination || + result.Planned != 1 || + result.Destinations[test.destination] != 1 { + t.Fatalf("plan = %+v, result = %+v", plan, result) + } + }) + } +} + +func TestRuntimeDeadLetterMetadataFallsBackToSourceWithoutRouter(t *testing.T) { + metadata := runtimeDeadLetterMetadata("sequence", "deadletter", "", DeadLetterReasonHandlerError, 2) + if metadata.ReplayDestination != "sequence" { + t.Fatalf("replay destination = %q, want sequence", metadata.ReplayDestination) + } +} diff --git a/pkg/queue/kafka.go b/pkg/queue/kafka.go index 6227cb2..029e9ea 100644 --- a/pkg/queue/kafka.go +++ b/pkg/queue/kafka.go @@ -389,12 +389,13 @@ func (k *Kafka) AddToDeadletter(payload []byte) error { } func (k *Kafka) addToDeadletter(payload []byte, reason DeadLetterReason, attempts int) error { - envelope, err := encodeDeadLetter(payload, DeadLetterMetadata{ - Source: k.options.ConsumerTopic, - Destination: k.options.DeadletterTopic, - Reason: reason, - Attempts: attempts, - }) + envelope, err := encodeDeadLetter(payload, runtimeDeadLetterMetadata( + k.options.ConsumerTopic, + k.options.DeadletterTopic, + k.options.RouterTopic, + reason, + attempts, + )) if err != nil { return err } diff --git a/pkg/queue/kafka_test.go b/pkg/queue/kafka_test.go index 410ee6c..dca55e6 100644 --- a/pkg/queue/kafka_test.go +++ b/pkg/queue/kafka_test.go @@ -121,6 +121,23 @@ func newTestKafka(t *testing.T, message *kafka.Message) (*Kafka, *fakeKafkaConsu return client, consumer, producer } +func TestKafkaDeadLetterEnvelopeRecordsRouter(t *testing.T) { + client, _, producer := newTestKafka(t, nil) + if err := client.addToDeadletter([]byte("payload"), DeadLetterReasonHandlerError, 0); err != nil { + t.Fatal(err) + } + if len(producer.messages) != 1 { + t.Fatalf("published messages = %d, want 1", len(producer.messages)) + } + message, err := decodeDeadLetter("message-1", producer.messages[0].Value) + if err != nil { + t.Fatal(err) + } + if message.DeadLetter.Source != "events" || message.DeadLetter.ReplayDestination != "router" { + t.Fatalf("dead-letter metadata = %+v", message.DeadLetter) + } +} + func TestKafkaDeadLetterReplayPublishesBeforeCommit(t *testing.T) { envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ Source: "events", diff --git a/pkg/queue/rabbitmq.go b/pkg/queue/rabbitmq.go index 7051363..311af0b 100644 --- a/pkg/queue/rabbitmq.go +++ b/pkg/queue/rabbitmq.go @@ -1355,12 +1355,13 @@ func (r *RabbitMQ) addToDeadletter(payload []byte, reason DeadLetterReason, atte } func (r *RabbitMQ) deadLetterEnvelope(payload []byte, reason DeadLetterReason, attempts int) ([]byte, error) { - return encodeDeadLetter(payload, DeadLetterMetadata{ - Source: r.options.ConsumerQueue, - Destination: r.options.DeadletterQueue, - Reason: reason, - Attempts: attempts, - }) + return encodeDeadLetter(payload, runtimeDeadLetterMetadata( + r.options.ConsumerQueue, + r.options.DeadletterQueue, + r.options.RouterQueue, + reason, + attempts, + )) } func (r *RabbitMQ) addToDeadletterConfirmed(payload []byte, reason DeadLetterReason) error { diff --git a/pkg/queue/sqs.go b/pkg/queue/sqs.go index fcd3032..b4b3c24 100644 --- a/pkg/queue/sqs.go +++ b/pkg/queue/sqs.go @@ -458,12 +458,13 @@ func (s *SQS) addToDeadletter(payload []byte, reason DeadLetterReason, attempts } func (s *SQS) deadLetterEnvelope(payload []byte, reason DeadLetterReason, attempts int) ([]byte, error) { - return encodeDeadLetter(payload, DeadLetterMetadata{ - Source: s.options.ConsumerQueue, - Destination: s.options.DeadletterQueue, - Reason: reason, - Attempts: attempts, - }) + return encodeDeadLetter(payload, runtimeDeadLetterMetadata( + s.options.ConsumerQueue, + s.options.DeadletterQueue, + s.options.RouterQueue, + reason, + attempts, + )) } func (s *SQS) SetDisasterRecoveryHandler(handler DisasterRecoveryHandler) { diff --git a/pkg/queue/sqs_test.go b/pkg/queue/sqs_test.go index 8d4fe65..333458b 100644 --- a/pkg/queue/sqs_test.go +++ b/pkg/queue/sqs_test.go @@ -121,6 +121,21 @@ func TestSQSDeadLetterReplaySendsBeforeDelete(t *testing.T) { } } +func TestSQSDeadLetterEnvelopeRecordsRouter(t *testing.T) { + client, _ := newTestSQS(t, types.Message{}) + envelope, err := client.deadLetterEnvelope([]byte("payload"), DeadLetterReasonHandlerError, 0) + if err != nil { + t.Fatal(err) + } + message, err := decodeDeadLetter("message-1", envelope) + if err != nil { + t.Fatal(err) + } + if message.DeadLetter.Source != "events" || message.DeadLetter.ReplayDestination != "router" { + t.Fatalf("dead-letter metadata = %+v", message.DeadLetter) + } +} + func TestSQSDeadLetterInspectReleasesMessages(t *testing.T) { message := types.Message{ MessageId: aws.String("legacy-1"), From 6c724af0672056658a8ad02fe08de65d888bbc4d Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Wed, 16 Sep 2026 15:25:46 +0000 Subject: [PATCH 3/6] fix(queue): preserve safe retry and transfer defaults Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- pkg/queue/deadletter_rabbitmq.go | 3 +++ pkg/queue/deadletter_rabbitmq_test.go | 19 ++++++++++++++++++ pkg/queue/rabbitmq.go | 23 ++++++++++++++------- pkg/queue/retry_test.go | 29 +++++---------------------- 4 files changed, 43 insertions(+), 31 deletions(-) diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index 50d6820..01652f6 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -29,6 +29,9 @@ func (r *RabbitMQ) PublishDeadLetter(ctx context.Context, payload []byte, metada // dead-letter envelopes. Each source delivery is acknowledged only after the // broker confirms the dead-letter publish. func (r *RabbitMQ) ReadMessagesToDeadletter(reason DeadLetterReason) error { + if r.options.ConsumerQueue == r.options.DeadletterQueue { + return fmt.Errorf("RabbitMQ consumer queue and dead-letter queue must differ for raw transfer") + } if err := r.requireConfirmedDelivery(); err != nil { return err } diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index cfcfbeb..d0eeafc 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -178,6 +178,25 @@ func TestRabbitReadMessagesToDeadletterRequiresConfirmedDelivery(t *testing.T) { } } +func TestRabbitReadMessagesToDeadletterRejectsSameQueue(t *testing.T) { + client, err := NewRabbitMQ(NewRabbitOptions(). + SetConsumerQueue("deadletter"). + SetDeadletterQueue("deadletter"). + SetHost("rabbitmq:5672"). + SetUsername("guest"). + SetPassword("guest"). + SetConfirmedDelivery(true). + Build()) + if err != nil { + t.Fatalf("NewRabbitMQ: %v", err) + } + + err = client.ReadMessagesToDeadletter(DeadLetterReasonHandlerError) + if err == nil || !strings.Contains(err.Error(), "must differ") { + t.Fatalf("error = %v, want same-queue rejection", err) + } +} + func TestRabbitDeadLetterEnvelopePreservesFailureReason(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). diff --git a/pkg/queue/rabbitmq.go b/pkg/queue/rabbitmq.go index 311af0b..5d2fbda 100644 --- a/pkg/queue/rabbitmq.go +++ b/pkg/queue/rabbitmq.go @@ -46,8 +46,9 @@ type RabbitOptions struct { VirtualHost string // MaxRetries caps how many times a PipelineRetry re-queues a message before - // it is parked on the deadletter queue. Zero preserves the historical - // unlimited retry behavior. + // it is parked on the deadletter queue; zero selects defaultMaxRetries. It is + // the hard stop that keeps a permanently failing payload from looping the + // consumer queue forever (see retryOrDeadletter). MaxRetries int // TLS configuration for secure connections (e.g., AWS Amazon MQ) @@ -153,7 +154,9 @@ func (b *RabbitOptionsBuilder) SetPrefetchCount(count int) *RabbitOptionsBuilder } // SetMaxRetries caps how many times a PipelineRetry re-queues a message before -// it is dead-lettered. Zero (the default) preserves unlimited retries. +// it is dead-lettered. Zero (the default) selects defaultMaxRetries. Set it to +// bound how long a transient sink failure is retried before the message is +// parked for inspection instead of being requeued forever. func (b *RabbitOptionsBuilder) SetMaxRetries(maxRetries int) *RabbitOptionsBuilder { b.options.MaxRetries = maxRetries return b @@ -1279,12 +1282,18 @@ func (r *RabbitMQ) publishWithDelayHeaders(queueName string, payload []byte, bac // first delivery (count 0) and set explicitly on every re-queue. const retryCountHeader = "x-retry-count" -// defaultMaxRetries preserves the bounded defaults used by Kafka and SQS. +// defaultMaxRetries bounds PipelineRetry re-queues when RabbitOptions.MaxRetries +// is unset. With the default 5s backoff this rides out a brief sink outage +// (~50s) before the message is dead-lettered instead of looping forever. const defaultMaxRetries = 10 -// maxRetries returns the configured PipelineRetry cap. Zero means unlimited. +// maxRetries returns the configured PipelineRetry cap, or defaultMaxRetries when +// unset. func (r *RabbitMQ) maxRetries() int { - return r.options.MaxRetries + if r.options.MaxRetries > 0 { + return r.options.MaxRetries + } + return defaultMaxRetries } // retryCount reads the x-retry-count header off a delivery, tolerating the @@ -1315,7 +1324,7 @@ func retryCount(headers amqp.Table) int { // after this returns nil, so a failed retry publish cannot silently lose it. func (r *RabbitMQ) retryOrDeadletter(headers amqp.Table, payload []byte, backoff time.Duration, publish func(string, []byte, amqp.Table) error) error { attempts := retryCount(headers) - if maxRetries := r.maxRetries(); maxRetries > 0 && attempts >= maxRetries { + if attempts >= r.maxRetries() { envelope, err := r.deadLetterEnvelope(payload, DeadLetterReasonRetryExhausted, attempts) if err != nil { return err diff --git a/pkg/queue/retry_test.go b/pkg/queue/retry_test.go index ac84ee0..6a85fb2 100644 --- a/pkg/queue/retry_test.go +++ b/pkg/queue/retry_test.go @@ -31,40 +31,21 @@ func TestSetMaxRetriesBuilder(t *testing.T) { if opts := NewRabbitOptions().SetMaxRetries(7).Build(); opts.MaxRetries != 7 { t.Errorf("MaxRetries = %d, want 7", opts.MaxRetries) } - // Unset preserves the historical unlimited retry behavior. + // Unset leaves the zero value so the client falls back to defaultMaxRetries. if def := NewRabbitOptions().Build(); def.MaxRetries != 0 { - t.Errorf("default MaxRetries = %d, want 0", def.MaxRetries) + t.Errorf("default MaxRetries = %d, want 0 (selects defaultMaxRetries at runtime)", def.MaxRetries) } } -func TestMaxRetries(t *testing.T) { - if got := newTestRabbit(t, 0).maxRetries(); got != 0 { - t.Errorf("maxRetries() with unset option = %d, want unlimited (0)", got) +func TestMaxRetriesDefault(t *testing.T) { + if got := newTestRabbit(t, 0).maxRetries(); got != defaultMaxRetries { + t.Errorf("maxRetries() with unset option = %d, want default %d", got, defaultMaxRetries) } if got := newTestRabbit(t, 3).maxRetries(); got != 3 { t.Errorf("maxRetries() = %d, want 3", got) } } -func TestRetryOrDeadletterRetriesWithoutCap(t *testing.T) { - rabbit := newTestRabbit(t, 0) - var queueName string - - err := rabbit.retryOrDeadletter(amqp.Table{retryCountHeader: int32(100)}, []byte("payload"), time.Nanosecond, func(gotQueue string, _ []byte, headers amqp.Table) error { - queueName = gotQueue - if got := retryCount(headers); got != 101 { - t.Fatalf("retry header = %d, want 101", got) - } - return nil - }) - if err != nil { - t.Fatalf("retryOrDeadletter() error = %v", err) - } - if queueName != "c" { - t.Fatalf("queue = %q, want consumer queue", queueName) - } -} - // TestRetryCount covers the header decode that drives the retry cap: a missing // header means a first attempt (0), and the several integer types AMQP may // decode the counter as are all accepted. From 2b7f87a229d92b9b0fc42d766b8628ac8d472cd4 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Thu, 17 Sep 2026 08:16:59 +0000 Subject: [PATCH 4/6] feat(queue): transform DLQ replay batches Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- pkg/queue/deadletter.go | 28 +++++++++++++++ pkg/queue/deadletter_kafka.go | 44 +++++++++++++++++------ pkg/queue/deadletter_rabbitmq.go | 45 +++++++++++++++++++----- pkg/queue/deadletter_rabbitmq_test.go | 8 ++++- pkg/queue/deadletter_sqs.go | 50 ++++++++++++++++++++------- pkg/queue/deadletter_test.go | 30 ++++++++++++++++ pkg/queue/kafka_test.go | 8 ++++- pkg/queue/sqs_test.go | 8 ++++- 8 files changed, 187 insertions(+), 34 deletions(-) diff --git a/pkg/queue/deadletter.go b/pkg/queue/deadletter.go index 8de2293..74bfdb3 100644 --- a/pkg/queue/deadletter.go +++ b/pkg/queue/deadletter.go @@ -77,8 +77,15 @@ type DeadLetterReplayRequest struct { Destination string Execute bool IdleTimeout time.Duration + Transform DeadLetterReplayTransformer } +// DeadLetterReplayTransformer replaces payloads for a planned replay batch. +// Implementations must return one payload for each input message, in the same +// order. Queue providers still choose the trusted replay destination from the +// envelope or request and settle messages only after publishing succeeds. +type DeadLetterReplayTransformer func(context.Context, []DeadLetterMessage) ([][]byte, error) + type DeadLetterReplayResult struct { Scanned int Matched int @@ -129,6 +136,27 @@ func runtimeDeadLetterMetadata(source, deadLetterDestination, routerDestination } } +func transformDeadLetterReplayMessages(ctx context.Context, transform DeadLetterReplayTransformer, messages []DeadLetterMessage) ([][]byte, error) { + payloads := make([][]byte, len(messages)) + if transform == nil { + for index := range messages { + payloads[index] = append([]byte(nil), messages[index].Payload...) + } + return payloads, nil + } + transformed, err := transform(ctx, messages) + if err != nil { + return nil, fmt.Errorf("transform dead-letter replay batch: %w", err) + } + if len(transformed) != len(messages) { + return nil, fmt.Errorf("transform dead-letter replay batch returned %d payloads for %d messages", len(transformed), len(messages)) + } + for index := range transformed { + payloads[index] = append([]byte(nil), transformed[index]...) + } + return payloads, nil +} + func encodeDeadLetterForDestination(payload []byte, metadata DeadLetterMetadata, destination string) ([]byte, error) { if metadata.Source == "" { return nil, fmt.Errorf("dead-letter source is required") diff --git a/pkg/queue/deadletter_kafka.go b/pkg/queue/deadletter_kafka.go index 7382096..c18cb4d 100644 --- a/pkg/queue/deadletter_kafka.go +++ b/pkg/queue/deadletter_kafka.go @@ -82,6 +82,12 @@ func (k *Kafka) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayR idleTimeout = 2 * time.Second } assigned := false + type replayItem struct { + sourceMessage *kafka.Message + message DeadLetterMessage + plan deadLetterReplayPlan + } + items := make([]replayItem, 0, request.Limit) for result.Scanned < request.Limit { message, done, err := readKafkaDeadLetter(ctx, consumer, idleTimeout, &assigned) if err != nil { @@ -98,7 +104,7 @@ func (k *Kafka) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayR if err != nil { return result, err } - if plan.destination == "" || !request.Execute { + if plan.destination == "" { if request.Execute && plan.matched { // Kafka commits are contiguous per partition. Do not process a // later message after retaining this one, because committing the @@ -107,20 +113,38 @@ func (k *Kafka) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayR } continue } + items = append(items, replayItem{sourceMessage: message, message: decoded, plan: plan}) + } + + decodedMessages := make([]DeadLetterMessage, len(items)) + for index := range items { + decodedMessages[index] = items[index].message + } + payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, decodedMessages) + if err != nil { + return result, err + } + if request.Execute && len(items) > 0 { if !k.options.DisableAutoTopicCreation { return result, fmt.Errorf("Kafka replay requires automatic topic creation to be disabled") } _, producer := k.clients() - if err := ensureKafkaTopicExists(producer, plan.destination, k.deliveryTimeout()); err != nil { - return result, fmt.Errorf("validate Kafka replay destination for message %q: %w", decoded.ID, err) - } - if err := k.publishContext(ctx, producer, plan.destination, decoded.Payload, nil); err != nil { - return result, fmt.Errorf("replay Kafka dead-letter message %q: %w", decoded.ID, err) - } - if _, err := consumer.CommitMessage(message); err != nil { - return result, fmt.Errorf("commit replayed Kafka dead-letter message %q: %w", decoded.ID, err) + validatedDestinations := make(map[string]struct{}) + for index, item := range items { + if _, exists := validatedDestinations[item.plan.destination]; !exists { + if err := ensureKafkaTopicExists(producer, item.plan.destination, k.deliveryTimeout()); err != nil { + return result, fmt.Errorf("validate Kafka replay destination for message %q: %w", item.message.ID, err) + } + validatedDestinations[item.plan.destination] = struct{}{} + } + if err := k.publishContext(ctx, producer, item.plan.destination, payloads[index], nil); err != nil { + return result, fmt.Errorf("replay Kafka dead-letter message %q: %w", item.message.ID, err) + } + if _, err := consumer.CommitMessage(item.sourceMessage); err != nil { + return result, fmt.Errorf("commit replayed Kafka dead-letter message %q: %w", item.message.ID, err) + } + result.Replayed++ } - result.Replayed++ } return result, nil } diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index 01652f6..5751553 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -118,6 +118,12 @@ func (r *RabbitMQ) ReplayDeadLetters(ctx context.Context, request DeadLetterRepl requeueRetained := func(operationErr error) error { return errors.Join(operationErr, r.restoreRabbitDeadLetters(retained)) } + type replayItem struct { + delivery amqp.Delivery + message DeadLetterMessage + plan deadLetterReplayPlan + } + items := make([]replayItem, 0, request.Limit) for result.Scanned < request.Limit { if err := ctx.Err(); err != nil { return result, requeueRetained(err) @@ -138,17 +144,31 @@ func (r *RabbitMQ) ReplayDeadLetters(ctx context.Context, request DeadLetterRepl if err != nil { return result, requeueRetained(err) } - if plan.destination == "" || !request.Execute { + if plan.destination == "" { continue } - if err := r.publishDeadLetterConfirmed(ctx, plan.destination, decoded.Payload); err != nil { - return result, requeueRetained(fmt.Errorf("replay RabbitMQ dead-letter message %q: %w", decoded.ID, err)) - } - if err := delivery.Ack(false); err != nil { - return result, requeueRetained(fmt.Errorf("settle replayed RabbitMQ dead-letter message %q: %w", decoded.ID, err)) + items = append(items, replayItem{delivery: delivery, message: decoded, plan: plan}) + } + + messages := make([]DeadLetterMessage, len(items)) + for index := range items { + messages[index] = items[index].message + } + payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, messages) + if err != nil { + return result, requeueRetained(err) + } + if request.Execute { + for index, item := range items { + if err := r.publishDeadLetterConfirmed(ctx, item.plan.destination, payloads[index]); err != nil { + return result, requeueRetained(fmt.Errorf("replay RabbitMQ dead-letter message %q: %w", item.message.ID, err)) + } + if err := item.delivery.Ack(false); err != nil { + return result, requeueRetained(fmt.Errorf("settle replayed RabbitMQ dead-letter message %q: %w", item.message.ID, err)) + } + retained = removeRabbitDeadLetter(retained, item.delivery.DeliveryTag) + result.Replayed++ } - retained = retained[:len(retained)-1] - result.Replayed++ } if err := r.restoreRabbitDeadLetters(retained); err != nil { return result, err @@ -156,6 +176,15 @@ func (r *RabbitMQ) ReplayDeadLetters(ctx context.Context, request DeadLetterRepl return result, nil } +func removeRabbitDeadLetter(deliveries []amqp.Delivery, deliveryTag uint64) []amqp.Delivery { + for index := range deliveries { + if deliveries[index].DeliveryTag == deliveryTag { + return append(deliveries[:index], deliveries[index+1:]...) + } + } + return deliveries +} + func (r *RabbitMQ) getDeadLetter() (amqp.Delivery, bool, error) { if r.deadLetterGet != nil { return r.deadLetterGet() diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index d0eeafc..1965741 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -65,7 +65,7 @@ func TestRabbitDeadLetterReplayPublishesBeforeAck(t *testing.T) { } published := false client.deadLetterReplayPublish = func(_ context.Context, destination string, payload []byte) error { - if destination != "events" || !bytes.Equal(payload, []byte("payload")) { + if destination != "events" || !bytes.Equal(payload, []byte("transformed")) { t.Fatalf("publish destination=%q payload=%q", destination, payload) } if len(acknowledger.acked) != 0 { @@ -78,6 +78,12 @@ func TestRabbitDeadLetterReplayPublishesBeforeAck(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + if len(messages) != 1 || string(messages[0].Payload) != "payload" { + t.Fatalf("transform messages = %+v", messages) + } + return [][]byte{[]byte("transformed")}, nil + }, }) if err != nil { t.Fatalf("ReplayDeadLetters: %v", err) diff --git a/pkg/queue/deadletter_sqs.go b/pkg/queue/deadletter_sqs.go index af00839..cc4e3b3 100644 --- a/pkg/queue/deadletter_sqs.go +++ b/pkg/queue/deadletter_sqs.go @@ -78,6 +78,13 @@ func (s *SQS) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayReq releaseRetained := func(operationErr error) error { return errors.Join(operationErr, s.releaseSQSDeadLetters(deadLetterQueueURL, retained)) } + type replayItem struct { + sourceMessage types.Message + message DeadLetterMessage + plan deadLetterReplayPlan + destinationURL string + } + items := make([]replayItem, 0, request.Limit) for result.Scanned < request.Limit { messages, receiveErr := s.receiveDeadLetterBatch(ctx, deadLetterQueueURL, request.Limit-result.Scanned) if receiveErr != nil { @@ -97,24 +104,41 @@ func (s *SQS) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayReq if planErr != nil { return result, releaseRetained(planErr) } - if plan.destination == "" || !request.Execute { + if plan.destination == "" { continue } - destinationURL, resolveErr := s.queueURL(ctx, plan.destination) - if resolveErr != nil { - return result, releaseRetained(resolveErr) - } - if destinationURL == deadLetterQueueURL { - return result, releaseRetained(fmt.Errorf("refusing to replay SQS dead-letter message %q back to %q", decoded.ID, s.options.DeadletterQueue)) - } - if err := s.publishReplay(ctx, plan.destination, destinationURL, decoded.Payload); err != nil { - return result, releaseRetained(fmt.Errorf("replay SQS dead-letter message %q: %w", decoded.ID, err)) + item := replayItem{sourceMessage: message, message: decoded, plan: plan} + if request.Execute { + destinationURL, resolveErr := s.queueURL(ctx, plan.destination) + if resolveErr != nil { + return result, releaseRetained(resolveErr) + } + if destinationURL == deadLetterQueueURL { + return result, releaseRetained(fmt.Errorf("refusing to replay SQS dead-letter message %q back to %q", decoded.ID, s.options.DeadletterQueue)) + } + item.destinationURL = destinationURL } + items = append(items, item) + } + } - if err := s.deleteSQSDeadLetter(ctx, deadLetterQueueURL, message); err != nil { - return result, releaseRetained(fmt.Errorf("settle replayed SQS dead-letter message %q: %w", decoded.ID, err)) + decodedMessages := make([]DeadLetterMessage, len(items)) + for index := range items { + decodedMessages[index] = items[index].message + } + payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, decodedMessages) + if err != nil { + return result, releaseRetained(err) + } + if request.Execute { + for index, item := range items { + if err := s.publishReplay(ctx, item.plan.destination, item.destinationURL, payloads[index]); err != nil { + return result, releaseRetained(fmt.Errorf("replay SQS dead-letter message %q: %w", item.message.ID, err)) + } + if err := s.deleteSQSDeadLetter(ctx, deadLetterQueueURL, item.sourceMessage); err != nil { + return result, releaseRetained(fmt.Errorf("settle replayed SQS dead-letter message %q: %w", item.message.ID, err)) } - retained = removeSQSDeadLetter(retained, aws.ToString(message.ReceiptHandle)) + retained = removeSQSDeadLetter(retained, aws.ToString(item.sourceMessage.ReceiptHandle)) result.Replayed++ } } diff --git a/pkg/queue/deadletter_test.go b/pkg/queue/deadletter_test.go index 81805a3..cb9b991 100644 --- a/pkg/queue/deadletter_test.go +++ b/pkg/queue/deadletter_test.go @@ -2,6 +2,7 @@ package queue import ( "bytes" + "context" "testing" "time" ) @@ -164,3 +165,32 @@ func TestRuntimeDeadLetterMetadataFallsBackToSourceWithoutRouter(t *testing.T) { t.Fatalf("replay destination = %q, want sequence", metadata.ReplayDestination) } } + +func TestTransformDeadLetterReplayMessages(t *testing.T) { + messages := []DeadLetterMessage{ + {ID: "one", Payload: []byte("first")}, + {ID: "two", Payload: []byte("second")}, + } + payloads, err := transformDeadLetterReplayMessages(context.Background(), func(_ context.Context, got []DeadLetterMessage) ([][]byte, error) { + if len(got) != 2 || got[0].ID != "one" || got[1].ID != "two" { + t.Fatalf("messages = %+v", got) + } + return [][]byte{[]byte("transformed-first"), []byte("transformed-second")}, nil + }, messages) + if err != nil { + t.Fatal(err) + } + if !bytes.Equal(payloads[0], []byte("transformed-first")) || + !bytes.Equal(payloads[1], []byte("transformed-second")) { + t.Fatalf("payloads = %q", payloads) + } +} + +func TestTransformDeadLetterReplayMessagesRequiresMatchingResultCount(t *testing.T) { + _, err := transformDeadLetterReplayMessages(context.Background(), func(context.Context, []DeadLetterMessage) ([][]byte, error) { + return nil, nil + }, []DeadLetterMessage{{ID: "one"}}) + if err == nil { + t.Fatal("expected transform result count error") + } +} diff --git a/pkg/queue/kafka_test.go b/pkg/queue/kafka_test.go index dca55e6..a415417 100644 --- a/pkg/queue/kafka_test.go +++ b/pkg/queue/kafka_test.go @@ -161,6 +161,12 @@ func TestKafkaDeadLetterReplayPublishesBeforeCommit(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + if len(messages) != 1 || string(messages[0].Payload) != "payload" { + t.Fatalf("transform messages = %+v", messages) + } + return [][]byte{[]byte("transformed")}, nil + }, }) if err != nil { t.Fatalf("ReplayDeadLetters: %v", err) @@ -168,7 +174,7 @@ func TestKafkaDeadLetterReplayPublishesBeforeCommit(t *testing.T) { if result.Replayed != 1 || consumer.commitCount != 1 || len(producer.messages) != 1 { t.Fatalf("result=%+v commits=%d messages=%+v", result, consumer.commitCount, producer.messages) } - if *producer.messages[0].TopicPartition.Topic != "events" || string(producer.messages[0].Value) != "payload" { + if *producer.messages[0].TopicPartition.Topic != "events" || string(producer.messages[0].Value) != "transformed" { t.Fatalf("replayed message = %+v", producer.messages[0]) } } diff --git a/pkg/queue/sqs_test.go b/pkg/queue/sqs_test.go index 333458b..13b3c2c 100644 --- a/pkg/queue/sqs_test.go +++ b/pkg/queue/sqs_test.go @@ -109,6 +109,12 @@ func TestSQSDeadLetterReplaySendsBeforeDelete(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + if len(messages) != 1 || string(messages[0].Payload) != "payload" { + t.Fatalf("transform messages = %+v", messages) + } + return [][]byte{[]byte("transformed")}, nil + }, }) if err != nil { t.Fatalf("ReplayDeadLetters: %v", err) @@ -116,7 +122,7 @@ func TestSQSDeadLetterReplaySendsBeforeDelete(t *testing.T) { if result.Replayed != 1 || len(fake.operations) != 2 || fake.operations[0] != "send" || fake.operations[1] != "delete" { t.Fatalf("result=%+v operations=%v", result, fake.operations) } - if aws.ToString(fake.sent[0].QueueUrl) != fake.queueURLs["events"] || aws.ToString(fake.sent[0].MessageBody) != "payload" { + if aws.ToString(fake.sent[0].QueueUrl) != fake.queueURLs["events"] || aws.ToString(fake.sent[0].MessageBody) != "transformed" { t.Fatalf("replay send = %+v", fake.sent[0]) } } From d0e77f973f1ab742a10b3748201e3ca13b05df83 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Thu, 17 Sep 2026 08:33:29 +0000 Subject: [PATCH 5/6] fix(queue): bound failed replay cleanup Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- README.md | 6 +++ pkg/queue/deadletter_rabbitmq.go | 15 +++++-- pkg/queue/deadletter_rabbitmq_test.go | 59 +++++++++++++++++++++++++++ pkg/queue/deadletter_sqs.go | 6 +-- pkg/queue/kafka_test.go | 33 +++++++++++++++ pkg/queue/sqs_test.go | 31 ++++++++++++++ 6 files changed, 143 insertions(+), 7 deletions(-) diff --git a/README.md b/README.md index cbfe99a..6e8d97b 100644 --- a/README.md +++ b/README.md @@ -121,6 +121,12 @@ in the envelope, and finally the source queue for envelopes written before their configured router, which resolves the next queue from the event's remaining stages, while non-pipeline workers preserve source-queue behavior. +`DeadLetterReplayRequest.Transform` can replace payloads for an entire planned +batch before publication. The callback receives decoded messages in replay +order and must return the same number of payloads. Providers retain the batch +when transformation fails and continue to publish each transformed payload +before settling its original dead-letter message. + ## Inspecting and Replaying Dead-Letter Messages The operational commands are provided by the separate diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index 5751553..a73a9be 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -257,10 +257,19 @@ func (r *RabbitMQ) publishDeadLetterConfirmed(ctx context.Context, destination s func (r *RabbitMQ) restoreRabbitDeadLetters(deliveries []amqp.Delivery) error { var result error - for _, delivery := range deliveries { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + for index, delivery := range deliveries { + if err := ctx.Err(); err != nil { + result = errors.Join(result, fmt.Errorf("restore retained RabbitMQ dead-letter messages: %w", err)) + for _, remaining := range deliveries[index:] { + if nackErr := remaining.Nack(false, true); nackErr != nil { + result = errors.Join(result, fmt.Errorf("requeue retained RabbitMQ dead-letter message: %w", nackErr)) + } + } + break + } err := r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, delivery.Body) - cancel() if err != nil { nackErr := delivery.Nack(false, true) result = errors.Join(result, fmt.Errorf("republish retained RabbitMQ dead-letter message: %w", err)) diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index 1965741..cde7281 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -3,6 +3,7 @@ package queue import ( "bytes" "context" + "errors" "strings" "testing" @@ -93,6 +94,64 @@ func TestRabbitDeadLetterReplayPublishesBeforeAck(t *testing.T) { } } +func TestRabbitDeadLetterReplayTransformFailureRestoresBatch(t *testing.T) { + client, err := NewRabbitMQ(NewRabbitOptions(). + SetConsumerQueue("events"). + SetDeadletterQueue("deadletter"). + SetHost("rabbitmq:5672"). + SetUsername("guest"). + SetPassword("guest"). + Build()) + if err != nil { + t.Fatalf("NewRabbitMQ: %v", err) + } + envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if err != nil { + t.Fatal(err) + } + acknowledger := &fakeRabbitAcknowledger{} + deliveries := []amqp.Delivery{{ + Acknowledger: acknowledger, + DeliveryTag: 1, + Body: envelope, + }} + client.deadLetterGet = func() (amqp.Delivery, bool, error) { + if len(deliveries) == 0 { + return amqp.Delivery{}, false, nil + } + delivery := deliveries[0] + deliveries = deliveries[1:] + return delivery, true, nil + } + eventPublishes := 0 + restored := 0 + client.deadLetterReplayPublish = func(_ context.Context, destination string, _ []byte) error { + if destination == "events" { + eventPublishes++ + } else if destination == "deadletter" { + restored++ + } + return nil + } + + _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 1, + Execute: true, + Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + return nil, errors.New("refresh failed") + }, + }) + if err == nil || !strings.Contains(err.Error(), "refresh failed") { + t.Fatalf("error = %v", err) + } + if eventPublishes != 0 || restored != 1 || len(acknowledger.acked) != 1 { + t.Fatalf("event publishes=%d restored=%d acknowledgements=%v", eventPublishes, restored, acknowledger.acked) + } +} + func TestRabbitPublishToDeadLetterQueueAddsEnvelope(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). diff --git a/pkg/queue/deadletter_sqs.go b/pkg/queue/deadletter_sqs.go index cc4e3b3..0468989 100644 --- a/pkg/queue/deadletter_sqs.go +++ b/pkg/queue/deadletter_sqs.go @@ -195,6 +195,8 @@ func (s *SQS) releaseSQSDeadLetters(queueURL string, messages []types.Message) e return nil } client, _ := s.clientAndConsumerURL() + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() if batchClient, ok := client.(sqsVisibilityBatchClient); ok { for start := 0; start < len(messages); start += int(defaultSQSMaxNumberOfMessages) { end := start + int(defaultSQSMaxNumberOfMessages) @@ -212,12 +214,10 @@ func (s *SQS) releaseSQSDeadLetters(queueURL string, messages []types.Message) e VisibilityTimeout: 0, }) } - ctx, cancel := context.WithTimeout(context.Background(), sqsOperationTimeout) output, err := batchClient.ChangeMessageVisibilityBatch(ctx, &awssqs.ChangeMessageVisibilityBatchInput{ QueueUrl: aws.String(queueURL), Entries: entries, }) - cancel() if err != nil { return fmt.Errorf("release SQS dead-letter message batch: %w", err) } @@ -234,13 +234,11 @@ func (s *SQS) releaseSQSDeadLetters(queueURL string, messages []types.Message) e if message.ReceiptHandle == nil || *message.ReceiptHandle == "" { return fmt.Errorf("SQS dead-letter message has no receipt handle") } - ctx, cancel := context.WithTimeout(context.Background(), sqsOperationTimeout) _, err := client.ChangeMessageVisibility(ctx, &awssqs.ChangeMessageVisibilityInput{ QueueUrl: aws.String(queueURL), ReceiptHandle: message.ReceiptHandle, VisibilityTimeout: 0, }) - cancel() if err != nil { return fmt.Errorf("release SQS dead-letter message: %w", err) } diff --git a/pkg/queue/kafka_test.go b/pkg/queue/kafka_test.go index a415417..b16fcbd 100644 --- a/pkg/queue/kafka_test.go +++ b/pkg/queue/kafka_test.go @@ -179,6 +179,39 @@ func TestKafkaDeadLetterReplayPublishesBeforeCommit(t *testing.T) { } } +func TestKafkaDeadLetterReplayTransformFailureDoesNotPublishOrCommit(t *testing.T) { + envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if err != nil { + t.Fatal(err) + } + client, consumer, producer := newTestKafka(t, &kafka.Message{ + TopicPartition: kafka.TopicPartition{ + Topic: stringPointer("deadletter"), + Partition: 1, + Offset: 42, + }, + Value: envelope, + }) + client.options.DisableAutoTopicCreation = true + + _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 1, + Execute: true, + Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + return nil, errors.New("refresh failed") + }, + }) + if err == nil || !strings.Contains(err.Error(), "refresh failed") { + t.Fatalf("error = %v", err) + } + if len(producer.messages) != 0 || consumer.commitCount != 0 { + t.Fatalf("messages=%d commits=%d", len(producer.messages), consumer.commitCount) + } +} + func TestKafkaDeadLetterReplayRejectsMissingDestinationBeforePublishOrCommit(t *testing.T) { envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ Source: "missing-events", diff --git a/pkg/queue/sqs_test.go b/pkg/queue/sqs_test.go index 13b3c2c..4a46c97 100644 --- a/pkg/queue/sqs_test.go +++ b/pkg/queue/sqs_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "strings" "testing" "github.com/aws/aws-sdk-go-v2/aws" @@ -127,6 +128,36 @@ func TestSQSDeadLetterReplaySendsBeforeDelete(t *testing.T) { } } +func TestSQSDeadLetterReplayTransformFailureReleasesBatch(t *testing.T) { + envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if err != nil { + t.Fatal(err) + } + client, fake := newTestSQS(t, types.Message{ + MessageId: aws.String("message-1"), + Body: aws.String(string(envelope)), + ReceiptHandle: aws.String("receipt"), + }) + fake.visibilityError = nil + + _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 1, + Execute: true, + Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + return nil, errors.New("refresh failed") + }, + }) + if err == nil || !strings.Contains(err.Error(), "refresh failed") { + t.Fatalf("error = %v", err) + } + if len(fake.sent) != 0 || len(fake.deleted) != 0 || len(fake.visibilityChanges) != 1 { + t.Fatalf("sent=%d deleted=%d visibility=%d", len(fake.sent), len(fake.deleted), len(fake.visibilityChanges)) + } +} + func TestSQSDeadLetterEnvelopeRecordsRouter(t *testing.T) { client, _ := newTestSQS(t, types.Message{}) envelope, err := client.deadLetterEnvelope([]byte("payload"), DeadLetterReasonHandlerError, 0) From 4a4583bbacc9a839644c3fb46363478c66ea8848 Mon Sep 17 00:00:00 2001 From: Kilian Boute Date: Thu, 17 Sep 2026 09:16:57 +0000 Subject: [PATCH 6/6] feat(queue): batch transformed DLQ replay Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- README.md | 23 +++- pkg/queue/deadletter.go | 112 ++++++++++++--- pkg/queue/deadletter_kafka.go | 117 +++++++++++----- pkg/queue/deadletter_rabbitmq.go | 94 ++++++++----- pkg/queue/deadletter_rabbitmq_test.go | 92 ++++++++++++- pkg/queue/deadletter_sqs.go | 190 +++++++++++++++++++------- pkg/queue/deadletter_test.go | 39 +++++- pkg/queue/kafka_test.go | 101 +++++++++++++- pkg/queue/sqs_test.go | 134 +++++++++++++++++- 9 files changed, 746 insertions(+), 156 deletions(-) diff --git a/README.md b/README.md index 6e8d97b..881557f 100644 --- a/README.md +++ b/README.md @@ -121,11 +121,24 @@ in the envelope, and finally the source queue for envelopes written before their configured router, which resolves the next queue from the event's remaining stages, while non-pipeline workers preserve source-queue behavior. -`DeadLetterReplayRequest.Transform` can replace payloads for an entire planned -batch before publication. The callback receives decoded messages in replay -order and must return the same number of payloads. Providers retain the batch -when transformation fails and continue to publish each transformed payload -before settling its original dead-letter message. +`DeadLetterReplayRequest.Transform` can replace or skip payloads before +publication. Set `BatchSize`, `BatchTimeout`, and `BatchDelay` to process a +larger bounded `Limit` in chunks within one provider operation. The callback +receives decoded messages in replay order and must return one transformation +for each message. A skipped message remains dead-lettered, while valid messages +in the same batch continue where the provider can settle them safely. Providers +retain the current batch when transformation fails and always publish a +transformed payload before settling its original dead-letter message. + +RabbitMQ keeps retained deliveries unacknowledged and SQS keeps them invisible +until the bounded operation finishes, preventing a retained message from being +scanned repeatedly across batches. Kafka offsets are contiguous per partition: +after a skipped message, later messages in that partition remain uncommitted, +while replay can continue on other partitions. Because retained RabbitMQ +deliveries and SQS receipt handles are held for the operation, callers should +treat `Limit` as both a work and memory safety bound. Batched SQS replay requires +either `BatchTimeout` or an overall context deadline so its visibility window +can safely cover the run. ## Inspecting and Replaying Dead-Letter Messages diff --git a/pkg/queue/deadletter.go b/pkg/queue/deadletter.go index 74bfdb3..81a61df 100644 --- a/pkg/queue/deadletter.go +++ b/pkg/queue/deadletter.go @@ -72,19 +72,26 @@ type DeadLetterInspectResult struct { } type DeadLetterReplayRequest struct { - Limit int - Source string - Destination string - Execute bool - IdleTimeout time.Duration - Transform DeadLetterReplayTransformer + Limit int + BatchSize int + BatchDelay time.Duration + BatchTimeout time.Duration + Source string + Destination string + Execute bool + IdleTimeout time.Duration + Transform DeadLetterReplayTransformer +} + +type DeadLetterReplayTransformation struct { + Payload []byte + Skip bool } -// DeadLetterReplayTransformer replaces payloads for a planned replay batch. -// Implementations must return one payload for each input message, in the same -// order. Queue providers still choose the trusted replay destination from the -// envelope or request and settle messages only after publishing succeeds. -type DeadLetterReplayTransformer func(context.Context, []DeadLetterMessage) ([][]byte, error) +// DeadLetterReplayTransformer replaces or skips payloads for a planned replay +// batch. Implementations must return one result for each input message, in the +// same order. A skipped message remains on the dead-letter destination. +type DeadLetterReplayTransformer func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) type DeadLetterReplayResult struct { Scanned int @@ -94,6 +101,7 @@ type DeadLetterReplayResult struct { Retained int Legacy int Unroutable int + Skipped int Destinations map[string]int } @@ -136,13 +144,16 @@ func runtimeDeadLetterMetadata(source, deadLetterDestination, routerDestination } } -func transformDeadLetterReplayMessages(ctx context.Context, transform DeadLetterReplayTransformer, messages []DeadLetterMessage) ([][]byte, error) { - payloads := make([][]byte, len(messages)) +func transformDeadLetterReplayMessages(ctx context.Context, transform DeadLetterReplayTransformer, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { + transformations := make([]DeadLetterReplayTransformation, len(messages)) + if len(messages) == 0 { + return transformations, nil + } if transform == nil { for index := range messages { - payloads[index] = append([]byte(nil), messages[index].Payload...) + transformations[index].Payload = append([]byte(nil), messages[index].Payload...) } - return payloads, nil + return transformations, nil } transformed, err := transform(ctx, messages) if err != nil { @@ -152,9 +163,12 @@ func transformDeadLetterReplayMessages(ctx context.Context, transform DeadLetter return nil, fmt.Errorf("transform dead-letter replay batch returned %d payloads for %d messages", len(transformed), len(messages)) } for index := range transformed { - payloads[index] = append([]byte(nil), transformed[index]...) + transformations[index] = DeadLetterReplayTransformation{ + Payload: append([]byte(nil), transformed[index].Payload...), + Skip: transformed[index].Skip, + } } - return payloads, nil + return transformations, nil } func encodeDeadLetterForDestination(payload []byte, metadata DeadLetterMetadata, destination string) ([]byte, error) { @@ -234,6 +248,53 @@ func validateDeadLetterLimit(limit int) error { return nil } +func validateDeadLetterReplayRequest(request DeadLetterReplayRequest) error { + if request.BatchDelay < 0 { + return fmt.Errorf("dead-letter batch delay cannot be negative") + } + if request.BatchTimeout < 0 { + return fmt.Errorf("dead-letter batch timeout cannot be negative") + } + if request.BatchSize == 0 { + return validateDeadLetterLimit(request.Limit) + } + if request.Limit < 1 || request.Limit > 1000000 { + return fmt.Errorf("dead-letter message limit must be between 1 and 1000000 for batched replay") + } + if request.BatchSize < 1 || request.BatchSize > 10000 { + return fmt.Errorf("dead-letter batch size must be between 1 and 10000") + } + return nil +} + +func deadLetterReplayBatchSize(request DeadLetterReplayRequest) int { + if request.BatchSize > 0 { + return request.BatchSize + } + return request.Limit +} + +func waitForDeadLetterReplayBatch(ctx context.Context, delay time.Duration) error { + if delay <= 0 { + return nil + } + timer := time.NewTimer(delay) + defer timer.Stop() + select { + case <-ctx.Done(): + return ctx.Err() + case <-timer.C: + return nil + } +} + +func deadLetterReplayBatchContext(ctx context.Context, timeout time.Duration) (context.Context, context.CancelFunc) { + if timeout <= 0 { + return context.WithCancel(ctx) + } + return context.WithTimeout(ctx, timeout) +} + func addDeadLetterInspection(result *DeadLetterInspectResult, message DeadLetterMessage, sourceFilter string) { result.Scanned++ if message.Legacy { @@ -311,3 +372,20 @@ func deadLetterSource(message DeadLetterMessage) string { } return message.DeadLetter.Source } + +func skipDeadLetterReplay(result *DeadLetterReplayResult, destination string, execute bool) { + result.Skipped++ + retainPlannedDeadLetterReplay(result, destination, execute) +} + +func retainPlannedDeadLetterReplay(result *DeadLetterReplayResult, destination string, execute bool) { + result.Planned-- + if count := result.Destinations[destination]; count <= 1 { + delete(result.Destinations, destination) + } else { + result.Destinations[destination] = count - 1 + } + if execute { + result.Retained++ + } +} diff --git a/pkg/queue/deadletter_kafka.go b/pkg/queue/deadletter_kafka.go index c18cb4d..5a9495a 100644 --- a/pkg/queue/deadletter_kafka.go +++ b/pkg/queue/deadletter_kafka.go @@ -63,7 +63,7 @@ func (k *Kafka) InspectDeadLetters(ctx context.Context, request DeadLetterInspec func (k *Kafka) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayRequest) (DeadLetterReplayResult, error) { var result DeadLetterReplayResult - if err := validateDeadLetterLimit(request.Limit); err != nil { + if err := validateDeadLetterReplayRequest(request); err != nil { return result, err } if request.Execute && request.Source != "" { @@ -87,64 +87,106 @@ func (k *Kafka) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayR message DeadLetterMessage plan deadLetterReplayPlan } - items := make([]replayItem, 0, request.Limit) + batchSize := deadLetterReplayBatchSize(request) + blockedPartitions := make(map[int32]struct{}) + validatedDestinations := make(map[string]struct{}) for result.Scanned < request.Limit { - message, done, err := readKafkaDeadLetter(ctx, consumer, idleTimeout, &assigned) - if err != nil { - return result, err - } - if done { - break + batchCtx, cancelBatch := deadLetterReplayBatchContext(ctx, request.BatchTimeout) + items := make([]replayItem, 0, batchSize) + batchScanned := 0 + exhausted := false + for batchScanned < batchSize && result.Scanned < request.Limit { + message, done, err := readKafkaDeadLetter(batchCtx, consumer, idleTimeout, &assigned) + if err != nil { + cancelBatch() + return result, err + } + if done { + exhausted = true + break + } + batchScanned++ + decoded, err := decodeDeadLetter(kafkaDeadLetterID(message), message.Value) + if err != nil { + cancelBatch() + return result, err + } + plan, err := planDeadLetterReplay(&result, decoded, request, k.options.DeadletterTopic) + if err != nil { + cancelBatch() + return result, err + } + if plan.destination == "" { + if request.Execute && plan.matched { + blockedPartitions[message.TopicPartition.Partition] = struct{}{} + } + continue + } + if request.Execute { + if _, blocked := blockedPartitions[message.TopicPartition.Partition]; blocked { + retainPlannedDeadLetterReplay(&result, plan.destination, true) + continue + } + } + items = append(items, replayItem{sourceMessage: message, message: decoded, plan: plan}) } - decoded, err := decodeDeadLetter(kafkaDeadLetterID(message), message.Value) - if err != nil { - return result, err + + decodedMessages := make([]DeadLetterMessage, len(items)) + for index := range items { + decodedMessages[index] = items[index].message } - plan, err := planDeadLetterReplay(&result, decoded, request, k.options.DeadletterTopic) + transformations, err := transformDeadLetterReplayMessages(batchCtx, request.Transform, decodedMessages) if err != nil { + cancelBatch() return result, err } - if plan.destination == "" { - if request.Execute && plan.matched { - // Kafka commits are contiguous per partition. Do not process a - // later message after retaining this one, because committing the - // later offset could make the retained message unreplayable. - break - } - continue - } - items = append(items, replayItem{sourceMessage: message, message: decoded, plan: plan}) - } - - decodedMessages := make([]DeadLetterMessage, len(items)) - for index := range items { - decodedMessages[index] = items[index].message - } - payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, decodedMessages) - if err != nil { - return result, err - } - if request.Execute && len(items) > 0 { - if !k.options.DisableAutoTopicCreation { + if request.Execute && len(items) > 0 && !k.options.DisableAutoTopicCreation { + cancelBatch() return result, fmt.Errorf("Kafka replay requires automatic topic creation to be disabled") } _, producer := k.clients() - validatedDestinations := make(map[string]struct{}) for index, item := range items { + partition := item.sourceMessage.TopicPartition.Partition + if transformations[index].Skip { + skipDeadLetterReplay(&result, item.plan.destination, request.Execute) + if request.Execute { + blockedPartitions[partition] = struct{}{} + } + continue + } + if !request.Execute { + continue + } + if _, blocked := blockedPartitions[partition]; blocked { + retainPlannedDeadLetterReplay(&result, item.plan.destination, true) + continue + } if _, exists := validatedDestinations[item.plan.destination]; !exists { if err := ensureKafkaTopicExists(producer, item.plan.destination, k.deliveryTimeout()); err != nil { + cancelBatch() return result, fmt.Errorf("validate Kafka replay destination for message %q: %w", item.message.ID, err) } validatedDestinations[item.plan.destination] = struct{}{} } - if err := k.publishContext(ctx, producer, item.plan.destination, payloads[index], nil); err != nil { + if err := k.publishContext(batchCtx, producer, item.plan.destination, transformations[index].Payload, nil); err != nil { + cancelBatch() return result, fmt.Errorf("replay Kafka dead-letter message %q: %w", item.message.ID, err) } if _, err := consumer.CommitMessage(item.sourceMessage); err != nil { + cancelBatch() return result, fmt.Errorf("commit replayed Kafka dead-letter message %q: %w", item.message.ID, err) } result.Replayed++ } + cancelBatch() + if exhausted || result.Scanned >= request.Limit { + break + } + if request.Execute { + if err := waitForDeadLetterReplayBatch(ctx, request.BatchDelay); err != nil { + return result, err + } + } } return result, nil } @@ -177,6 +219,9 @@ func readKafkaDeadLetter(ctx context.Context, consumer KafkaConsumer, idleTimeou if !errors.As(err, &kafkaError) || kafkaError.Code() != kafka.ErrTimedOut { return nil, false, err } + if err := ctx.Err(); err != nil { + return nil, false, err + } if *assigned { return nil, true, nil } diff --git a/pkg/queue/deadletter_rabbitmq.go b/pkg/queue/deadletter_rabbitmq.go index a73a9be..a9e3e33 100644 --- a/pkg/queue/deadletter_rabbitmq.go +++ b/pkg/queue/deadletter_rabbitmq.go @@ -105,7 +105,7 @@ func (r *RabbitMQ) InspectDeadLetters(ctx context.Context, request DeadLetterIns func (r *RabbitMQ) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayRequest) (DeadLetterReplayResult, error) { var result DeadLetterReplayResult - if err := validateDeadLetterLimit(request.Limit); err != nil { + if err := validateDeadLetterReplayRequest(request); err != nil { return result, err } if r.deadLetterGet == nil || r.deadLetterReplayPublish == nil { @@ -123,52 +123,80 @@ func (r *RabbitMQ) ReplayDeadLetters(ctx context.Context, request DeadLetterRepl message DeadLetterMessage plan deadLetterReplayPlan } - items := make([]replayItem, 0, request.Limit) + batchSize := deadLetterReplayBatchSize(request) for result.Scanned < request.Limit { - if err := ctx.Err(); err != nil { - return result, requeueRetained(err) - } - delivery, ok, err := r.getDeadLetter() - if err != nil { - return result, requeueRetained(err) - } - if !ok { - break + batchCtx, cancelBatch := deadLetterReplayBatchContext(ctx, request.BatchTimeout) + items := make([]replayItem, 0, batchSize) + batchScanned := 0 + exhausted := false + for batchScanned < batchSize && result.Scanned < request.Limit { + if err := batchCtx.Err(); err != nil { + cancelBatch() + return result, requeueRetained(err) + } + delivery, ok, err := r.getDeadLetter() + if err != nil { + cancelBatch() + return result, requeueRetained(err) + } + if !ok { + exhausted = true + break + } + batchScanned++ + retained = append(retained, delivery) + decoded, err := decodeDeadLetter(rabbitDeadLetterID(delivery), delivery.Body) + if err != nil { + cancelBatch() + return result, requeueRetained(err) + } + plan, err := planDeadLetterReplay(&result, decoded, request, r.options.DeadletterQueue) + if err != nil { + cancelBatch() + return result, requeueRetained(err) + } + if plan.destination != "" { + items = append(items, replayItem{delivery: delivery, message: decoded, plan: plan}) + } } - retained = append(retained, delivery) - decoded, err := decodeDeadLetter(rabbitDeadLetterID(delivery), delivery.Body) - if err != nil { - return result, requeueRetained(err) + + messages := make([]DeadLetterMessage, len(items)) + for index := range items { + messages[index] = items[index].message } - plan, err := planDeadLetterReplay(&result, decoded, request, r.options.DeadletterQueue) + transformations, err := transformDeadLetterReplayMessages(batchCtx, request.Transform, messages) if err != nil { + cancelBatch() return result, requeueRetained(err) } - if plan.destination == "" { - continue - } - items = append(items, replayItem{delivery: delivery, message: decoded, plan: plan}) - } - - messages := make([]DeadLetterMessage, len(items)) - for index := range items { - messages[index] = items[index].message - } - payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, messages) - if err != nil { - return result, requeueRetained(err) - } - if request.Execute { for index, item := range items { - if err := r.publishDeadLetterConfirmed(ctx, item.plan.destination, payloads[index]); err != nil { + if transformations[index].Skip { + skipDeadLetterReplay(&result, item.plan.destination, request.Execute) + continue + } + if !request.Execute { + continue + } + if err := r.publishDeadLetterConfirmed(batchCtx, item.plan.destination, transformations[index].Payload); err != nil { + cancelBatch() return result, requeueRetained(fmt.Errorf("replay RabbitMQ dead-letter message %q: %w", item.message.ID, err)) } if err := item.delivery.Ack(false); err != nil { + cancelBatch() return result, requeueRetained(fmt.Errorf("settle replayed RabbitMQ dead-letter message %q: %w", item.message.ID, err)) } retained = removeRabbitDeadLetter(retained, item.delivery.DeliveryTag) result.Replayed++ } + cancelBatch() + if exhausted || result.Scanned >= request.Limit { + break + } + if request.Execute { + if err := waitForDeadLetterReplayBatch(ctx, request.BatchDelay); err != nil { + return result, requeueRetained(err) + } + } } if err := r.restoreRabbitDeadLetters(retained); err != nil { return result, err @@ -271,8 +299,8 @@ func (r *RabbitMQ) restoreRabbitDeadLetters(deliveries []amqp.Delivery) error { } err := r.publishDeadLetterConfirmed(ctx, r.options.DeadletterQueue, delivery.Body) if err != nil { - nackErr := delivery.Nack(false, true) result = errors.Join(result, fmt.Errorf("republish retained RabbitMQ dead-letter message: %w", err)) + nackErr := delivery.Nack(false, true) if nackErr != nil { result = errors.Join(result, fmt.Errorf("requeue retained RabbitMQ dead-letter message: %w", nackErr)) } diff --git a/pkg/queue/deadletter_rabbitmq_test.go b/pkg/queue/deadletter_rabbitmq_test.go index cde7281..5962cad 100644 --- a/pkg/queue/deadletter_rabbitmq_test.go +++ b/pkg/queue/deadletter_rabbitmq_test.go @@ -4,6 +4,7 @@ import ( "bytes" "context" "errors" + "slices" "strings" "testing" @@ -79,11 +80,11 @@ func TestRabbitDeadLetterReplayPublishesBeforeAck(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { if len(messages) != 1 || string(messages[0].Payload) != "payload" { t.Fatalf("transform messages = %+v", messages) } - return [][]byte{[]byte("transformed")}, nil + return []DeadLetterReplayTransformation{{Payload: []byte("transformed")}}, nil }, }) if err != nil { @@ -140,7 +141,7 @@ func TestRabbitDeadLetterReplayTransformFailureRestoresBatch(t *testing.T) { _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + Transform: func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { return nil, errors.New("refresh failed") }, }) @@ -152,6 +153,91 @@ func TestRabbitDeadLetterReplayTransformFailureRestoresBatch(t *testing.T) { } } +func TestRabbitDeadLetterReplayBatchesAndRetainsSkippedMessages(t *testing.T) { + client, err := NewRabbitMQ(NewRabbitOptions(). + SetConsumerQueue("events"). + SetDeadletterQueue("deadletter"). + SetHost("rabbitmq:5672"). + SetUsername("guest"). + SetPassword("guest"). + Build()) + if err != nil { + t.Fatalf("NewRabbitMQ: %v", err) + } + + acknowledger := &fakeRabbitAcknowledger{} + deliveries := make([]amqp.Delivery, 0, 5) + for index, payload := range []string{"one", "poison", "three", "four", "five"} { + envelope, encodeErr := encodeDeadLetter([]byte(payload), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if encodeErr != nil { + t.Fatal(encodeErr) + } + deliveries = append(deliveries, amqp.Delivery{ + Acknowledger: acknowledger, + DeliveryTag: uint64(index + 1), + Body: envelope, + }) + } + client.deadLetterGet = func() (amqp.Delivery, bool, error) { + if len(deliveries) == 0 { + return amqp.Delivery{}, false, nil + } + delivery := deliveries[0] + deliveries = deliveries[1:] + return delivery, true, nil + } + var batches []int + var replayed, restored []string + client.deadLetterReplayPublish = func(_ context.Context, destination string, payload []byte) error { + if destination == "events" { + replayed = append(replayed, string(payload)) + } else if destination == "deadletter" { + message, decodeErr := decodeDeadLetter("restored", payload) + if decodeErr != nil { + t.Fatal(decodeErr) + } + restored = append(restored, string(message.Payload)) + } + return nil + } + + result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 5, + BatchSize: 2, + Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { + batches = append(batches, len(messages)) + transformed := make([]DeadLetterReplayTransformation, len(messages)) + for index, message := range messages { + if string(message.Payload) == "poison" { + transformed[index].Skip = true + } else { + transformed[index].Payload = append([]byte("fresh-"), message.Payload...) + } + } + return transformed, nil + }, + }) + if err != nil { + t.Fatalf("ReplayDeadLetters: %v", err) + } + if !slices.Equal(batches, []int{2, 2, 1}) { + t.Fatalf("transform batch sizes = %v", batches) + } + if !slices.Equal(replayed, []string{"fresh-one", "fresh-three", "fresh-four", "fresh-five"}) || + !slices.Equal(restored, []string{"poison"}) { + t.Fatalf("replayed=%v restored=%v", replayed, restored) + } + if result.Scanned != 5 || result.Matched != 5 || result.Planned != 4 || + result.Replayed != 4 || result.Retained != 1 || result.Skipped != 1 || + result.Destinations["events"] != 4 { + t.Fatalf("result = %+v", result) + } +} + func TestRabbitPublishToDeadLetterQueueAddsEnvelope(t *testing.T) { client, err := NewRabbitMQ(NewRabbitOptions(). SetConsumerQueue("events"). diff --git a/pkg/queue/deadletter_sqs.go b/pkg/queue/deadletter_sqs.go index 0468989..506d456 100644 --- a/pkg/queue/deadletter_sqs.go +++ b/pkg/queue/deadletter_sqs.go @@ -39,7 +39,12 @@ func (s *SQS) InspectDeadLetters(ctx context.Context, request DeadLetterInspectR held := make([]types.Message, 0, request.Limit) for result.Scanned < request.Limit { - messages, receiveErr := s.receiveDeadLetterBatch(ctx, deadLetterQueueURL, request.Limit-result.Scanned) + messages, receiveErr := s.receiveDeadLetterBatch( + ctx, + deadLetterQueueURL, + request.Limit-result.Scanned, + deadLetterVisibilityTimeout(ctx, s.options.visibilityTimeout()), + ) if receiveErr != nil { return result, errors.Join(receiveErr, s.releaseSQSDeadLetters(deadLetterQueueURL, held)) } @@ -64,7 +69,10 @@ func (s *SQS) InspectDeadLetters(ctx context.Context, request DeadLetterInspectR func (s *SQS) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayRequest) (DeadLetterReplayResult, error) { var result DeadLetterReplayResult - if err := validateDeadLetterLimit(request.Limit); err != nil { + if err := validateDeadLetterReplayRequest(request); err != nil { + return result, err + } + if err := validateSQSDeadLetterReplayRequest(ctx, request, s.options.waitTimeSeconds()); err != nil { return result, err } if err := s.ensureConnected(); err != nil { @@ -79,68 +87,102 @@ func (s *SQS) ReplayDeadLetters(ctx context.Context, request DeadLetterReplayReq return errors.Join(operationErr, s.releaseSQSDeadLetters(deadLetterQueueURL, retained)) } type replayItem struct { - sourceMessage types.Message - message DeadLetterMessage - plan deadLetterReplayPlan - destinationURL string + sourceMessage types.Message + message DeadLetterMessage + plan deadLetterReplayPlan + } + batchSize := deadLetterReplayBatchSize(request) + visibilityTimeout, err := deadLetterReplayVisibilityTimeout(ctx, s.options.visibilityTimeout(), request) + if err != nil { + return result, err } - items := make([]replayItem, 0, request.Limit) for result.Scanned < request.Limit { - messages, receiveErr := s.receiveDeadLetterBatch(ctx, deadLetterQueueURL, request.Limit-result.Scanned) - if receiveErr != nil { - return result, releaseRetained(receiveErr) - } - if len(messages) == 0 { - break - } - retained = append(retained, messages...) - for _, message := range messages { - id := aws.ToString(message.MessageId) - decoded, decodeErr := decodeDeadLetter(id, []byte(aws.ToString(message.Body))) - if decodeErr != nil { - return result, releaseRetained(decodeErr) + batchCtx, cancelBatch := deadLetterReplayBatchContext(ctx, request.BatchTimeout) + items := make([]replayItem, 0, batchSize) + batchScanned := 0 + exhausted := false + for batchScanned < batchSize && result.Scanned < request.Limit { + receiveLimit := batchSize - batchScanned + if remaining := request.Limit - result.Scanned; receiveLimit > remaining { + receiveLimit = remaining } - plan, planErr := planDeadLetterReplay(&result, decoded, request, s.options.DeadletterQueue) - if planErr != nil { - return result, releaseRetained(planErr) + messages, receiveErr := s.receiveDeadLetterBatch(batchCtx, deadLetterQueueURL, receiveLimit, visibilityTimeout) + if receiveErr != nil { + cancelBatch() + return result, releaseRetained(receiveErr) } - if plan.destination == "" { - continue + if len(messages) == 0 { + exhausted = true + break } - item := replayItem{sourceMessage: message, message: decoded, plan: plan} - if request.Execute { - destinationURL, resolveErr := s.queueURL(ctx, plan.destination) - if resolveErr != nil { - return result, releaseRetained(resolveErr) + batchScanned += len(messages) + for _, message := range messages { + retained = append(retained, types.Message{ReceiptHandle: message.ReceiptHandle}) + } + for _, message := range messages { + id := aws.ToString(message.MessageId) + decoded, decodeErr := decodeDeadLetter(id, []byte(aws.ToString(message.Body))) + if decodeErr != nil { + cancelBatch() + return result, releaseRetained(decodeErr) + } + plan, planErr := planDeadLetterReplay(&result, decoded, request, s.options.DeadletterQueue) + if planErr != nil { + cancelBatch() + return result, releaseRetained(planErr) } - if destinationURL == deadLetterQueueURL { - return result, releaseRetained(fmt.Errorf("refusing to replay SQS dead-letter message %q back to %q", decoded.ID, s.options.DeadletterQueue)) + if plan.destination != "" { + items = append(items, replayItem{sourceMessage: message, message: decoded, plan: plan}) } - item.destinationURL = destinationURL } - items = append(items, item) } - } - decodedMessages := make([]DeadLetterMessage, len(items)) - for index := range items { - decodedMessages[index] = items[index].message - } - payloads, err := transformDeadLetterReplayMessages(ctx, request.Transform, decodedMessages) - if err != nil { - return result, releaseRetained(err) - } - if request.Execute { + decodedMessages := make([]DeadLetterMessage, len(items)) + for index := range items { + decodedMessages[index] = items[index].message + } + transformations, err := transformDeadLetterReplayMessages(batchCtx, request.Transform, decodedMessages) + if err != nil { + cancelBatch() + return result, releaseRetained(err) + } for index, item := range items { - if err := s.publishReplay(ctx, item.plan.destination, item.destinationURL, payloads[index]); err != nil { + if transformations[index].Skip { + skipDeadLetterReplay(&result, item.plan.destination, request.Execute) + continue + } + if !request.Execute { + continue + } + destinationURL, resolveErr := s.queueURL(batchCtx, item.plan.destination) + if resolveErr != nil { + cancelBatch() + return result, releaseRetained(resolveErr) + } + if destinationURL == deadLetterQueueURL { + cancelBatch() + return result, releaseRetained(fmt.Errorf("refusing to replay SQS dead-letter message %q back to %q", item.message.ID, s.options.DeadletterQueue)) + } + if err := s.publishReplay(batchCtx, item.plan.destination, destinationURL, transformations[index].Payload); err != nil { + cancelBatch() return result, releaseRetained(fmt.Errorf("replay SQS dead-letter message %q: %w", item.message.ID, err)) } - if err := s.deleteSQSDeadLetter(ctx, deadLetterQueueURL, item.sourceMessage); err != nil { + if err := s.deleteSQSDeadLetter(batchCtx, deadLetterQueueURL, item.sourceMessage); err != nil { + cancelBatch() return result, releaseRetained(fmt.Errorf("settle replayed SQS dead-letter message %q: %w", item.message.ID, err)) } retained = removeSQSDeadLetter(retained, aws.ToString(item.sourceMessage.ReceiptHandle)) result.Replayed++ } + cancelBatch() + if exhausted || result.Scanned >= request.Limit { + break + } + if request.Execute { + if err := waitForDeadLetterReplayBatch(ctx, request.BatchDelay); err != nil { + return result, releaseRetained(err) + } + } } if err := s.releaseSQSDeadLetters(deadLetterQueueURL, retained); err != nil { return result, err @@ -172,7 +214,7 @@ func (s *SQS) publishReplay(ctx context.Context, queueName, queueURL string, pay return nil } -func (s *SQS) receiveDeadLetterBatch(ctx context.Context, queueURL string, limit int) ([]types.Message, error) { +func (s *SQS) receiveDeadLetterBatch(ctx context.Context, queueURL string, limit int, visibilityTimeout int32) ([]types.Message, error) { client, _ := s.clientAndConsumerURL() batchSize := limit if batchSize > int(defaultSQSMaxNumberOfMessages) { @@ -181,7 +223,7 @@ func (s *SQS) receiveDeadLetterBatch(ctx context.Context, queueURL string, limit output, err := client.ReceiveMessage(ctx, &awssqs.ReceiveMessageInput{ QueueUrl: aws.String(queueURL), MaxNumberOfMessages: int32(batchSize), - VisibilityTimeout: deadLetterVisibilityTimeout(ctx, s.options.visibilityTimeout()), + VisibilityTimeout: visibilityTimeout, WaitTimeSeconds: s.options.waitTimeSeconds(), }) if err != nil { @@ -190,6 +232,58 @@ func (s *SQS) receiveDeadLetterBatch(ctx context.Context, queueURL string, limit return output.Messages, nil } +func deadLetterReplayVisibilityTimeout(ctx context.Context, configured int32, request DeadLetterReplayRequest) (int32, error) { + timeout := configured + if timeout < 300 { + timeout = 300 + } + batchSize := deadLetterReplayBatchSize(request) + batches := (request.Limit + batchSize - 1) / batchSize + const maximum = 12 * time.Hour + required := time.Duration(0) + if request.BatchTimeout > 0 { + if request.BatchTimeout >= maximum || request.BatchDelay >= maximum-request.BatchTimeout { + return 0, fmt.Errorf("SQS batched replay duration exceeds the 12-hour maximum visibility timeout") + } + perBatch := request.BatchTimeout + request.BatchDelay + required = 5 * time.Minute + if batches > int((maximum-required)/perBatch) { + return 0, fmt.Errorf("SQS batched replay duration exceeds the 12-hour maximum visibility timeout") + } else { + required += time.Duration(batches) * perBatch + } + } + if deadline, ok := ctx.Deadline(); ok { + contextBound := time.Until(deadline) + 30*time.Second + if contextBound < 0 { + contextBound = 0 + } + if required == 0 || contextBound < required { + required = contextBound + } + } + seconds := int32((required + time.Second - 1) / time.Second) + if seconds > timeout { + timeout = seconds + } + if timeout > int32(maximum/time.Second) { + return 0, fmt.Errorf("SQS batched replay duration exceeds the 12-hour maximum visibility timeout") + } + return timeout, nil +} + +func validateSQSDeadLetterReplayRequest(ctx context.Context, request DeadLetterReplayRequest, waitTimeSeconds int32) error { + if request.BatchSize > 0 && request.BatchTimeout == 0 { + if _, bounded := ctx.Deadline(); !bounded { + return fmt.Errorf("SQS batched replay requires BatchTimeout or a context deadline") + } + } + if request.BatchTimeout > 0 && request.BatchTimeout <= time.Duration(waitTimeSeconds)*time.Second { + return fmt.Errorf("SQS dead-letter batch timeout must exceed the %s long-poll duration", time.Duration(waitTimeSeconds)*time.Second) + } + return nil +} + func (s *SQS) releaseSQSDeadLetters(queueURL string, messages []types.Message) error { if len(messages) == 0 { return nil diff --git a/pkg/queue/deadletter_test.go b/pkg/queue/deadletter_test.go index cb9b991..d3b5a47 100644 --- a/pkg/queue/deadletter_test.go +++ b/pkg/queue/deadletter_test.go @@ -171,26 +171,53 @@ func TestTransformDeadLetterReplayMessages(t *testing.T) { {ID: "one", Payload: []byte("first")}, {ID: "two", Payload: []byte("second")}, } - payloads, err := transformDeadLetterReplayMessages(context.Background(), func(_ context.Context, got []DeadLetterMessage) ([][]byte, error) { + transformations, err := transformDeadLetterReplayMessages(context.Background(), func(_ context.Context, got []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { if len(got) != 2 || got[0].ID != "one" || got[1].ID != "two" { t.Fatalf("messages = %+v", got) } - return [][]byte{[]byte("transformed-first"), []byte("transformed-second")}, nil + return []DeadLetterReplayTransformation{ + {Payload: []byte("transformed-first")}, + {Payload: []byte("transformed-second")}, + }, nil }, messages) if err != nil { t.Fatal(err) } - if !bytes.Equal(payloads[0], []byte("transformed-first")) || - !bytes.Equal(payloads[1], []byte("transformed-second")) { - t.Fatalf("payloads = %q", payloads) + if !bytes.Equal(transformations[0].Payload, []byte("transformed-first")) || + !bytes.Equal(transformations[1].Payload, []byte("transformed-second")) { + t.Fatalf("transformations = %+v", transformations) } } func TestTransformDeadLetterReplayMessagesRequiresMatchingResultCount(t *testing.T) { - _, err := transformDeadLetterReplayMessages(context.Background(), func(context.Context, []DeadLetterMessage) ([][]byte, error) { + _, err := transformDeadLetterReplayMessages(context.Background(), func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { return nil, nil }, []DeadLetterMessage{{ID: "one"}}) if err == nil { t.Fatal("expected transform result count error") } } + +func TestValidateDeadLetterReplayRequestAllowsLargeBatchedLimit(t *testing.T) { + if err := validateDeadLetterReplayRequest(DeadLetterReplayRequest{ + Limit: 30000, + BatchSize: 100, + BatchDelay: time.Second, + BatchTimeout: time.Minute, + }); err != nil { + t.Fatal(err) + } + if err := validateDeadLetterReplayRequest(DeadLetterReplayRequest{ + Limit: 30000, + BatchSize: 0, + }); err == nil { + t.Fatal("expected the legacy unbatched limit to remain capped") + } + if err := validateDeadLetterReplayRequest(DeadLetterReplayRequest{ + Limit: 1, + BatchSize: 1, + BatchTimeout: -time.Second, + }); err == nil { + t.Fatal("expected a negative batch timeout to be rejected") + } +} diff --git a/pkg/queue/kafka_test.go b/pkg/queue/kafka_test.go index b16fcbd..2f05bb7 100644 --- a/pkg/queue/kafka_test.go +++ b/pkg/queue/kafka_test.go @@ -29,6 +29,36 @@ type assignmentThenMessageKafkaConsumer struct { readCount int } +type queuedKafkaConsumer struct { + messages []*kafka.Message + index int + commits []*kafka.Message +} + +func (*queuedKafkaConsumer) SubscribeTopics([]string, kafka.RebalanceCb) error { + return nil +} + +func (f *queuedKafkaConsumer) ReadMessage(time.Duration) (*kafka.Message, error) { + if f.index >= len(f.messages) { + return nil, kafka.NewError(kafka.ErrTimedOut, "timed out", false) + } + message := f.messages[f.index] + f.index++ + return message, nil +} + +func (f *queuedKafkaConsumer) CommitMessage(message *kafka.Message) ([]kafka.TopicPartition, error) { + f.commits = append(f.commits, message) + return nil, nil +} + +func (*queuedKafkaConsumer) Assignment() ([]kafka.TopicPartition, error) { + return []kafka.TopicPartition{{Partition: 0}, {Partition: 1}}, nil +} + +func (*queuedKafkaConsumer) Close() error { return nil } + func (*assignmentThenMessageKafkaConsumer) SubscribeTopics([]string, kafka.RebalanceCb) error { return nil } @@ -161,11 +191,11 @@ func TestKafkaDeadLetterReplayPublishesBeforeCommit(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { if len(messages) != 1 || string(messages[0].Payload) != "payload" { t.Fatalf("transform messages = %+v", messages) } - return [][]byte{[]byte("transformed")}, nil + return []DeadLetterReplayTransformation{{Payload: []byte("transformed")}}, nil }, }) if err != nil { @@ -200,7 +230,7 @@ func TestKafkaDeadLetterReplayTransformFailureDoesNotPublishOrCommit(t *testing. _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + Transform: func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { return nil, errors.New("refresh failed") }, }) @@ -212,6 +242,71 @@ func TestKafkaDeadLetterReplayTransformFailureDoesNotPublishOrCommit(t *testing. } } +func TestKafkaDeadLetterReplayBlocksOnlySkippedPartition(t *testing.T) { + var messages []*kafka.Message + for index, item := range []struct { + partition int32 + payload string + }{ + {partition: 0, payload: "one"}, + {partition: 0, payload: "poison"}, + {partition: 0, payload: "blocked"}, + {partition: 1, payload: "other-partition"}, + } { + envelope, err := encodeDeadLetter([]byte(item.payload), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if err != nil { + t.Fatal(err) + } + messages = append(messages, &kafka.Message{ + TopicPartition: kafka.TopicPartition{ + Topic: stringPointer("deadletter"), + Partition: item.partition, + Offset: kafka.Offset(index), + }, + Value: envelope, + }) + } + client, _, producer := newTestKafka(t, nil) + consumer := &queuedKafkaConsumer{messages: messages} + client.Consumer = consumer + client.options.DisableAutoTopicCreation = true + + result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 4, + BatchSize: 4, + Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { + transformed := make([]DeadLetterReplayTransformation, len(messages)) + for index, message := range messages { + if string(message.Payload) == "poison" { + transformed[index].Skip = true + } else { + transformed[index].Payload = append([]byte("fresh-"), message.Payload...) + } + } + return transformed, nil + }, + }) + if err != nil { + t.Fatalf("ReplayDeadLetters: %v", err) + } + if len(producer.messages) != 2 || string(producer.messages[0].Value) != "fresh-one" || + string(producer.messages[1].Value) != "fresh-other-partition" { + t.Fatalf("published messages = %+v", producer.messages) + } + if len(consumer.commits) != 2 || consumer.commits[0] != messages[0] || consumer.commits[1] != messages[3] { + t.Fatalf("commits = %+v", consumer.commits) + } + if result.Scanned != 4 || result.Matched != 4 || result.Planned != 2 || + result.Replayed != 2 || result.Retained != 2 || result.Skipped != 1 || + result.Destinations["events"] != 2 { + t.Fatalf("result = %+v", result) + } +} + func TestKafkaDeadLetterReplayRejectsMissingDestinationBeforePublishOrCommit(t *testing.T) { envelope, err := encodeDeadLetter([]byte("payload"), DeadLetterMetadata{ Source: "missing-events", diff --git a/pkg/queue/sqs_test.go b/pkg/queue/sqs_test.go index 4a46c97..f6f596c 100644 --- a/pkg/queue/sqs_test.go +++ b/pkg/queue/sqs_test.go @@ -4,8 +4,11 @@ import ( "context" "encoding/json" "errors" + "fmt" + "slices" "strings" "testing" + "time" "github.com/aws/aws-sdk-go-v2/aws" awssqs "github.com/aws/aws-sdk-go-v2/service/sqs" @@ -34,8 +37,14 @@ func (f *fakeSQSClient) GetQueueUrl(_ context.Context, input *awssqs.GetQueueUrl return &awssqs.GetQueueUrlOutput{QueueUrl: aws.String(queueURL)}, nil } -func (f *fakeSQSClient) ReceiveMessage(context.Context, *awssqs.ReceiveMessageInput, ...func(*awssqs.Options)) (*awssqs.ReceiveMessageOutput, error) { - return &awssqs.ReceiveMessageOutput{Messages: f.receiveMessages}, nil +func (f *fakeSQSClient) ReceiveMessage(_ context.Context, input *awssqs.ReceiveMessageInput, _ ...func(*awssqs.Options)) (*awssqs.ReceiveMessageOutput, error) { + count := int(input.MaxNumberOfMessages) + if count > len(f.receiveMessages) { + count = len(f.receiveMessages) + } + messages := append([]types.Message(nil), f.receiveMessages[:count]...) + f.receiveMessages = f.receiveMessages[count:] + return &awssqs.ReceiveMessageOutput{Messages: messages}, nil } func (f *fakeSQSClient) SendMessage(_ context.Context, input *awssqs.SendMessageInput, _ ...func(*awssqs.Options)) (*awssqs.SendMessageOutput, error) { @@ -110,11 +119,11 @@ func TestSQSDeadLetterReplaySendsBeforeDelete(t *testing.T) { result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(_ context.Context, messages []DeadLetterMessage) ([][]byte, error) { + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { if len(messages) != 1 || string(messages[0].Payload) != "payload" { t.Fatalf("transform messages = %+v", messages) } - return [][]byte{[]byte("transformed")}, nil + return []DeadLetterReplayTransformation{{Payload: []byte("transformed")}}, nil }, }) if err != nil { @@ -146,7 +155,7 @@ func TestSQSDeadLetterReplayTransformFailureReleasesBatch(t *testing.T) { _, err = client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ Limit: 1, Execute: true, - Transform: func(context.Context, []DeadLetterMessage) ([][]byte, error) { + Transform: func(context.Context, []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { return nil, errors.New("refresh failed") }, }) @@ -158,6 +167,121 @@ func TestSQSDeadLetterReplayTransformFailureReleasesBatch(t *testing.T) { } } +func TestSQSDeadLetterReplayBatchesAndRetainsSkippedMessages(t *testing.T) { + client, fake := newTestSQS(t, types.Message{}) + fake.deleteError = nil + fake.visibilityError = nil + fake.receiveMessages = nil + for index, payload := range []string{"one", "poison", "three", "four", "five"} { + envelope, err := encodeDeadLetter([]byte(payload), DeadLetterMetadata{ + Source: "events", + Destination: "deadletter", + }) + if err != nil { + t.Fatal(err) + } + fake.receiveMessages = append(fake.receiveMessages, types.Message{ + MessageId: aws.String(fmt.Sprintf("message-%d", index)), + Body: aws.String(string(envelope)), + ReceiptHandle: aws.String(fmt.Sprintf("receipt-%d", index)), + }) + } + + var batches []int + result, err := client.ReplayDeadLetters(context.Background(), DeadLetterReplayRequest{ + Limit: 5, + BatchSize: 2, + BatchTimeout: time.Minute, + Execute: true, + Transform: func(_ context.Context, messages []DeadLetterMessage) ([]DeadLetterReplayTransformation, error) { + batches = append(batches, len(messages)) + transformed := make([]DeadLetterReplayTransformation, len(messages)) + for index, message := range messages { + if string(message.Payload) == "poison" { + transformed[index].Skip = true + } else { + transformed[index].Payload = append([]byte("fresh-"), message.Payload...) + } + } + return transformed, nil + }, + }) + if err != nil { + t.Fatalf("ReplayDeadLetters: %v", err) + } + if !slices.Equal(batches, []int{2, 2, 1}) { + t.Fatalf("transform batch sizes = %v", batches) + } + if len(fake.sent) != 4 || len(fake.deleted) != 4 || len(fake.visibilityChanges) != 1 { + t.Fatalf("sent=%d deleted=%d visibility=%d", len(fake.sent), len(fake.deleted), len(fake.visibilityChanges)) + } + if result.Scanned != 5 || result.Matched != 5 || result.Planned != 4 || + result.Replayed != 4 || result.Retained != 1 || result.Skipped != 1 || + result.Destinations["events"] != 4 { + t.Fatalf("result = %+v", result) + } +} + +func TestSQSDeadLetterReplayVisibilityCoversBatchedOperation(t *testing.T) { + timeout, err := deadLetterReplayVisibilityTimeout(context.Background(), 30, DeadLetterReplayRequest{ + Limit: 30000, + BatchSize: 100, + BatchDelay: time.Second, + BatchTimeout: time.Minute, + }) + if err != nil { + t.Fatal(err) + } + if timeout != 18600 { + t.Fatalf("visibility timeout = %d, want 18600", timeout) + } + + _, err = deadLetterReplayVisibilityTimeout(context.Background(), 30, DeadLetterReplayRequest{ + Limit: 1000000, + BatchSize: 100, + BatchTimeout: time.Minute, + }) + if err == nil || !strings.Contains(err.Error(), "12-hour maximum") { + t.Fatalf("error = %v", err) + } +} + +func TestSQSDeadLetterBatchedReplayRequiresBoundedOperation(t *testing.T) { + request := DeadLetterReplayRequest{ + Limit: 30000, + BatchSize: 100, + } + err := validateSQSDeadLetterReplayRequest(context.Background(), request, 20) + if err == nil || !strings.Contains(err.Error(), "BatchTimeout or a context deadline") { + t.Fatalf("error = %v", err) + } + + ctx, cancel := context.WithTimeout(context.Background(), time.Minute) + defer cancel() + if err := validateSQSDeadLetterReplayRequest(ctx, request, 20); err != nil { + t.Fatal(err) + } + longCtx, longCancel := context.WithTimeout(context.Background(), 6*time.Hour) + defer longCancel() + timeout, err := deadLetterReplayVisibilityTimeout(longCtx, 30, DeadLetterReplayRequest{ + Limit: 100000, + BatchSize: 10, + BatchDelay: time.Second, + }) + if err != nil { + t.Fatal(err) + } + if timeout < 6*60*60 { + t.Fatalf("visibility timeout = %d, want the overall context bound", timeout) + } + + request.BatchTimeout = 20 * time.Second + err = validateSQSDeadLetterReplayRequest(context.Background(), request, 20) + if err == nil || !strings.Contains(err.Error(), "must exceed") { + t.Fatalf("error = %v", err) + } +} + func TestSQSDeadLetterEnvelopeRecordsRouter(t *testing.T) { client, _ := newTestSQS(t, types.Message{}) envelope, err := client.deadLetterEnvelope([]byte("payload"), DeadLetterReasonHandlerError, 0)