From c725676f856a3cc253ba402c5dad48bb85abb4f1 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Sat, 10 Oct 2026 19:15:08 +0300 Subject: [PATCH] =?UTF-8?q?chore(iot):=20=D1=83=D0=B4=D0=B0=D0=BB=D0=B5?= =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20=D1=81=D0=BE=D0=BE=D0=B1=D1=89=D0=B5=D0=BD?= =?UTF-8?q?=D0=B8=D0=B9=20=D0=BF=D0=B0=D0=BA=D0=B5=D1=82=D0=B0=D0=BC=D0=B8?= =?UTF-8?q?=20DeleteMessageBatch=20(=D0=AD=D1=82=D0=B0=D0=BF=205)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Было по одному DeleteMessage на сообщение. Теперь DeleteMessageBatch (до 10 за вызов) с логированием Failed-записей; хелпер deleteMessage удалён. --- internal/service/consumer/consumer.go | 38 ++++++++++++++++++--------- 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/internal/service/consumer/consumer.go b/internal/service/consumer/consumer.go index 0c3c5a8..de8f0f2 100644 --- a/internal/service/consumer/consumer.go +++ b/internal/service/consumer/consumer.go @@ -155,9 +155,32 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * log.Error("consumer: flush failed — сообщения НЕ удалены, вернутся после visibility timeout", "err", err) continue } - for _, p := range pending { - if err := deleteMessage(ctx, sqsClient, queueURL, p.handle, log); err != nil { - log.Error("consumer: DeleteMessage failed", "err", err, "message_id", p.id) + // 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)) } } } @@ -166,15 +189,6 @@ func Run(ctx context.Context, cfg *config.Config, sqsClient *sqs.Client, store * return nil } -// deleteMessage удаляет обработанное сообщение из очереди. -func deleteMessage(ctx context.Context, client *sqs.Client, queueURL string, receiptHandle string, log *slog.Logger) error { - _, err := client.DeleteMessage(ctx, &sqs.DeleteMessageInput{ - QueueUrl: aws.String(queueURL), - ReceiptHandle: aws.String(receiptHandle), - }) - 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{