Repinoid
fd8ffc329d
Все операции с содержимым очереди (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 больше не держит глобальный лок.
93 lines
4.1 KiB
Go
93 lines
4.1 KiB
Go
package gosqs
|
||
|
||
import (
|
||
"net/http"
|
||
"time"
|
||
|
||
"shared-sqs/app/interfaces"
|
||
"shared-sqs/app/models"
|
||
"shared-sqs/app/persistence"
|
||
"shared-sqs/app/utils"
|
||
|
||
log "github.com/sirupsen/logrus"
|
||
)
|
||
|
||
// CreateQueueV1 — создаёт очередь для тенанта из request context.
|
||
//
|
||
// Логическая схема:
|
||
// 1. Разбор тела запроса (Query/JSON) → CreateQueueRequest.
|
||
// 2. Тенант из контекста (SigV4 → AccessKey).
|
||
// 3. Валидация имени очереди по AWS-правилам: до 80 символов, [a-zA-Z0-9_-],
|
||
// опциональный суффикс .fifo (ValidateQueueName).
|
||
// 4. Под Lock:
|
||
// - если очередь уже есть — просто возвращаем её URL (идемпотентность AWS);
|
||
// - проверка лимита очередей тенанта (MaxQueues) → LimitExceeded;
|
||
// - создание очереди с дефолтами и применение переданных атрибутов
|
||
// (setQueueAttributesV1 с валидацией диапазонов);
|
||
// - вставка в map с tenant-scoped ключом "{tenantAccessKey}:{queueName}".
|
||
// 5. Сохранение очереди в Redis (асинхронно, маршалинг под Lock).
|
||
// 6. Ответ: QueueUrl.
|
||
func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) {
|
||
requestBody := models.NewCreateQueueRequest()
|
||
ok := utils.REQUEST_TRANSFORMER(requestBody, req, false)
|
||
if !ok {
|
||
log.Error("Invalid Request - CreateQueueV1")
|
||
return utils.CreateErrorResponseV1("InvalidParameterValue", true)
|
||
}
|
||
|
||
t := getTenantFromContext(req)
|
||
if t == nil {
|
||
return utils.CreateErrorResponseV1("InvalidClientTokenId", true)
|
||
}
|
||
|
||
// Валидация имени очереди по AWS правилам: макс 80 chars, [a-zA-Z0-9_-], опц. .fifo
|
||
queueName := requestBody.QueueName
|
||
if err := ValidateQueueName(queueName); err != nil {
|
||
return utils.CreateErrorResponseV1("InvalidParameterValue", true)
|
||
}
|
||
key := tenantQueueKey(t.AccessKey, queueName)
|
||
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 {
|
||
// Проверка лимита очередей тенанта
|
||
if t.MaxQueues > 0 && countTenantQueues(t.AccessKey) >= t.MaxQueues {
|
||
models.SyncQueues.Unlock()
|
||
return utils.CreateErrorResponseV1("LimitExceeded", true)
|
||
}
|
||
log.Infof("Creating Queue: %s (tenant: %s)", queueName, t.ID)
|
||
models.SyncQueues.AddQueue(key, queue)
|
||
}
|
||
// Сохраняем очередь в Redis пока держим Lock — консистентный снапшот
|
||
persistence.SaveQueue(key, models.SyncQueues.Queues[key])
|
||
models.SyncQueues.Unlock()
|
||
respStruct := models.CreateQueueResponse{
|
||
Xmlns: models.BaseXmlns,
|
||
Result: models.CreateQueueResult{QueueUrl: queueUrl},
|
||
Metadata: models.BaseResponseMetadata,
|
||
}
|
||
return http.StatusOK, respStruct
|
||
}
|