From 2e1c594892b468ae2973e3ba2889dc58c0113c77 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 7 Oct 2026 08:35:52 +0300 Subject: [PATCH] =?UTF-8?q?fix(P1.2-revue):=20=D0=B1=D0=B0=D1=82=D1=87-?= =?UTF-8?q?=D0=B2=D0=B8=D0=B4=D0=B8=D0=BC=D0=BE=D1=81=D1=82=D1=8C=20=D0=BD?= =?UTF-8?q?=D0=B5=20=D0=B7=D0=B0=D0=B2=D0=BE=D0=B4=D0=B8=D0=BB=D0=B0=20?= =?UTF-8?q?=D0=B2=20DLQ=20+=20TOCTOU=20=D0=BB=D0=B8=D0=BC=D0=B8=D1=82=20?= =?UTF-8?q?=D0=B2=20SendMessageBatch?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 1. ChangeMessageVisibilityBatch при VisibilityTimeout=0 лишь инкрементировал Retry и НЕ переносил сообщение в DLQ (логика была только в одиночной версии). Вынесен единый метод Queue.ResetVisibilityAndTryDLQ — сброс видимости + DLQ-перенос. Обе версии привели к одному пути. 2. SendMessageBatch: проверка OOM-лимита была только по устаревшему снимку (RLock → Lock) — добавлена повторная проверка под per-queue Lock (TOCTOU). 3. Инварианты InFlightIndexes/ensureIndexes зафиксированы в док-комментариях. Тест: TestChangeVisibilityBatch_MovesToDLQ (падает на старой логике, зелёный на новой). --- app/gosqs/change_message_visibility.go | 27 +++--------- app/gosqs/change_message_visibility_batch.go | 23 ++++++----- app/gosqs/send_message_batch.go | 8 ++++ app/models/models.go | 43 +++++++++++++++++++- 4 files changed, 67 insertions(+), 34 deletions(-) diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index e46ccb8..d466889 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -58,30 +58,13 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo var msgRemoved bool // O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза. if i, ok := queue.IndexOfReceipt(receiptHandle); ok { - msg := &queue.Messages[i] - timeout := queue.VisibilityTimeout if visibilityTimeout == 0 { - // Возврат видимости: сброс handle + снятие с индекса in-flight. - queue.ClearInFlight(i, receiptHandle) - msg.ReceiptTime = time.Now().UTC() - msg.ReceiptHandle = "" - msg.VisibilityTimeout = time.Now().Add(time.Duration(timeout) * time.Second) - msg.Retry++ - changedMsgUuid = msg.Uuid - if queue.MaxReceiveCount > 0 && - queue.DeadLetterQueue != nil && - msg.Retry >= queue.MaxReceiveCount { - // Перенос в DLQ + удаление из исходной очереди (tombstone, O(1)). - dlqMsg := *msg - queue.DeadLetterQueue.LockMu() - queue.DeadLetterQueue.AppendMessage(dlqMsg) - queue.DeadLetterQueue.UnlockMu() - queue.RemoveIndex(i) - msgRemoved = true - } + // Возврат видимости (+ DLQ-перенос при исчерпании MaxReceiveCount) — + // единый путь с батчевой версией. + changedMsgUuid, msgRemoved = queue.ResetVisibilityAndTryDLQ(i, receiptHandle) } else { - msg.VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second) - changedMsgUuid = msg.Uuid + queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second) + changedMsgUuid = queue.Messages[i].Uuid } messageFound = true } diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index c4ef84f..66ad0ff 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -82,19 +82,22 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract messageFound := false // O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза. if i, ok := queue.IndexOfReceipt(entry.ReceiptHandle); ok { - msg := &queue.Messages[i] if entry.VisibilityTimeout == 0 { - // Сброс: сообщение снова видимо, счётчик retry++ - queue.ClearInFlight(i, entry.ReceiptHandle) - msg.ReceiptTime = time.Now().UTC() - msg.ReceiptHandle = "" - msg.VisibilityTimeout = time.Now().Add(time.Duration(queue.VisibilityTimeout) * time.Second) - msg.Retry++ + // Сброс + DLQ-перенос при исчерпании MaxReceiveCount — единый путь + // с одиночной версией (раньше батч лишь инкрементировал Retry и не + // переносил в DLQ — счётчик и перенос могли расходиться). + uuid, moved := queue.ResetVisibilityAndTryDLQ(i, entry.ReceiptHandle) + if moved { + // Сообщение перенесено в DLQ и удалено из очереди — чистим Redis. + persistence.DeleteMessagePersist(key, uuid) + } else if j, ok := queue.IndexOfUUID(uuid); ok { + persistence.SaveMessage(key, &queue.Messages[j]) + } } else { - msg.VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second) + queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second) + // Персистим изменённое сообщение отдельно + persistence.SaveMessage(key, &queue.Messages[i]) } - // Персистим изменённое сообщение отдельно - persistence.SaveMessage(key, msg) messageFound = true } diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 1bd8ef3..a635845 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -104,6 +104,14 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody queue.LockMu() defer queue.UnlockMu() + + // Повторная проверка лимита под per-queue Lock: снимок currentMsgCount выше + // был сделан под RLock, между ним и взятием Lock очередь могла измениться + // (гонка TOCTOU). Здесь — обязательная проверка по свежему числу живых. + if queue.LiveCount()+len(sendEntries) > MaxMessagesForQueue(queue.IsFIFO) { + return utils.CreateErrorResponseV1("OverLimit", true) + } + newMsgs := make([]models.SqsMessage, 0, len(sendEntries)) for _, sendEntry := range sendEntries { msg := models.SqsMessage{MessageBody: sendEntry.MessageBody} diff --git a/app/models/models.go b/app/models/models.go index 33fd4fa..a71876e 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -191,8 +191,12 @@ func (q *Queue) InitDuplicatation(deduplicationId string) { const compactDeadThreshold = 1024 // ensureIndexes — ленивая инициализация/починка индексов. -// При корректном состоянии — две дешёвые проверки; O(N) — если Messages -// изменили в обход методов (загрузка из Redis, тесты, старый JSON-формат). +// При корректном состоянии — две дешёвые проверки; O(N) — только если состояние +// рассинхронизировалось (прямая замена Messages вне методов, загрузка из Redis, +// старый JSON-формат). ВАЖНО (инвариант): оценка «в порядке» проверяет лишь +// согласованность чисел и не перестраивает уже корректные map-индексы — поэтому +// мутировать Messages ВНЕ методов (AppendMessage/RemoveIndex/Compact/Reset), +// например менять ReceiptHandle на месте, НЕЛЬЗЯ. // Вызывать под write-блокировкой. func (q *Queue) ensureIndexes() { if len(q.deadIdx) == len(q.Messages) && @@ -386,8 +390,43 @@ func (q *Queue) ClearInFlight(i int, receiptHandle string) { } } +// ResetVisibilityAndTryDLQ — сбрасывает видимость сообщения по индексу и, если +// исчерпан MaxReceiveCount при настроенной DLQ, переносит его в DLQ и удаляет +// из исходной очереди (tombstone). Возвращает false, если сообщение на самом +// деле перенесено в DLQ (удалено из очереди). +// +// Единый путь для одиночной и батчевой ChangeMessageVisibility: раньше логика +// DLQ была только в одиночной версии, а батч лишь инкрементировал Retry — из-за +// этого retry-счётчик и DLQ-перенос могли расходиться. +// Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой. +// Возвращает: msgUuid, movedToDLQ. +func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid string, movedToDLQ bool) { + q.ensureIndexes() + q.ClearInFlight(i, receiptHandle) + msg := &q.Messages[i] + msg.ReceiptTime = time.Now().UTC() + msg.ReceiptHandle = "" + msg.VisibilityTimeout = time.Now().Add(time.Duration(q.VisibilityTimeout) * time.Second) + msg.Retry++ + msgUuid = msg.Uuid + if q.MaxReceiveCount > 0 && q.DeadLetterQueue != nil && msg.Retry >= q.MaxReceiveCount { + dlqMsg := *msg + q.DeadLetterQueue.LockMu() + q.DeadLetterQueue.AppendMessage(dlqMsg) + q.DeadLetterQueue.UnlockMu() + q.RemoveIndex(i) + return msgUuid, true + } + return msgUuid, false +} + // InFlightIndexes — снимок индексов in-flight сообщений (для visibility-тика). // Возвращает срез индексов в Messages. Вызывать под write-блокировкой. +// +// ВАЖНО: индексы — снимок на момент вызова. Некоторые могут стать неактуальными +// (handle переехал/сообщение удалено), поэтому потребитель ОБЯЗАН перепроверять +// каждое сообщение по индексу (живость: в границах среза, не tombstone, handle +// не пуст) — как это делает processQueueTick. func (q *Queue) InFlightIndexes() []int { idx := make([]int, 0, len(q.receiptIdx)) for _, i := range q.receiptIdx {