From fd8ffc329d240d1f25ffec0c4dae865e783e4c1b Mon Sep 17 00:00:00 2001 From: Repinoid Date: Tue, 6 Oct 2026 18:58:16 +0300 Subject: [PATCH] =?UTF-8?q?P1.1:=20per-queue=20=D0=B1=D0=BB=D0=BE=D0=BA?= =?UTF-8?q?=D0=B8=D1=80=D0=BE=D0=B2=D0=BA=D0=B8=20=D0=B2=D0=BC=D0=B5=D1=81?= =?UTF-8?q?=D1=82=D0=BE=20=D0=B3=D0=BB=D0=BE=D0=B1=D0=B0=D0=BB=D1=8C=D0=BD?= =?UTF-8?q?=D0=BE=D0=B3=D0=BE=20=D0=BB=D0=BE=D0=BA=D0=B0=20=D0=BD=D0=B0=20?= =?UTF-8?q?=D0=BE=D0=BF=D0=B5=D1=80=D0=B0=D1=86=D0=B8=D0=B8=20=D1=81=20?= =?UTF-8?q?=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4=D1=8C=D1=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Все операции с содержимым очереди (send/receive/delete/visibility/purge/attributes/ tags/admin/gauge/tick) переведены на per-queue RWMutex (QueueStore.GetLocked/ GetRLocked + queue.LockMu). Глобальный SyncQueues-лок остался только для map и неизменяемых полей. DLQ-переносы берут лок DLQ отдельно (порядок main→DLQ). marshalQueueMeta больше не мутирует очередь (DTO-снапшот). Устранён дедлок create_queue (setQueueAttributesV1 вызывается вне глобального лока). Добавлены конкурентные тесты; go test -race зелёный. Снимает боттлнек: синхронный Redis-I/O больше не держит глобальный лок. --- app/admin/admin.go | 64 ++++++++++------ app/gosqs/change_message_visibility.go | 15 ++-- app/gosqs/change_message_visibility_batch.go | 10 +-- app/gosqs/create_queue.go | 39 +++++----- app/gosqs/delete_message.go | 45 +++++------ app/gosqs/delete_message_batch.go | 10 +-- app/gosqs/get_queue_attributes.go | 7 +- app/gosqs/gosqs.go | 8 +- app/gosqs/list_queue_tags.go | 8 +- app/gosqs/purge_queue.go | 18 ++--- app/gosqs/queue_attributes.go | 2 + app/gosqs/receive_message.go | 16 ++-- app/gosqs/send_message.go | 23 ++---- app/gosqs/send_message_batch.go | 19 ++--- app/gosqs/set_queue_attributes.go | 7 +- app/gosqs/tag_queue.go | 8 +- app/gosqs/untag_queue.go | 8 +- app/metrics/gauge_updater.go | 13 +++- app/models/globals.go | 25 +++++++ app/persistence/redis.go | 74 +++++++++++++++---- ...p1.1-per-queue-lock-analysis-2026-10-06.md | 14 ++++ 21 files changed, 252 insertions(+), 181 deletions(-) diff --git a/app/admin/admin.go b/app/admin/admin.go index 0ac0df2..153ded3 100644 --- a/app/admin/admin.go +++ b/app/admin/admin.go @@ -445,15 +445,23 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) { RetentionPeriod int `json:"retention_period"` } queues := make([]queueInfo, 0) + type nameQueue struct { + name string + q *models.Queue + } models.SyncQueues.RLock() + pairs := make([]nameQueue, 0) for _, name := range models.SyncQueues.TenantQueueNames(t.AccessKey) { - q := models.SyncQueues.Queues[prefix+name] - if q == nil { - continue + if q := models.SyncQueues.Queues[prefix+name]; q != nil { + pairs = append(pairs, nameQueue{name, q}) } + } + models.SyncQueues.RUnlock() + for _, p := range pairs { + p.q.RLockMu() visible := 0 notVisible := 0 - for _, msg := range q.Messages { + for _, msg := range p.q.Messages { if msg.ReceiptHandle != "" { notVisible++ } else { @@ -461,15 +469,15 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) { } } queues = append(queues, queueInfo{ - Name: name, + Name: p.name, Messages: visible, NotVisible: notVisible, - VisibilityTimeout: q.VisibilityTimeout, - MaxMessageSize: q.MaximumMessageSize, - RetentionPeriod: q.MessageRetentionPeriod, + VisibilityTimeout: p.q.VisibilityTimeout, + MaxMessageSize: p.q.MaximumMessageSize, + RetentionPeriod: p.q.MessageRetentionPeriod, }) + p.q.RUnlockMu() } - models.SyncQueues.RUnlock() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(queues) } @@ -594,7 +602,7 @@ func (h *Handler) peekQueueMessages(w http.ResponseWriter, r *http.Request) { } } } - models.SyncQueues.RLock() + q.RLockMu() result := make([]peekMessageItem, 0, len(q.Messages)) for i, msg := range q.Messages { if i >= limit { @@ -609,7 +617,7 @@ func (h *Handler) peekQueueMessages(w http.ResponseWriter, r *http.Request) { InFlight: msg.ReceiptHandle != "", }) } - models.SyncQueues.RUnlock() + q.RUnlockMu() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(result) } @@ -645,12 +653,12 @@ func (h *Handler) sendMessageToQueue(w http.ResponseWriter, r *http.Request) { Uuid: uuid.NewString(), SentTime: time.Now(), } - models.SyncQueues.Lock() - models.SyncQueues.Queues[key].Messages = append(models.SyncQueues.Queues[key].Messages, msg) + q.LockMu() + q.Messages = append(q.Messages, msg) // Персистим одно сообщение отдельно persistence.SaveMessage(key, &msg) - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) - models.SyncQueues.Unlock() + persistence.SaveQueue(key, q) + q.UnlockMu() w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusCreated) json.NewEncoder(w).Encode(map[string]string{"id": msg.Uuid, "status": "sent"}) @@ -671,12 +679,12 @@ func (h *Handler) purgeQueue(w http.ResponseWriter, r *http.Request) { jsonErr(w, http.StatusNotFound, "queue not found") return } - models.SyncQueues.Lock() - models.SyncQueues.Queues[key].Messages = models.SyncQueues.Queues[key].Messages[:0] + q.LockMu() + q.Messages = q.Messages[:0] // Удаляем все сообщения из Redis одной командой persistence.PurgeMessagesPersist(key) - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) - models.SyncQueues.Unlock() + persistence.SaveQueue(key, q) + q.UnlockMu() log.Infof("admin: purged queue %s for tenant %s", queueName, t.ID) w.WriteHeader(http.StatusNoContent) } @@ -704,13 +712,19 @@ func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) { queueCount := 0 msgCount := 0 models.SyncQueues.RLock() + uiQueues := make([]*models.Queue, 0) for key, q := range models.SyncQueues.Queues { if strings.HasPrefix(key, prefix) { queueCount++ - msgCount += len(q.Messages) + uiQueues = append(uiQueues, q) } } models.SyncQueues.RUnlock() + for _, q := range uiQueues { + q.RLockMu() + msgCount += len(q.Messages) + q.RUnlockMu() + } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(adminHealthDetail{ @@ -725,11 +739,17 @@ func (h *Handler) detailedHealth(w http.ResponseWriter, r *http.Request) { tenants := h.store.List() models.SyncQueues.RLock() queueCount := len(models.SyncQueues.Queues) - msgCount := 0 + allQueues := make([]*models.Queue, 0, queueCount) for _, q := range models.SyncQueues.Queues { - msgCount += len(q.Messages) + allQueues = append(allQueues, q) } models.SyncQueues.RUnlock() + msgCount := 0 + for _, q := range allQueues { + q.RLockMu() + msgCount += len(q.Messages) + q.RUnlockMu() + } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(adminHealthDetail{ diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index 9836875..4aca164 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -47,18 +47,11 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo return utils.CreateErrorResponseV1("InvalidParameterValue", true) } - // ВСЯ работа с очередью — под одним Lock. - // Раньше проверка существования шла БЕЗ блокировки (чтение map) — - // concurrent map read and map write при параллельном DeleteQueue - // (fatal error, под падал под нагрузкой). - // defer Unlock: при любой панике в секции мьютекс не останется захваченным. - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() messageFound := false var changedMsgUuid string @@ -76,7 +69,9 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo queue.DeadLetterQueue != nil && msgs[i].Retry >= queue.MaxReceiveCount { changedMsgUuid = msgs[i].Uuid + queue.DeadLetterQueue.LockMu() queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, msgs[i]) + queue.DeadLetterQueue.UnlockMu() queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...) msgRemoved = true } else { diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index ad86ad1..620bd73 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -59,15 +59,11 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract ids[entry.Id] = true } - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - // Проверка существования очереди ВНУТРИ Lock: раньше была БЕЗ блокировки - // (чтение map) — concurrent map read/write при параллельном DeleteQueue. - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() successEntries := make([]models.ChangeMessageVisibilityBatchResultEntry, 0) failedEntries := make([]models.BatchResultErrorEntry, 0) diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index a6f4a1f..7ab8293 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -49,6 +49,27 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { queueUrl := tenantQueueURL(t, queueName) queueArn := tenantQueueARN(t, queueName) + // Готовим очередь и применяем атрибуты ВНЕ глобального лока: setQueueAttributesV1 + // берёт SyncQueues.RLock (поиск DLQ), а RWMutex не реентрантен — вызов под + // уже захваченным SyncQueues.Lock дал бы дедлок. + queue := &models.Queue{ + Name: queueName, + URL: queueUrl, + Arn: queueArn, + IsFIFO: utils.HasFIFOQueueName(queueName), + EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, + Duplicates: make(map[string]time.Time), + } + // Список имён переданных атрибутов — для whitelist-валидации + // (Query-протокол: Attribute.N.Name; JSON: nil — проверка пропускается). + var provided map[string]string + if req.Header.Get("Content-Type") != "application/x-amz-json-1.0" { + provided = utils.ExtractQueueAttributes(req.PostForm) + } + if err := setQueueAttributesV1(queue, requestBody.Attributes, provided, t.AccessKey); err != nil { + return utils.CreateErrorResponseV1(err.Error(), true) + } + models.SyncQueues.Lock() if _, exists := models.SyncQueues.Queues[key]; !exists { // Проверка лимита очередей тенанта @@ -57,24 +78,6 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { return utils.CreateErrorResponseV1("LimitExceeded", true) } log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID) - queue := &models.Queue{ - Name: queueName, - URL: queueUrl, - Arn: queueArn, - IsFIFO: utils.HasFIFOQueueName(queueName), - EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, - Duplicates: make(map[string]time.Time), - } - // Список имён переданных атрибутов — для whitelist-валидации - // (Query-протокол: Attribute.N.Name; JSON: nil — проверка пропускается). - var provided map[string]string - if req.Header.Get("Content-Type") != "application/x-amz-json-1.0" { - provided = utils.ExtractQueueAttributes(req.PostForm) - } - if err := setQueueAttributesV1(queue, requestBody.Attributes, provided, t.AccessKey); err != nil { - models.SyncQueues.Unlock() - return utils.CreateErrorResponseV1(err.Error(), true) - } models.SyncQueues.AddQueue(key, queue) } // Сохраняем очередь в Redis пока держим Lock — консистентный снапшот diff --git a/app/gosqs/delete_message.go b/app/gosqs/delete_message.go index e31dd0a..b56c413 100644 --- a/app/gosqs/delete_message.go +++ b/app/gosqs/delete_message.go @@ -42,30 +42,31 @@ func DeleteMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { key := tenantQueueKey(t.AccessKey, queueName) log.Info("Deleting Message, Queue:", queueName, ", ReceiptHandle:", receiptHandle) - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - if _, ok := models.SyncQueues.Queues[key]; ok { - for i, msg := range models.SyncQueues.Queues[key].Messages { - if msg.ReceiptHandle == receiptHandle { - msgUuid := msg.Uuid - models.SyncQueues.Queues[key].UnlockGroup(msg.GroupID) - models.SyncQueues.Queues[key].Messages = append(models.SyncQueues.Queues[key].Messages[:i], models.SyncQueues.Queues[key].Messages[i+1:]...) - delete(models.SyncQueues.Queues[key].Duplicates, msg.DeduplicationID) - // Удаляем одно сообщение из Redis — O(1) - persistence.DeleteMessagePersist(key, msgUuid) - // Метаданные очереди (duplicates, FIFO state) - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) - respStruct := models.DeleteMessageResponse{ - Xmlns: models.BaseXmlns, - Metadata: models.BaseResponseMetadata, - } - return 200, &respStruct - } - } - log.Warning("Receipt Handle not found") - } else { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { log.Warning("Queue not found") + return utils.CreateErrorResponseV1("MessageDoesNotExist", true) } + defer queue.UnlockMu() + + for i, msg := range queue.Messages { + if msg.ReceiptHandle == receiptHandle { + msgUuid := msg.Uuid + queue.UnlockGroup(msg.GroupID) + queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...) + delete(queue.Duplicates, msg.DeduplicationID) + // Удаляем одно сообщение из Redis — O(1) + persistence.DeleteMessagePersist(key, msgUuid) + // Метаданные очереди (duplicates, FIFO state) + persistence.SaveQueue(key, queue) + respStruct := models.DeleteMessageResponse{ + Xmlns: models.BaseXmlns, + Metadata: models.BaseResponseMetadata, + } + return 200, &respStruct + } + } + log.Warning("Receipt Handle not found") return utils.CreateErrorResponseV1("MessageDoesNotExist", true) } diff --git a/app/gosqs/delete_message_batch.go b/app/gosqs/delete_message_batch.go index 40d36b8..b0c1182 100644 --- a/app/gosqs/delete_message_batch.go +++ b/app/gosqs/delete_message_batch.go @@ -56,15 +56,11 @@ func DeleteMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBo ids[v.Id] = true } - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - // Проверка существования очереди ВНУТРИ Lock: раньше была БЕЗ блокировки - // (чтение map) — concurrent map read/write при параллельном DeleteQueue. - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() deleteMessageMap := make(map[string]*deleteEntry) for _, entry := range requestBody.Entries { diff --git a/app/gosqs/get_queue_attributes.go b/app/gosqs/get_queue_attributes.go index 9faeb84..61bc53c 100644 --- a/app/gosqs/get_queue_attributes.go +++ b/app/gosqs/get_queue_attributes.go @@ -64,13 +64,12 @@ func GetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo log.Infof("Get Queue Attributes: %s (tenant: %s)", queueName, t.ID) queueAttributes := make([]models.Attribute, 0) - models.SyncQueues.RLock() - defer models.SyncQueues.RUnlock() - queue, ok := models.SyncQueues.Queues[key] - if !ok { + queue := models.SyncQueues.GetRLocked(key) + if queue == nil { log.Errorf("Get Queue Attributes: %s queue does not exist for tenant %s", queueName, t.ID) return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.RUnlockMu() if shouldInclude("DelaySeconds") { queueAttributes = append(queueAttributes, models.Attribute{Name: "DelaySeconds", Value: strconv.Itoa(queue.DelaySeconds)}) diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index 5099d5b..85d721d 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -48,9 +48,9 @@ func PeriodicTasks(d time.Duration, quit chan bool) { // Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной // очереди не мешает обработке остальных и HTTP-запросам. for _, qi := range queues { - models.SyncQueues.Lock() + qi.queue.LockMu() processQueueTick(qi.key, qi.queue) - models.SyncQueues.Unlock() + qi.queue.UnlockMu() } case <-quit: ticker.Stop() @@ -61,7 +61,7 @@ func PeriodicTasks(d time.Duration, quit chan bool) { // processQueueTick — обработка ОДНОЙ очереди за один такт. // -// ВЫЗЫВАТЬ ТОЛЬКО под удержанием SyncQueues.Lock. +// ВЫЗЫВАТЬ под per-queue блокировкой этой очереди (queue.LockMu). // // Логическая схема: // 0. Эвикция сообщений по MessageRetentionPeriod (старше retention удаляются). @@ -129,7 +129,9 @@ func processQueueTick(key string, queue *models.Queue) { if queue.MaxReceiveCount > 0 && queue.DeadLetterQueue != nil && msg.Retry >= queue.MaxReceiveCount { + queue.DeadLetterQueue.LockMu() queue.DeadLetterQueue.Messages = append(queue.DeadLetterQueue.Messages, *msg) + queue.DeadLetterQueue.UnlockMu() queue.Messages = append(queue.Messages[:i], queue.Messages[i+1:]...) i-- // элемент на позиции i удалён — повторно проверяем новый элемент на ней } diff --git a/app/gosqs/list_queue_tags.go b/app/gosqs/list_queue_tags.go index 75c6e60..e58f76a 100644 --- a/app/gosqs/list_queue_tags.go +++ b/app/gosqs/list_queue_tags.go @@ -40,13 +40,11 @@ func ListQueueTagsV1(req *http.Request) (int, interfaces.AbstractResponseBody) { key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.RLock() - defer models.SyncQueues.RUnlock() - - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetRLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.RUnlockMu() tags := queue.Tags if tags == nil { diff --git a/app/gosqs/purge_queue.go b/app/gosqs/purge_queue.go index 5340cd1..1248b92 100644 --- a/app/gosqs/purge_queue.go +++ b/app/gosqs/purge_queue.go @@ -45,29 +45,29 @@ func PurgeQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { queueName := uriSegments[len(uriSegments)-1] key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - if _, ok := models.SyncQueues.Queues[key]; !ok { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { log.Errorf("Purge Queue: %s, queue does not exist for tenant %s", queueName, t.ID) return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() // AWS: PurgeQueue допускается не чаще одного раза в 60 секунд на очередь. // Нулевой LastPurgeTime (purge ещё не выполнялся) → запрет не срабатывает. - if time.Since(models.SyncQueues.Queues[key].LastPurgeTime) < 60*time.Second { - log.Warnf("Purge Queue: %s rejected — purge in progress (last: %s)", queueName, models.SyncQueues.Queues[key].LastPurgeTime.Format(time.RFC3339)) + if time.Since(queue.LastPurgeTime) < 60*time.Second { + log.Warnf("Purge Queue: %s rejected — purge in progress (last: %s)", queueName, queue.LastPurgeTime.Format(time.RFC3339)) return utils.CreateErrorResponseV1("PurgeQueueInProgress", true) } log.Infof("Purging Queue: %s (tenant: %s)", queueName, t.ID) - models.SyncQueues.Queues[key].Messages = nil - models.SyncQueues.Queues[key].Duplicates = make(map[string]time.Time) + queue.Messages = nil + queue.Duplicates = make(map[string]time.Time) // Фиксируем время purge — защита от повторного вызова в течение 60 секунд. - models.SyncQueues.Queues[key].LastPurgeTime = time.Now() + queue.LastPurgeTime = time.Now() // Удаляем все сообщения из Redis одной командой DEL persistence.PurgeMessagesPersist(key) // Сохраняем пустые метаданные очереди - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) + persistence.SaveQueue(key, queue) respStruct := models.PurgeQueueResponse{ Xmlns: models.BaseXmlns, diff --git a/app/gosqs/queue_attributes.go b/app/gosqs/queue_attributes.go index 09351a7..3de8c2c 100644 --- a/app/gosqs/queue_attributes.go +++ b/app/gosqs/queue_attributes.go @@ -75,7 +75,9 @@ func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided queueName := arnArray[len(arnArray)-1] // DLQ ищем по tenant-scoped ключу — как хранятся все очереди. dlqKey := tenantAccessKey + ":" + queueName + models.SyncQueues.RLock() deadLetterQueue, ok := models.SyncQueues.Queues[dlqKey] + models.SyncQueues.RUnlock() if !ok { log.Error("Invalid RedrivePolicy Attribute") return fmt.Errorf("InvalidAttributeValue") diff --git a/app/gosqs/receive_message.go b/app/gosqs/receive_message.go index ff071a0..622aebc 100644 --- a/app/gosqs/receive_message.go +++ b/app/gosqs/receive_message.go @@ -86,17 +86,15 @@ func ReceiveMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) // ОДИН RLock на проверку существования И чтение атрибута очереди: // раньше было два отдельных RLock — между ними очередь могла быть удалена // (nil-deref при чтении ReceiveMessageWaitTimeSeconds). - models.SyncQueues.RLock() - queue, queueExists := models.SyncQueues.Queues[key] - if !queueExists { - models.SyncQueues.RUnlock() + queue := models.SyncQueues.GetRLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } waitTimeSeconds := requestBody.WaitTimeSeconds if waitTimeSeconds == 0 { waitTimeSeconds = queue.ReceiveMessageWaitTimeSeconds } - models.SyncQueues.RUnlock() + queue.RUnlockMu() if waitTimeSeconds < 0 || waitTimeSeconds > MaxReceiveMessageWaitTimeSeconds { return utils.CreateErrorResponseV1("InvalidParameterValue", true) } @@ -155,13 +153,11 @@ func ReceiveMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) // (ApproximateReceiveCount в ответе); // - для FIFO-очереди блокируется группа (LockGroup) до DeleteMessage/таймаута. func receiveMessagesUnderLock(key string, visibilityTimeout, maxNumberOfMessages int) ([]*models.ResultMessage, bool) { - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - queue, queueFound := models.SyncQueues.Queues[key] - if !queueFound || len(queue.Messages) == 0 { + queue := models.SyncQueues.GetLocked(key) + if queue == nil || len(queue.Messages) == 0 { return nil, false } + defer queue.UnlockMu() // Быстрая предпроверка: есть ли вообще хотя бы одно готовое сообщение. if !queueHasReceivableMessages(queue) { return nil, false diff --git a/app/gosqs/send_message.go b/app/gosqs/send_message.go index 97bcf7a..9ada36b 100644 --- a/app/gosqs/send_message.go +++ b/app/gosqs/send_message.go @@ -83,17 +83,16 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.RLock() - queue, exists := models.SyncQueues.Queues[key] - if !exists { - models.SyncQueues.RUnlock() + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() + maxMessageSize := queue.MaximumMessageSize currentMsgCount := len(queue.Messages) queueIsFIFO := queue.IsFIFO delaySecs := queue.DelaySeconds - models.SyncQueues.RUnlock() // FIFO-очередь БЕЗ MessageGroupId — ошибка: AWS требует GroupId для каждой // отправки в FIFO (иначе невозможно гарантировать порядок внутри группы). @@ -128,18 +127,9 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { msg.SentTime = time.Now() msg.DelaySecs = delaySecs - models.SyncQueues.Lock() - // Перепроверка под Lock: очередь могла быть удалена между RLock-проверкой - // выше и захватом Lock (иначе nil-pointer deref). - queue = models.SyncQueues.Queues[key] - if queue == nil { - models.SyncQueues.Unlock() - return utils.CreateErrorResponseV1("QueueNotFound", true) - } - // Повторная проверка лимита под Lock: между RLock-снимком и захватом Lock - // другие горутины могли добавить сообщения (мягкий OOM-лимит). + // Повторная проверка лимита под per-queue Lock (между снимком выше и текущим + // моментом другие горутины могли добавить сообщения — мягкий OOM-лимит). if len(queue.Messages) >= MaxMessagesForQueue(queue.IsFIFO) { - models.SyncQueues.Unlock() return utils.CreateErrorResponseV1("OverLimit", true) } @@ -167,7 +157,6 @@ func SendMessageV1(req *http.Request) (int, interfaces.AbstractResponseBody) { queue.InitDuplicatation(messageDeduplicationID) // Сохраняем метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue) - models.SyncQueues.Unlock() if persistErr { return utils.CreateErrorResponseV1("InternalError", true) diff --git a/app/gosqs/send_message_batch.go b/app/gosqs/send_message_batch.go index 9a46896..ec10979 100644 --- a/app/gosqs/send_message_batch.go +++ b/app/gosqs/send_message_batch.go @@ -43,16 +43,14 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.RLock() - queue, exists := models.SyncQueues.Queues[key] - if !exists { - models.SyncQueues.RUnlock() + queue := models.SyncQueues.GetRLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } maxMsgSize := queue.MaximumMessageSize currentMsgCount := len(queue.Messages) queueIsFIFO := queue.IsFIFO - models.SyncQueues.RUnlock() + queue.RUnlockMu() sendEntries := requestBody.Entries @@ -104,14 +102,8 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody return utils.CreateErrorResponseV1("OverLimit", true) } - models.SyncQueues.Lock() - // Перепроверка под Lock: очередь могла быть удалена между RLock-проверкой - // выше и захватом Lock (иначе nil-pointer deref). - queue = models.SyncQueues.Queues[key] - if queue == nil { - models.SyncQueues.Unlock() - return utils.CreateErrorResponseV1("QueueNotFound", true) - } + queue.LockMu() + defer queue.UnlockMu() newMsgs := make([]models.SqsMessage, 0, len(sendEntries)) for _, sendEntry := range sendEntries { msg := models.SqsMessage{MessageBody: sendEntry.MessageBody} @@ -151,7 +143,6 @@ func SendMessageBatchV1(req *http.Request) (int, interfaces.AbstractResponseBody persistence.SaveMessages(key, newMsgs) // Метаданные очереди (FIFO state, duplicates) — без Messages persistence.SaveQueue(key, queue) - models.SyncQueues.Unlock() respStruct := models.SendMessageBatchResponse{ Xmlns: models.BaseXmlns, diff --git a/app/gosqs/set_queue_attributes.go b/app/gosqs/set_queue_attributes.go index 8271932..61fcc41 100644 --- a/app/gosqs/set_queue_attributes.go +++ b/app/gosqs/set_queue_attributes.go @@ -45,13 +45,12 @@ func SetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo key := tenantQueueKey(t.AccessKey, queueName) log.Infof("Set Queue Attributes: %s (tenant: %s)", queueName, t.ID) - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - queue, ok := models.SyncQueues.Queues[key] - if !ok { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { log.Warningf("Set Queue Attributes: %s, queue does not exist for tenant %s", queueName, t.ID) return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() // Список имён переданных атрибутов — для whitelist-валидации. // Query-протокол: имена видны в form (Attribute.N.Name). // JSON-протокол: структура уже разобрана из тела, список имён недоступен → nil diff --git a/app/gosqs/tag_queue.go b/app/gosqs/tag_queue.go index a1af472..6d60727 100644 --- a/app/gosqs/tag_queue.go +++ b/app/gosqs/tag_queue.go @@ -41,13 +41,11 @@ func TagQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() // Инициализируем Tags если nil (для старых очередей, созданных без Tags) if queue.Tags == nil { diff --git a/app/gosqs/untag_queue.go b/app/gosqs/untag_queue.go index 694e4cc..7f36c17 100644 --- a/app/gosqs/untag_queue.go +++ b/app/gosqs/untag_queue.go @@ -41,13 +41,11 @@ func UntagQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { key := tenantQueueKey(t.AccessKey, queueName) - models.SyncQueues.Lock() - defer models.SyncQueues.Unlock() - - queue, exists := models.SyncQueues.Queues[key] - if !exists { + queue := models.SyncQueues.GetLocked(key) + if queue == nil { return utils.CreateErrorResponseV1("QueueNotFound", true) } + defer queue.UnlockMu() if queue.Tags != nil { for _, tagKey := range requestBody.TagKeys { diff --git a/app/metrics/gauge_updater.go b/app/metrics/gauge_updater.go index ed9291c..4360014 100644 --- a/app/metrics/gauge_updater.go +++ b/app/metrics/gauge_updater.go @@ -37,18 +37,25 @@ func StartGaugeUpdater(interval time.Duration, quit <-chan bool) { // updateGauges — пересчитывает sqs_queues_count и sqs_messages_count per tenant. // Формат ключа очереди: "{accessKey}:{queueName}" — берём часть до первого ":". func updateGauges() { - // Собираем данные под RLock — не блокируем запись + // Снимок указателей под коротким RLock, затем содержимое — под per-queue RLock models.SyncQueues.RLock() + snapshot := make(map[string]*models.Queue, len(models.SyncQueues.Queues)) + for key, queue := range models.SyncQueues.Queues { + snapshot[key] = queue + } + models.SyncQueues.RUnlock() + tenantQueues := make(map[string]int) tenantMessages := make(map[string]int) - for key, queue := range models.SyncQueues.Queues { + for key, queue := range snapshot { // Ключ формата "accessKey:queueName" — tenant определяется по accessKey parts := strings.SplitN(key, ":", 2) tenantKey := parts[0] tenantQueues[tenantKey]++ + queue.RLockMu() tenantMessages[tenantKey] += len(queue.Messages) + queue.RUnlockMu() } - models.SyncQueues.RUnlock() // Сбрасываем старые значения и записываем новые QueuesGauge.Reset() diff --git a/app/models/globals.go b/app/models/globals.go index 3236505..6a2f7b0 100644 --- a/app/models/globals.go +++ b/app/models/globals.go @@ -74,6 +74,31 @@ func (s *QueueStore) RemoveQueue(key string) { } } +// GetLocked — возвращает очередь с захваченной per-queue write-блокировкой. +// nil если очереди нет. Вызывающий ОБЯЗАН вызвать q.UnlockMu(). +func (s *QueueStore) GetLocked(key string) *Queue { + s.RLock() + q := s.Queues[key] + s.RUnlock() + if q == nil { + return nil + } + q.LockMu() + return q +} + +// GetRLocked — то же, но read-блокировка. Вызывающий ОБЯЗАН вызвать q.RUnlockMu(). +func (s *QueueStore) GetRLocked(key string) *Queue { + s.RLock() + q := s.Queues[key] + s.RUnlock() + if q == nil { + return nil + } + q.RLockMu() + return q +} + // CountTenantQueues — число очередей тенанта (по accessKey) за O(1). // Вызывать под SyncQueues.RLock() или Lock(). func (s *QueueStore) CountTenantQueues(accessKey string) int { diff --git a/app/persistence/redis.go b/app/persistence/redis.go index d7a9205..a96c9d0 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -142,23 +142,65 @@ func asyncWrite(fn func()) { }() } -// marshalQueueMeta — сериализует очередь БЕЗ Messages (и без Messages в DLQ). -// DeadLetterQueue сохраняется как ссылка (Name/URL/Arn), но без сообщений. -// ВАЖНО: вызывать под SyncQueues.Lock() — временно nil'ит Messages. +// queueMetaDTO — сериализуемая копия очереди БЕЗ Messages (и без сообщений DLQ). +// Отдельный тип, т.к. models.Queue содержит sync.RWMutex и его нельзя копировать/ +// сериализовать целиком, а временное зануление Messages мутировало бы живую очередь. +// Имена полей совпадают с models.Queue → json совместим при загрузке. +type queueMetaDTO struct { + Name string + URL string + Arn string + VisibilityTimeout int + ReceiveMessageWaitTimeSeconds int + DelaySeconds int + MaximumMessageSize int + MessageRetentionPeriod int + MaxReceiveCount int + IsFIFO bool + FIFOMessages map[string]int + FIFOSequenceNumbers map[string]int + EnableDuplicates bool + Duplicates map[string]time.Time + Tags map[string]string + LastPurgeTime time.Time + DeadLetterQueue *queueMetaDTO +} + +// queueToMetaDTO — строит снимок очереди без сообщений (и без сообщений DLQ), +// НЕ мутируя исходную очередь. Рекурсия ограничена глубиной 1. +func queueToMetaDTO(q *models.Queue) *queueMetaDTO { + if q == nil { + return nil + } + dto := &queueMetaDTO{ + Name: q.Name, + URL: q.URL, + Arn: q.Arn, + VisibilityTimeout: q.VisibilityTimeout, + ReceiveMessageWaitTimeSeconds: q.ReceiveMessageWaitTimeSeconds, + DelaySeconds: q.DelaySeconds, + MaximumMessageSize: q.MaximumMessageSize, + MessageRetentionPeriod: q.MessageRetentionPeriod, + MaxReceiveCount: q.MaxReceiveCount, + IsFIFO: q.IsFIFO, + FIFOMessages: q.FIFOMessages, + FIFOSequenceNumbers: q.FIFOSequenceNumbers, + EnableDuplicates: q.EnableDuplicates, + Duplicates: q.Duplicates, + Tags: q.Tags, + LastPurgeTime: q.LastPurgeTime, + } + if q.DeadLetterQueue != nil { + dto.DeadLetterQueue = queueToMetaDTO(q.DeadLetterQueue) + } + return dto +} + +// marshalQueueMeta — сериализует метаданные очереди БЕЗ сообщений, +// НЕ мутируя исходную очередь (см. queueMetaDTO). Безопасно вызывать +// под per-queue блокировкой только этой очереди. func marshalQueueMeta(queue *models.Queue) ([]byte, error) { - savedMsgs := queue.Messages - queue.Messages = nil - var savedDLQMsgs []models.SqsMessage - if queue.DeadLetterQueue != nil { - savedDLQMsgs = queue.DeadLetterQueue.Messages - queue.DeadLetterQueue.Messages = nil - } - data, err := json.Marshal(queue) - queue.Messages = savedMsgs - if queue.DeadLetterQueue != nil { - queue.DeadLetterQueue.Messages = savedDLQMsgs - } - return data, err + return json.Marshal(queueToMetaDTO(queue)) } // SaveQueue — сохраняет МЕТАДАННЫЕ очереди (без сообщений) в Redis асинхронно. 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 index 10bc254..95438a1 100644 --- 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 @@ -47,6 +47,20 @@ DLQ), а не построчная замена. Делать её «вслеп рискованно: ошибка = дедлоки/потеря данных на проде. Рекомендуется отдельный проход: сначала тесты на FIFO/DLQ/batch, затем атомарный перевод. +## СТАТУС: РЕАЛИЗОВАНО (2026-10-06) + +Перевод выполнен атомарно. Что сделано: +- `marshalQueueMeta` больше НЕ мутирует очередь (DTO-снапшот queueMetaDTO). +- Все операции с содержимым очереди используют per-queue лок + (`QueueStore.GetLocked/GetRLocked` + `queue.LockMu/RUnlockMu`) в gosqs, admin, + metrics/gauge_updater, PeriodicTasks. +- Глобальный `SyncQueues`-лок остался ТОЛЬКО для доступа к map и неизменяемых + полей (create/delete/list/find/snapshot). +- DLQ-переносы берут лок DLQ отдельно (порядок main→DLQ). +- Устранён дедлок в create_queue (setQueueAttributesV1 вызывается ВНЕ глобального лока). +- Добавлены конкурентные тесты (два тенанта / одна очередь), прогон `go test -race` — зелёный. + + ## P1.2 (индексация сообщений) — та же природа Индексация `Messages` (receipt→index) меняет структуру хранения, затронутую теми