Nail/SQS-service
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

P1.1: per-queue блокировки вместо глобального лока на операции с очередью

Browse Source 
Все операции с содержимым очереди (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
больше не держит глобальный лок.
This commit is contained in:
Repinoid
2026-10-06 18:58:16 +03:00
parent ed17c6cbd0
commit fd8ffc329d
21 changed files with 252 additions and 181 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/gosqs/change_message_visibility.go
+5 -10
View File
@@ -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 {
app/gosqs/change_message_visibility_batch.go
+3 -7
View File
@@ -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)
app/gosqs/create_queue.go
+21 -18
View File
@@ -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 — консистентный снапшот
app/gosqs/delete_message.go
+23 -22
View File
@@ -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)
}
app/gosqs/delete_message_batch.go
+3 -7
View File
@@ -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 {
app/gosqs/get_queue_attributes.go
+3 -4
View File
@@ -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)})
app/gosqs/gosqs.go
+5 -3
View File
@@ -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 удалён — повторно проверяем новый элемент на ней
}
app/gosqs/list_queue_tags.go
+3 -5
View File
@@ -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 {
app/gosqs/purge_queue.go
+9 -9
View File
@@ -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,
app/gosqs/queue_attributes.go
+2
View File
@@ -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")
app/gosqs/receive_message.go
+6 -10
View File
@@ -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
app/gosqs/send_message.go
+6 -17
View File
@@ -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)
app/gosqs/send_message_batch.go
+5 -14
View File
@@ -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,
app/gosqs/set_queue_attributes.go
+3 -4
View File
@@ -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
app/gosqs/tag_queue.go
+3 -5
View File
@@ -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 {
app/gosqs/untag_queue.go
+3 -5
View File
@@ -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 {