feat(consumer): битые envelope → DLQ вместо потери
processTelemetry теперь возвращает sentinel errInvalidEnvelope при битом JSON (раньше — молчаливый nil → mусор подтверждался и терялся). Consumer при этой ошибке кладёт тело в DLQ (SQS_DLQ_NAME, дефолт iot-telemetry-dlq) и подтверждает основное сообщение; без DLQ — только логирует, но НЕ подтверждает (оседает). Добавлены SQSDLQName в config + DefaultSQSDLQName. Реальные ошибки PG (не invalid) по-прежнему не подтверждаются и уходят в ретрай после visibility timeout.
This commit is contained in:
@@ -6,6 +6,7 @@ package consumer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"log/slog"
|
||||
"time"
|
||||
|
||||
@@ -29,7 +30,14 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName)
|
||||
// DLQ присутствует не всегда: резолвим лениво, отказ не фатален (битые
|
||||
// сообщения тогда просто НЕ подтверждаются и оседают после ретраев).
|
||||
dlqURL, dlqErr := sqsclient.ResolveQueueURL(ctx, sqsClient, cfg.SQSDLQName)
|
||||
if dlqErr != nil {
|
||||
log.Warn("consumer: DLQ не найдена, битые сообщения будут только логироваться", "dlq", cfg.SQSDLQName, "err", dlqErr)
|
||||
dlqURL = ""
|
||||
}
|
||||
log.Info("consumer: SQS queue resolved", "queue", cfg.SQSQueueName, "dlq", cfg.SQSDLQName)
|
||||
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
@@ -70,6 +78,20 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store *
|
||||
continue
|
||||
}
|
||||
if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
|
||||
// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
|
||||
// кладём тело в DLQ, затем подтверждаем основное сообщение.
|
||||
if errors.Is(err, errInvalidEnvelope) {
|
||||
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))
|
||||
}
|
||||
} else {
|
||||
log.Warn("consumer: invalid envelope dropped (DLQ отсутствует)", "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))
|
||||
// Сообщение НЕ подтверждаем — вернётся после visibility timeout.
|
||||
// Backoff с jitter — не долбить упавший PG в цикле
|
||||
@@ -115,6 +137,15 @@ func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, rec
|
||||
return err
|
||||
}
|
||||
|
||||
// sendToDLQ — кладёт битое неразборчивое сообщение в DLQ для диагностики.
|
||||
func sendToDLQ(ctx context.Context, client *sqs.Client, dlqURL, body string, log *slog.Logger) error {
|
||||
_, err := client.SendMessage(ctx, &sqs.SendMessageInput{
|
||||
QueueUrl: aws.String(dlqURL),
|
||||
MessageBody: aws.String(body),
|
||||
})
|
||||
return err
|
||||
}
|
||||
|
||||
// sleepCtx — ожидание с прерыванием по ctx. false = ctx завершён.
|
||||
func sleepCtx(ctx context.Context, d time.Duration) bool {
|
||||
select {
|
||||
|
||||
Reference in New Issue
Block a user