From 5ad86d629633830b147cb1100e6710e77577ef06 Mon Sep 17 00:00:00 2001 From: windyboy Date: Wed, 19 Nov 2025 17:09:52 +0800 Subject: [PATCH] =?UTF-8?q?=E2=9C=A8=20Enable=20Dead-Letter=20Queue=20(DLQ?= =?UTF-8?q?)=20in=20development=20configuration=20and=20enhance=20NATS=20c?= =?UTF-8?q?onsumer=20error=20handling.=20Update=20`config.dev.toml`=20to?= =?UTF-8?q?=20enable=20DLQ=20routing=20for=20failed=20messages.=20Modify?= =?UTF-8?q?=20NATS=20consumer=20to=20append=20DLQ=20subject=20to=20stream?= =?UTF-8?q?=20subjects=20if=20enabled.=20Improve=20error=20logging=20in=20?= =?UTF-8?q?message=20processing=20to=20include=20message=20IDs=20and=20DLQ?= =?UTF-8?q?=20routing=20status,=20ensuring=20better=20observability=20and?= =?UTF-8?q?=20error=20management.?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- configs/config.dev.toml | 2 +- internal/infra/nats/consumer.go | 4 ++ internal/infra/nats/message_processor.go | 61 ++++++++++++++++++++++-- internal/infra/nats/stream_manager.go | 42 ++++++++++++++-- 4 files changed, 98 insertions(+), 11 deletions(-) diff --git a/configs/config.dev.toml b/configs/config.dev.toml index 15505a6..9c8787f 100644 --- a/configs/config.dev.toml +++ b/configs/config.dev.toml @@ -123,6 +123,6 @@ health_timeout = "2s" [dlq] # Dead-Letter Queue configuration (only applies when mode = "jetstream") # enabled: Enable DLQ routing for poison messages (messages that fail after max_deliver attempts) -enabled = false # Set to true when switching to JetStream mode +enabled = true # Set to true when switching to JetStream mode # subject: NATS subject where failed messages will be published for manual inspection subject = "caatsm.dlq" diff --git a/internal/infra/nats/consumer.go b/internal/infra/nats/consumer.go index 6eb0049..e83011d 100644 --- a/internal/infra/nats/consumer.go +++ b/internal/infra/nats/consumer.go @@ -105,6 +105,10 @@ func ProvideConsumer( if publisherSubject := strings.TrimSpace(cfg.Publisher.Topic); publisherSubject != "" { streamSubjects = append(streamSubjects, publisherSubject) } + // Add DLQ subject to stream if DLQ is enabled + if normCfg.dlqSubject != "" { + streamSubjects = append(streamSubjects, normCfg.dlqSubject) + } streamSubjects = dedupeSubjects(streamSubjects) consumer.streamManager = NewStreamManager(js, normCfg.streamName, streamSubjects, logger) diff --git a/internal/infra/nats/message_processor.go b/internal/infra/nats/message_processor.go index 655f9ef..dbb0760 100644 --- a/internal/infra/nats/message_processor.go +++ b/internal/infra/nats/message_processor.go @@ -174,14 +174,32 @@ func (p *defaultBatchProcessor) resolveMsgID(msg *nats.Msg) (string, string, err // handleMessageError handles errors that occur during message processing. func (p *defaultBatchProcessor) handleMessageError(ctx context.Context, msg *nats.Msg, err error, elapsed time.Duration) { + // Check if context is cancelled before processing + select { + case <-ctx.Done(): + p.logger.Warn("Skipping error handling due to context cancellation", + zap.String("subject", msg.Subject), + ) + return + default: + } + + // Extract message ID for better error logging + msgID, _, _ := p.resolveMsgID(msg) + if msgID == "" { + msgID = "unknown" + } + + isPermanent := app.IsPermanent(err) p.logger.Error("Failed to process message", zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), zap.Error(err), - zap.Bool("permanent", app.IsPermanent(err)), + zap.Bool("permanent", isPermanent), ) result := obsmetrics.ResultFail - if app.IsPermanent(err) { + if isPermanent { result = obsmetrics.ResultPermanentFail } p.telemetry.RecordMessageHandled(ctx, p.streamName, p.consumerName, result, elapsed) @@ -191,7 +209,6 @@ func (p *defaultBatchProcessor) handleMessageError(ctx context.Context, msg *nat consecutiveErrors = *p.consecutiveProcessErrors } - isPermanent := app.IsPermanent(err) processingResult := ProcessingErrorResult{IsPermanent: isPermanent} if !isPermanent && consecutiveErrors >= 10 { processingResult.ShouldApplyBackpressure = true @@ -222,14 +239,48 @@ func (p *defaultBatchProcessor) handlePermanentError(ctx context.Context, msg *n return } + // Extract message ID for better logging + msgID, _, _ := p.resolveMsgID(msg) + if msgID == "" { + msgID = "unknown" + } + // Poison/permanent message: route to DLQ if configured, then ACK + dlqRouted := false if p.dlqHandler != nil { if dlqErr := p.dlqHandler.RouteToDLQ(ctx, msg, err); dlqErr != nil { - p.logger.Error("Failed to route permanent-error message to DLQ", zap.Error(dlqErr)) + p.logger.Error("Failed to route permanent-error message to DLQ", + zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), + zap.Error(dlqErr), + zap.NamedError("original_error", err), + ) + // Note: We still ACK the message even if DLQ routing fails to prevent + // infinite redelivery of poison messages. The error is logged for manual investigation. + } else { + dlqRouted = true + p.logger.Info("Permanent-error message routed to DLQ", + zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), + ) } + } else { + p.logger.Warn("Permanent-error message but DLQ handler not configured - message will be ACKed without DLQ routing", + zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), + zap.String("hint", "Enable DLQ by setting dlq.enabled=true and dlq.subject in config to route poison messages for inspection"), + ) } + + // ACK the message to prevent redelivery + // Even if DLQ routing failed, we ACK to avoid infinite retries of poison messages if ackErr := msg.Ack(); ackErr != nil { - p.logger.Error("Failed to ACK permanent-error message", zap.Error(ackErr)) + p.logger.Error("Failed to ACK permanent-error message", + zap.String("subject", msg.Subject), + zap.String("msg_id", msgID), + zap.Bool("dlq_routed", dlqRouted), + zap.Error(ackErr), + ) } } diff --git a/internal/infra/nats/stream_manager.go b/internal/infra/nats/stream_manager.go index 335dbd0..de0c588 100644 --- a/internal/infra/nats/stream_manager.go +++ b/internal/infra/nats/stream_manager.go @@ -40,12 +40,44 @@ func NewStreamManager(js nats.JetStreamContext, streamName string, subjects []st // EnsureStream ensures that the configured JetStream stream exists, creating it if necessary func (sm *StreamManager) EnsureStream(cfg *StreamConfig) error { // Check if stream already exists - _, err := sm.js.StreamInfo(sm.streamName) + info, err := sm.js.StreamInfo(sm.streamName) if err == nil { - sm.logger.Info("JetStream stream verified", - zap.String("stream", sm.streamName), - zap.Strings("subjects", sm.subjects), - ) + // Stream exists - check if we need to add any missing subjects + existingSubjects := make(map[string]bool) + for _, subj := range info.Config.Subjects { + existingSubjects[subj] = true + } + + // Check if any configured subjects are missing + missingSubjects := []string{} + for _, subj := range sm.subjects { + if !existingSubjects[subj] { + missingSubjects = append(missingSubjects, subj) + } + } + + if len(missingSubjects) > 0 { + // Update stream to include missing subjects + updatedSubjects := info.Config.Subjects + updatedSubjects = append(updatedSubjects, missingSubjects...) + info.Config.Subjects = updatedSubjects + + _, updateErr := sm.js.UpdateStream(&info.Config) + if updateErr != nil { + return fmt.Errorf("failed to update stream %s with new subjects %v: %w", sm.streamName, missingSubjects, updateErr) + } + + sm.logger.Info("Updated JetStream stream with new subjects", + zap.String("stream", sm.streamName), + zap.Strings("added_subjects", missingSubjects), + zap.Strings("all_subjects", updatedSubjects), + ) + } else { + sm.logger.Info("JetStream stream verified", + zap.String("stream", sm.streamName), + zap.Strings("subjects", sm.subjects), + ) + } return nil }