155 lines
10 KiB
Markdown
155 lines
10 KiB
Markdown
# 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 не попадают)
|
||
|
||
```go
|
||
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. Критерий готовности
|
||
|
||
1. `DeleteMessage`/`ChangeMessageVisibility` не сканируют срез.
|
||
2. visibility-перерасчёт в тике идёт по числу in-flight, не по размеру очереди.
|
||
3. Все существующие тесты зелёные; добавлены тесты на tombstone/компакцию.
|
||
4. Redis-схема не изменилась; рестарт не «воскрешает» удалённые сообщения.
|
||
|
||
## 6. РЕАЛИЗОВАНО (2026-10-07) — и найденный по ходу дефект
|
||
|
||
Коммиты: `22e9bc8` (код), `e05af04` (тесты примитивов), `5b3fb51` (фикс ниже).
|
||
|
||
### 6.1. Найденный дефект (МОЯ РЕГРЕССИЯ) — утечка per-queue блокировки
|
||
|
||
При переводе `receiveMessagesUnderLock` на `LiveCount()` я сохранил исходный
|
||
порядок строк:
|
||
|
||
```go
|
||
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) — то есть
|
||
дефект проявлялся не на каждом сценарии, а на последовательных циклах полного
|
||
опустошения очереди. Поэтому стресс-проверка была обязательной.
|