diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index fd0f293..6c54538 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -100,12 +100,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * } var pending []pendingAck for _, msg := range resp.Messages { - if msg.Body == nil || *msg.Body == "" { - log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId)) - pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) - continue + // nil/пустое тело не подтверждаем молча: прогоняем через processTelemetry, + // который вернёт errInvalidEnvelope → DLQ (ревью 2026-10-10: пустое тело + // подтверждалось без DLQ). + body := "" + if msg.Body != nil { + body = *msg.Body } - if err := processTelemetry(ctx, *msg.Body, store, log); err != nil { + if err := processTelemetry(ctx, body, store, log); err != nil { // Битый envelope → DLQ (а не бесконечный ретрай и не потеря): // кладём тело в DLQ, затем подтверждаем основное сообщение. if errors.Is(err, errInvalidEnvelope) { @@ -115,7 +117,7 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * // зациклиться навсегда (находка #2): после maxReceiveCount // дропаем битое с громким логом. if dlqURL != "" { - if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr == nil { + if derr := sendToDLQ(ctx, sqsClient, dlqURL, body, log); derr == nil { pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) continue } else {