Repinoid
c725676f85
Было по одному DeleteMessage на сообщение. Теперь DeleteMessageBatch (до 10 за вызов) с логированием Failed-записей; хелпер deleteMessage удалён.
217 lines
9.2 KiB
Go
217 lines
9.2 KiB
Go
// Package consumer — чтение shared-SQS → per-tenant PostgreSQL.
|
||
//
|
||
// Роль старого sqs-consumer внутри монолита: long-poll цикл, обработка
|
||
// envelope телеметрии, DeleteMessage после успешной записи (at-least-once).
|
||
package consumer
|
||
|
||
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"
|
||
"gitea.services.ngcloud.ru/Nail/IoT/internal/storage/iotpg"
|
||
)
|
||
|
||
// errorBackoff — пауза между попытками при ошибках SQS.
|
||
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)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
// 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)
|
||
|
||
for {
|
||
if ctx.Err() != nil {
|
||
break
|
||
}
|
||
|
||
resp, err := sqsClient.ReceiveMessage(ctx, &sqs.ReceiveMessageInput{
|
||
QueueUrl: aws.String(queueURL),
|
||
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 {
|
||
break
|
||
}
|
||
log.Error("consumer: ReceiveMessage failed", "err", err)
|
||
if !sleepCtx(ctx, errorBackoff) {
|
||
break
|
||
}
|
||
continue
|
||
}
|
||
|
||
processBackoff := time.Second
|
||
// Копим подтверждения, но НЕ удаляем из SQS сразу: InsertTelemetry лишь
|
||
// буферизует и возвращает nil до фактической записи. Удаляем только после
|
||
// успешного Flush — иначе краш/flush-ошибка = потеря подтверждённых
|
||
// сообщений (P0-7).
|
||
type pendingAck struct {
|
||
handle string
|
||
id string
|
||
}
|
||
var pending []pendingAck
|
||
for _, msg := range resp.Messages {
|
||
if msg.Body == nil || *msg.Body == "" {
|
||
log.Warn("consumer: empty message body", "message_id", aws.ToString(msg.MessageId))
|
||
pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
|
||
continue
|
||
}
|
||
if err := processTelemetry(ctx, *msg.Body, store, log); err != nil {
|
||
// Битый envelope → DLQ (а не бесконечный ретрай и не потеря):
|
||
// кладём тело в DLQ, затем подтверждаем основное сообщение.
|
||
if errors.Is(err, errInvalidEnvelope) {
|
||
// Битый 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", "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: invalid envelope deferred (DLQ недоступна, будет повтор)",
|
||
"message_id", aws.ToString(msg.MessageId), "receive_count", recv)
|
||
}
|
||
continue
|
||
}
|
||
log.Error("consumer: process telemetry", "err", err, "message_id", aws.ToString(msg.MessageId))
|
||
// Сообщение НЕ подтверждаем — вернётся после visibility timeout.
|
||
// Backoff с jitter — не долбить упавший PG в цикле
|
||
// (фикс HIGH из ревью 2026-08-16: лавина ретраев).
|
||
if !sleepCtx(ctx, processBackoff) {
|
||
break
|
||
}
|
||
processBackoff *= 2
|
||
if processBackoff > maxProcessBackoff {
|
||
processBackoff = maxProcessBackoff
|
||
}
|
||
continue
|
||
}
|
||
processBackoff = time.Second
|
||
pending = append(pending, pendingAck{aws.ToString(msg.ReceiptHandle), aws.ToString(msg.MessageId)})
|
||
}
|
||
|
||
if len(pending) == 0 {
|
||
continue
|
||
}
|
||
// Сначала гарантируем запись в БД, только потом удаляем из SQS.
|
||
if err := store.Flush(ctx); err != nil {
|
||
log.Error("consumer: flush failed — сообщения НЕ удалены, вернутся после visibility timeout", "err", err)
|
||
continue
|
||
}
|
||
// DeleteMessageBatch: до 10 записей за вызов (было по одному DeleteMessage
|
||
// на сообщение — лишние round-trip, ревью 2026-10-10).
|
||
for i := 0; i < len(pending); i += 10 {
|
||
end := i + 10
|
||
if end > len(pending) {
|
||
end = len(pending)
|
||
}
|
||
chunk := pending[i:end]
|
||
entries := make([]sqstypes.DeleteMessageBatchRequestEntry, 0, len(chunk))
|
||
for j := range chunk {
|
||
entries = append(entries, sqstypes.DeleteMessageBatchRequestEntry{
|
||
Id: aws.String(strconv.Itoa(j)),
|
||
ReceiptHandle: aws.String(chunk[j].handle),
|
||
})
|
||
}
|
||
out, err := sqsClient.DeleteMessageBatch(ctx, &sqs.DeleteMessageBatchInput{
|
||
QueueUrl: aws.String(queueURL),
|
||
Entries: entries,
|
||
})
|
||
if err != nil {
|
||
log.Error("consumer: DeleteMessageBatch failed", "err", err, "count", len(chunk))
|
||
continue
|
||
}
|
||
for _, f := range out.Failed {
|
||
log.Error("consumer: DeleteMessageBatch entry failed",
|
||
"id", aws.ToString(f.Id), "code", aws.ToString(f.Code), "msg", aws.ToString(f.Message))
|
||
}
|
||
}
|
||
}
|
||
|
||
log.Info("consumer: shutting down")
|
||
return nil
|
||
}
|
||
|
||
// 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
|
||
}
|
||
|
||
// 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 {
|
||
case <-ctx.Done():
|
||
return false
|
||
case <-time.After(d):
|
||
return true
|
||
}
|
||
}
|