fix(iot): не терять битые сообщения при недоступной DLQ (🔴7)
- ResolveQueueURL для DLQ теперь ретраится 3 раза с backoff: временная ошибка больше не отключает DLQ навсегда. - Битый envelope подтверждается только если DLQ реально приняла его; при недоступной DLQ/сбое sendToDLQ сообщение НЕ подтверждается (вернётся после visibility timeout), а не дропается.
This commit is contained in:
@@ -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))
|
||||
|
||||
Reference in New Issue
Block a user