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 {