diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index 4a640d7..bec668e 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -30,12 +30,24 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * if err != nil { return err } - // DLQ присутствует не всегда: резолвим лениво, отказ не фатален (битые - // сообщения тогда просто НЕ подтверждаются и оседают после ретраев). - dlqURL, dlqErr := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSDLQName) - if dlqErr != nil { - log.Warn("consumer: DLQ не найдена, битые сообщения будут только логироваться", "dlq", cfg.SQSDLQName, "err", dlqErr) - dlqURL = "" + // DLQ резолвим с ретраями: временная ошибка ResolveQueueURL НЕ должна навсегда + // отключить DLQ — иначе битые сообщения подтверждались бы и терялись без следа + // (ревью 2026-10-10, 🔴7). При окончательном провале dlqURL остаётся "" — + // битые сообщения тогда НЕ подтверждаются (см. обработку errInvalidEnvelope). + dlqURL := "" + for attempt := 1; attempt <= 3; attempt++ { + u, err := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSDLQName) + if err == nil { + dlqURL = u + break + } + log.Warn("consumer: DLQ resolve failed", "dlq", cfg.SQSDLQName, "attempt", attempt, "err", err) + if !sleepCtx(ctx, errorBackoff) { + break + } + } + if dlqURL == "" { + log.Warn("consumer: DLQ недоступна — битые сообщения НЕ будут подтверждаться (останутся в очереди)", "dlq", cfg.SQSDLQName) } log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName, "dlq", cfg.SQSDLQName) @@ -81,15 +93,19 @@ 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). if dlqURL != "" { - if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr != nil { - log.Error("consumer: send to DLQ failed", "err", derr, "message_id", aws.ToString(msg.MessageId)) + if derr := sendToDLQ(ctx, sqsClient, dlqURL, *msg.Body, log); derr == nil { + pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) + } else { + log.Error("consumer: send to DLQ failed — сообщение НЕ подтверждено, вернётся после visibility timeout", "err", derr, "message_id", aws.ToString(msg.MessageId)) } } else { - log.Warn("consumer: invalid envelope dropped (DLQ отсутствует)", "message_id", aws.ToString(msg.MessageId)) + log.Warn("consumer: DLQ недоступна — битое сообщение НЕ подтверждено, вернётся после visibility timeout", "message_id", aws.ToString(msg.MessageId)) } - // Подтверждаем: DLQ-отправка (или логирование) — терминальная обработка. - pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)}) continue } log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))