10 KiB
P1.2 — индексация сообщений: дизайн и решение (2026-10-07)
Статус: ЗАКРЫВАЕМ отсрочку от 2026-10-06. Задача — сделать DeleteMessage и visibility-перерасчёт НЕ зависящими от полного размера очереди. Источник:
plan-for-flash-2026-10-05.md(P1.2),p1.1-per-queue-lock-analysis-2026-10-06.md.
1. Постановка
Queue.Messages []SqsMessage — единственное хранилище сообщений. Пути:
| Путь | Сейчас |
|---|---|
ReceiveMessage (receiveMessagesUnderLock) |
скан всего среза (нужен — ищем готовые) |
| DeleteMessage | скан среза по ReceiptHandle, затем сдвиг append(Messages[:i], Messages[i+1:]...) |
| DeleteMessageBatch | полная пересборка среза |
| ChangeMessageVisibility(±Batch) | скан по ReceiptHandle |
processQueueTick (видимость + DLQ) |
скан всего среза |
| Retention-эвикция | скан всего среза (компакция на месте) |
numberOfHiddenMessagesInQueue, admin/health/peek/metrics |
скан |
2. Почему «просто индекс receipt→index» недостаточно
Индекс поверх среза не даёт O(1) на удалении: физическое удаление из середины среза сдвигает все последующие элементы и ломает все сохранённые индексы (O(N) на переиндексацию). Именно это было причиной отсрочки 2026-10-06.
3. Решение: tombstone + индексы + амортизированная компакция
Ключевая идея: удаление не двигает элементы. Слот помечается «мёртвым»
(tombstone) за O(1); физическая чистка (компакция) выполняется редко —
раз в секунду в processQueueTick и по порогу при удалении.
Это сохраняет порядок сообщений (важно для FIFO и порядка выдачи), не меняет Redis-схему (сообщения и так лежат отдельными HASH-полями) и делает индексы стабильными между компакциями.
3.1. Новые поля models.Queue (все json:"-", в Redis не попадают)
Messages []SqsMessage // физическое хранилище; слот может быть tombstone
deadIdx []bool // параллельный Messages: true = слот удалён
receiptIdx map[string]int // ReceiptHandle → индекс (только in-flight)
uuidIdx map[string]int // Uuid → индекс (только живые)
liveCount int // число живых сообщений
deadCount int // число tombstone-слотов
Схема Redis не меняется: queueMetaDTO сообщения не содержит, а сами сообщения
хранятся отдельными полями ssq:msg:{key} → UUID. Tombstone-сообщения из Redis
удаляются сразу (как и сейчас, HDel синхронно), поэтому «воскрешения» после
рестарта не будет.
3.2. API Queue (примитивы, мутировать Messages напрямую запрещено)
AppendMessage(m SqsMessage) int— добавить (append) + проставить индексы;RemoveIndex(i int)— tombstone за O(1) + снять с индексов;IndexOfReceipt(rh string) (int, bool)/IndexOfUUID(u string) (int, bool);MarkInFlight(i int, rh string)/ClearInFlight(i int, rh string);IsDead(i int) bool,LiveCount() int,InFlightIndexes() []int;Compact()— физическое удаление tombstone + перестройка индексов (O(N));Reset()— очистка (purge);RebuildIndexes()— восстановление индексов (загрузка из Redis, ленивая защита).
Индексы строятся лениво (ensureIndexes), поэтому Queue, собранные литералом
в тестах/загрузчике, работают без изменений вызовов.
3.3. Что становится O(1) / O(in-flight)
- DeleteMessage →
IndexOfReceipt+RemoveIndex— O(1) (+ редкая компакция). - ChangeMessageVisibility(±Batch) →
IndexOfReceipt— O(1) на запись. - visibility-перерасчёт в тике → итерация ТОЛЬКО по
receiptIdx(in-flight), а не по всем сообщениям — O(in-flight). - DLQ-перенос в тике — внутри той же итерации по in-flight.
3.4. Что осознанно остаётся O(N)
- receive — нужно найти подходящие сообщения; полный O(1) требует ready-heap и в объём P1.2 не входит (критерий плана — delete + visibility).
- retention-эвикция — раз в секунду, скан (отдельный пункт P1.3, уже сделан).
Компакция порогом: при deadCount > 1024 && deadCount > liveCount — прямо в
RemoveIndex (амортизированно O(1) на удаление).
4. Затронутые файлы
app/models/models.go, app/models/helpers.go, app/gosqs/{receive_message, delete_message, delete_message_batch, change_message_visibility, change_message_visibility_batch, gosqs, send_message, send_message_batch, get_queue_attributes}.go, app/admin/admin.go, app/metrics/gauge_updater.go,
app/persistence/redis.go, app/gosqs/tenant_flow_test.go.
5. Критерий готовности
DeleteMessage/ChangeMessageVisibilityне сканируют срез.- visibility-перерасчёт в тике идёт по числу in-flight, не по размеру очереди.
- Все существующие тесты зелёные; добавлены тесты на tombstone/компакцию.
- Redis-схема не изменилась; рестарт не «воскрешает» удалённые сообщения.
6. РЕАЛИЗОВАНО (2026-10-07) — и найденный по ходу дефект
Коммиты: 22e9bc8 (код), e05af04 (тесты примитивов), 5b3fb51 (фикс ниже).
6.1. Найденный дефект (МОЯ РЕГРЕССИЯ) — утечка per-queue блокировки
При переводе receiveMessagesUnderLock на LiveCount() я сохранил исходный
порядок строк:
queue := models.SyncQueues.GetLocked(key) // взяли Lock
if queue == nil || queue.LiveCount() == 0 {
return nil, false // ← возврат БЕЗ Unlock
}
defer queue.UnlockMu()
Возврат на «пустой» очереди оставлял queue.mu захваченным НАВСЕГДА: любой
последующий Lock/RLock этой очереди ждал вечно (в Go мьютекс не
освобождается при выходе из функции).
Почему стало заметно именно сейчас: раньше условие было len(queue.Messages) == 0
и срабатывало только на полностью пустой очереди; после перехода на tombstone
добавился новый случай — «все сообщения удалены, но срез ещё не компактирован»
(LiveCount()==0 при len(Messages)>0), который в стресс-сценарии возникает
на каждом цикле.
Диагностика (без гаданий):
- симптом:
ReadTimeoutпри нагрузке 2000 сообщений, единичные всплески ~1.9 с наdelete_batch; - снят стек горутин по
SIGQUIT→ десятки горутин вsync.RWMutex.RLockнаQueue.RLockMu(models.go:101), включаяmetrics.updateGauges; HTTP-хендлерыGetQueueAttributesV1— там же. - аудит всех
GetLocked/GetRLockedпоказал: дефект ровно один — вreceive_message.go.
Фикс (5b3fb51): defer queue.UnlockMu() — ДО любой проверки/возврата.
Регрессионный тест TestReceiveEmptyQueue_DoesNotLeakLock (на дефекте падает
за 2 с, на фиксе — зелёный).
6.2. Результаты проверок
| Проверка | Результат |
|---|---|
go build ./..., go vet ./... |
OK |
go test -race ./app/... |
зелёные (admin, models, gosqs) |
tests/api_test.sh (aws CLI, локальный сервис) |
PASS=22 FAIL=0 |
tests/sdk_test.py (boto3) |
PASS=15 FAIL=0 |
tests/fifo_dlq_probe.py |
fifo_order/fifo_dedup/dlq = PASS |
| Стресс 3×2000 сообщений (send-batch → receive → delete-batch) | send ≈0.6 с, drain ≈1.2 с на цикл; left=0/0; ни одного вызова >0.5 с |
Стресс подтверждает и отсутствие «протечки» tombstone: после каждого цикла
очередь возвращается к 0 живых (CompactIfNeeded в тике работает).
6.3. Замечание
В первом прогоне api/sdk/fifo уже проходили (22/22, 15/15, PASS×3) — то есть дефект проявлялся не на каждом сценарии, а на последовательных циклах полного опустошения очереди. Поэтому стресс-проверка была обязательной.