diff --git a/HISTORY/2026-10-07-session-log.md b/HISTORY/2026-10-07-session-log.md index b5e764f..1ad026b 100644 --- a/HISTORY/2026-10-07-session-log.md +++ b/HISTORY/2026-10-07-session-log.md @@ -109,3 +109,25 @@ Redis-схема не меняется (сообщения и так лежат весь блок, включая порядок `defer` — старый код содержал латентную утечку лока, которую новый критерий сделал массовой. 4. Не запускать сервис с `| head` — SIGPIPE убивает процесс на середине теста. + +## 7. Код-ревью (роль ПРО) и фиксы — коммит `48c38c4` + +Проведён архитектурный разбор всех затронутых файлов. Найдено и исправлено: + +1. **🔥 Батч-видимость не заводила в DLQ** (`change_message_visibility_batch.go`). + При `VisibilityTimeout=0` батч лишь инкрементировал `Retry` и НЕ переносил + сообщение в DLQ — логика была только в одиночной версии, счётчик и перенос + расходились. Вынесен единый метод `models.Queue.ResetVisibilityAndTryDLQ` + (сброс видимости + DLQ-перенос), обе версии приведены к одному пути. + Тест `TestChangeVisibilityBatch_MovesToDLQ` (падает на старой логике, зелёный на новой). +2. **TOCTOU лимит в `SendMessageBatch`**: проверка OOM-лимита была только по + устаревшему снимку (RLock → Lock). Добавлена повторная проверка под per-queue Lock. +3. Инварианты `InFlightIndexes` (снимок может содержать устаревшие индексы — + потребитель обязан перепроверять) и `ensureIndexes` (проверяет согласованность + чисел, а не валидность map-индексов → `Messages` нельзя мутировать вне методов) + зафиксированы в док-комментариях. +4. `Compact` упрощён: один `RebuildIndexes` на выходе + `ensureIndexes` на входе + (убран дублирующий блок), `deadCount` пересчитывается внутри. + +Осталось осознанно не тронуто: `DeleteMessageBatch` итерация с `RemoveIndex` внутри +безопасна (tombstone не меняет длину) — покрыто тестами. diff --git a/app/models/models.go b/app/models/models.go index a71876e..27d5370 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -298,15 +298,12 @@ func (q *Queue) CompactIfNeeded() { // Compact — физически удаляет tombstone-слоты, сохраняя порядок живых, и // перестраивает индексы. O(N). Вызывать под write-блокировкой. func (q *Queue) Compact() { + // ensureIndexes гарантирует len(deadIdx)==len(Messages) (иначе слоты разъедутся) + // и корректность deadCount. При согласованном состоянии — две дешёвые проверки. + q.ensureIndexes() if q.deadCount == 0 { return } - if len(q.deadIdx) != len(q.Messages) { - q.RebuildIndexes() - if q.deadCount == 0 { - return - } - } n := 0 for i := range q.Messages { if q.deadIdx[i] { @@ -323,10 +320,9 @@ func (q *Queue) Compact() { tail[i] = SqsMessage{} } q.Messages = q.Messages[:n] - // После чистки tombstone нет — сбрасываем флаги, чтобы RebuildIndexes - // построил индексы только по живым (иначе старые deadIdx-флаги «съедут»). + // После чистки tombstone нет: обнуляем флаги и перестраиваем индексы по живым + // (RebuildIndexes заново посчитает liveCount/deadCount и заполнит receipt/uuid). q.deadIdx = q.deadIdx[:0] - q.deadCount = 0 q.RebuildIndexes() }