From d7701f28456c0048ade0561b7aac9570a921c037 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sat, 10 Oct 2026 19:10:12 +0300
Subject: [PATCH] =?UTF-8?q?fix(iot):=20=D0=BE=D0=B3=D1=80=D0=B0=D0=BD?=
 =?UTF-8?q?=D0=B8=D1=87=D0=B8=D1=82=D1=8C=20=D0=BF=D0=BE=D0=B2=D1=82=D0=BE?=
 =?UTF-8?q?=D1=80=D1=8B=20=D1=8F=D0=B4=D0=BE=D0=B2=D0=B8=D1=82=D0=BE=D0=B3?=
 =?UTF-8?q?=D0=BE=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD=D0=B8=D1=8F?=
 =?UTF-8?q?=20(=D0=BD=D0=B0=D1=85=D0=BE=D0=B4=D0=BA=D0=B0=20#2)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Было: при недоступной DLQ битое сообщение НЕ подтверждалось и крутилось на
основной очереди бесконечно. Теперь читается ApproximateReceiveCount; после
maxReceiveCount=5 битое сообщение подтверждается (дроп с громким логом), а не
зацикливается. При доступной DLQ — уходит туда на первом приёме.
---
 internal/service/consumer/consumer.go | 41 +++++++++++++++++++++++----
 1 file changed, 35 insertions(+), 6 deletions(-)

diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go
index 6171b24..0c3c5a8 100644
--- a/internal/service/consumer/consumer.go
+++ b/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 реально приняла сообщение: иначе
-					// битый теряется без следа (ревью 🔴7). При недоступной DLQ или
-					// сбое sendToDLQ НЕ подтверждаем — сообщение вернётся после
-					// visibility timeout (at-least-once).
+					// Битый envelope → DLQ. Подтверждаем только если DLQ реально
+					// приняла сообщение (иначе битый теряется — 🔴7). Если DLQ
+					// недоступна/сбой — ограничиваем число повторов, чтобы не
+					// зациклиться навсегда (находка #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 {