fix(iot): ограничить повторы ядовитого сообщения (находка #2)

Было: при недоступной DLQ битое сообщение НЕ подтверждалось и крутилось на
основной очереди бесконечно. Теперь читается ApproximateReceiveCount; после
maxReceiveCount=5 битое сообщение подтверждается (дроп с громким логом), а не
зацикливается. При доступной DLQ — уходит туда на первом приёме.

internal/service/consumer/consumer.go

@@ -8,10 +8,12 @@ import (
 	"context"
 	"errors"
 	"log/slog"
+	"strconv"
 	"time"
 
 	"github.com/aws/aws-sdk-go-v2/aws"
 	"github.com/aws/aws-sdk-go-v2/service/sqs"
+	sqstypes "github.com/aws/aws-sdk-go-v2/service/sqs/types"
 
 	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/config"
 	"gitea.services.ngcloud.ru/Nail/IoT/internal/service/sqsclient"
@@ -24,6 +26,12 @@ const errorBackoff = 5 * time.Second
 // maxProcessBackoff — потолок backoff после ошибок обработки (PG).
 const maxProcessBackoff = 60 * time.Second
 
+// maxReceiveCount — после стольких приёмов «ядовитое» сообщение дропается, если
+// DLQ недоступна: иначе оно крутилось бы на основной очереди бесконечно
+// (находка #2 — без чтения ApproximateReceiveCount). При доступной DLQ битое
+// уходит туда на первом же приёме.
+const maxReceiveCount = 5
+
 // Run — бесконечный long-poll цикл потребителя.
 func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *iotpg.IoTPostgresStore, log *slog.Logger) error {
 	queueURL, err := sqsclient.ResolveQueueURLWithRetry(ctx, sqsClient, cfg.SQSQueueName)
@@ -61,6 +69,11 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 			MaxNumberOfMessages: 10,
 			WaitTimeSeconds:     int32(cfg.SQSLongPollSeconds),
 			VisibilityTimeout:   int32(cfg.SQSVisibilityTimeout),
+			// Просим ApproximateReceiveCount — по нему ограничиваем повторы
+			// «ядовитого» сообщения (см. maxReceiveCount, находка #2).
+			MessageSystemAttributeNames: []sqstypes.MessageSystemAttributeName{
+				sqstypes.MessageSystemAttributeNameApproximateReceiveCount,
+			},
 		})
 		if err != nil {
 			if ctx.Err() != nil {
@@ -93,18 +106,27 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
 				// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
 				// кладём тело в DLQ, затем подтверждаем основное сообщение.
 				if errors.Is(err, errInvalidEnvelope) {
-					// Подтверждаем ТОЛЬКО если DLQ реально приняла сообщение: иначе
+					// Битый envelope → DLQ. Подтверждаем только если DLQ реально
-					// битый теряется без следа (ревью 🔴7). При недоступной DLQ или
+					// приняла сообщение (иначе битый теряется — 🔴7). Если DLQ
-					// сбое sendToDLQ НЕ подтверждаем — сообщение вернётся после
+					// недоступна/сбой — ограничиваем число повторов, чтобы не
-					// visibility timeout (at-least-once).
+					// зациклиться навсегда (находка #2): после maxReceiveCount
+					// дропаем битое с громким логом.
 					if dlqURL != "" {
 						if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr == nil {
 							pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
+							continue
 						} else {
-							log.Error("consumer: send to DLQ failed — сообщение НЕ подтверждено, вернётся после visibility timeout", "err", derr, "message_id", aws.ToString(msg.MessageId))
+							log.Error("consumer: send to DLQ failed", "err", derr, "message_id", aws.ToString(msg.MessageId))
 						}
+					}
+					recv := receiveCount(msg)
+					if recv >= maxReceiveCount {
+						log.Error("consumer: invalid envelope dropped after maxReceiveCount (DLQ недоступна)",
+							"message_id", aws.ToString(msg.MessageId), "receive_count", recv, "max", maxReceiveCount)
+						pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
 					} else {
-						log.Warn("consumer: DLQ недоступна — битое сообщение НЕ подтверждено, вернётся после visibility timeout", "message_id", aws.ToString(msg.MessageId))
+						log.Warn("consumer: invalid envelope deferred (DLQ недоступна, будет повтор)",
+							"message_id", aws.ToString(msg.MessageId), "receive_count", recv)
 					}
 					continue
 				}
@@ -162,6 +184,13 @@ func sendToDLQ(ctx context.Context, client *sqs.Client, dlqURL, body string, log
 	return err
 }
 
+// receiveCount — ApproximateReceiveCount сообщения (число приёмов из очереди).
+// 0, если атрибут не запрошен/не задан.
+func receiveCount(msg sqstypes.Message) int {
+	n, _ := strconv.Atoi(msg.Attributes[string(sqstypes.MessageSystemAttributeNameApproximateReceiveCount)])
+	return n
+}
+
 // sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён.
 func sleepCtx(ctx context.Context, d time.Duration) bool {
 	select {