fix(P1.2-revue): батч-видимость не заводила в DLQ + TOCTOU лимит в SendMessageBatch
1. ChangeMessageVisibilityBatch при VisibilityTimeout=0 лишь инкрементировал Retry и НЕ переносил сообщение в DLQ (логика была только в одиночной версии). Вынесен единый метод Queue.ResetVisibilityAndTryDLQ — сброс видимости + DLQ-перенос. Обе версии привели к одному пути. 2. SendMessageBatch: проверка OOM-лимита была только по устаревшему снимку (RLock → Lock) — добавлена повторная проверка под per-queue Lock (TOCTOU). 3. Инварианты InFlightIndexes/ensureIndexes зафиксированы в док-комментариях. Тест: TestChangeVisibilityBatch_MovesToDLQ (падает на старой логике, зелёный на новой).
This commit is contained in:
Repinoid
@@ -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
|
||||
}
|
||||
|
||||
@@ -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++
|
||||
} else {
|
||||
msg.VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second)
|
||||
// Сброс + 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 {
|
||||
queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(entry.VisibilityTimeout) * time.Second)
|
||||
// Персистим изменённое сообщение отдельно
|
||||
persistence.SaveMessage(key, msg)
|
||||
persistence.SaveMessage(key, &queue.Messages[i])
|
||||
}
|
||||
messageFound = true
|
||||
}
|
||||
|
||||
|
||||
@@ -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}
|
||||
|
||||
+41
-2
@@ -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 {
|
||||
|
||||
Reference in New Issue
Block a user