diff --git a/app/gosqs/get_queue_attributes.go b/app/gosqs/get_queue_attributes.go index 8e5c68e..9faeb84 100644 --- a/app/gosqs/get_queue_attributes.go +++ b/app/gosqs/get_queue_attributes.go @@ -91,7 +91,7 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessages", Value: strconv.Itoa(len(queue.Messages))}) } if shouldInclude("ApproximateNumberOfMessagesNotVisible") { - queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(*queue))}) + queueAttributes = append(queueAttributes, models.Attribute{Name: "ApproximateNumberOfMessagesNotVisible", Value: strconv.Itoa(numberOfHiddenMessagesInQueue(queue))}) } if shouldInclude("CreatedTimestamp") { queueAttributes = append(queueAttributes, models.Attribute{Name: "CreatedTimestamp", Value: "0000000000"}) diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index b13eb3f..5099d5b 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -136,7 +136,7 @@ func processQueueTick(key string, queue *models.Queue) { } } -func numberOfHiddenMessagesInQueue(queue models.Queue) int { +func numberOfHiddenMessagesInQueue(queue *models.Queue) int { num := 0 for _, m := range queue.Messages { if m.ReceiptHandle != "" || m.DelaySecs > 0 && time.Now().Before(m.SentTime.Add(time.Duration(m.DelaySecs)*time.Second)) { diff --git a/app/models/models.go b/app/models/models.go index 332e172..16ddf31 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -2,6 +2,7 @@ package models import ( "strconv" + "sync" "time" log "github.com/sirupsen/logrus" @@ -46,6 +47,11 @@ func (m *SqsMessage) IsReadyForReceipt() bool { } type Queue struct { + // mu — per-queue блокировка содержимого очереди (Messages, Duplicates, FIFO- + // состояние, LastPurgeTime). Заменяет удержание глобального SyncQueues.Lock + // на время операций с одной очередью. sync.RWMutex сериализации не подлежит. + mu sync.RWMutex `json:"-"` + Name string URL string Arn string @@ -53,7 +59,7 @@ type Queue struct { ReceiveMessageWaitTimeSeconds int DelaySeconds int MaximumMessageSize int - MessageRetentionPeriod int // seconds // TODO - not used in the code yet + MessageRetentionPeriod int // seconds (реализовано в processQueueTick) Messages []SqsMessage DeadLetterQueue *Queue MaxReceiveCount int @@ -69,6 +75,13 @@ type Queue struct { LastPurgeTime time.Time } +// LockMu/UnlockMu/RLockMu/RUnlockMu — доступ к per-queue блокировке извне пакета models +// (поле mu неэкспортируемо). +func (q *Queue) LockMu() { q.mu.Lock() } +func (q *Queue) UnlockMu() { q.mu.Unlock() } +func (q *Queue) RLockMu() { q.mu.RLock() } +func (q *Queue) RUnlockMu() { q.mu.RUnlock() } + // NextSequenceNumber — возвращает следующий порядковый номер сообщения FIFO-группы. // // Логическая схема: каждый вызов инкрементирует счётчик группы и возвращает diff --git a/doc/thinking/p1.1-per-queue-lock-analysis-2026-10-06.md b/doc/thinking/p1.1-per-queue-lock-analysis-2026-10-06.md new file mode 100644 index 0000000..10bc254 --- /dev/null +++ b/doc/thinking/p1.1-per-queue-lock-analysis-2026-10-06.md @@ -0,0 +1,54 @@ +# P1.1 (per-queue mutex) — анализ и состояние (2026-10-06) + +## Зачем нужен + +Сейчас все операции с очередями держат ОДИН глобальный `models.SyncQueues.RWMutex`. +Проблемы: +1. Любая операция блокирует весь сервис. При этом часть операций делает + **синхронный Redis-I/O прямо под глобальным локом** (таймаут до 3–5с): + `SaveMessageSync` (send), `DeleteMessagePersist` (delete), `DeleteMessagesPersist` + (retention-tick), `PurgeMessagesPersist` (purge), `DeleteQueue`. При деградации + Redis весь сервис встаёт на секунды. +2. `receiveMessagesUnderLock` держит глобальный лок на время O(N)-скана сообщений. + +## Введённый groundwork (безопасно, поведение не меняется) + +- `models.Queue` получил поле `mu sync.RWMutex` (не сериализуется) + методы + `LockMu/UnlockMu/RLockMu/RUnlockMu`. +- Исправлено копирование `Queue` by value (`numberOfHiddenMessagesInQueue`) — + иначе `go vet` ругался на copylocks. +- Глобальный лок пока используется везде (как раньше) — **поведение корректно**. + +## Блокеры полного перехода (почему это не механическая замена) + +1. **DLQ — общий указатель.** `Queue.DeadLetterQueue *Queue`. Перенос сообщения в + DLQ мутирует `DeadLetterQueue.Messages` из-под лока ИСХОДНОЙ очереди + (`gosqs.go:130`, `change_message_visibility.go:76`). При per-queue локах нужен + порядок взятия (сначала исходная, потом DLQ) + гарантия отсутствия цикла. +2. **`marshalQueueMeta` мутирует ЧУЖУЮ очередь.** `persistence/redis.go:146-160` + временно зануляет `queue.Messages` И `queue.DeadLetterQueue.Messages` ради + сериализации. Под глобальным локом это безопасно; под per-queue — гонка с + операциями над DLQ. Нужен НЕмутирующий снапшот (DTO без `Messages`), + но нельзя копировать `Queue` целиком (копируется мьютекс) → нужен отдельный + тип-снапшот. +3. **DLQ-указатель после загрузки из Redis не релинкуется** на объект из map + (загружается как вложенная копия) — предсуществующая проблема, которую + P1.1 обнажит (переносы в DLQ пойдут в detached-копию). +4. **Объём — атомарный.** Переводить на per-queue локи надо СРАЗУ все операции + с содержимым очереди (send/receive/delete/visibility/purge/attributes/tags/ + admin/gauge-tick) — частичный перевод создаёт рассинхрон (операции перестанут + взаимоисключаться). Это ~20+ мест в 8 файлах. +5. **Нет покрытия** для FIFO/batch/DLQ/visibility — самых сложных путей. + +## Вывод + +P1.1 — это дизайнерская задача (порядок локов + неммутирующий снапшот + релинк +DLQ), а не построчная замена. Делать её «вслепую» без покрытия FIFO/batch/DLQ +рискованно: ошибка = дедлоки/потеря данных на проде. Рекомендуется отдельный +проход: сначала тесты на FIFO/DLQ/batch, затем атомарный перевод. + +## P1.2 (индексация сообщений) — та же природа + +Индексация `Messages` (receipt→index) меняет структуру хранения, затронутую теми +же путями (receive/delete/visibility/tick/сериализация), и требует покрытия тех +же путей. Рекомендуется вместе с P1.1.