Files
SQS-service/app/models/globals.go
T
Repinoid fd8ffc329d P1.1: per-queue блокировки вместо глобального лока на операции с очередью
Все операции с содержимым очереди (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
больше не держит глобальный лок.
2026-10-06 18:58:16 +03:00

124 lines
4.2 KiB
Go

package models
import (
"strings"
"sync"
)
// CurrentEnvironment should get overwritten when the app starts up and loads the config. For the
// sake of generating "partial" apps piece-meal during test automation we'll slap these placeholder
// values in here so the resource URLs aren't wonky like `http://://new-queue`.
var CurrentEnvironment = Environment{
Host: "host",
Port: "port",
Region: "region",
AccountID: "accountID",
}
var LogMessages bool
var LogFile string
// QueueStore — хранилище очередей: блокировка + map ключ→очередь + индекс tenant→очереди.
//
// Ключ очереди в Queues — "{accessKey}:{queueName}" (tenant-scoped изоляция).
// Индекс tenantQueues (accessKey → набор имён очередей) позволяет считать и
// перечислять очереди тенанта за O(число его очередей), без скана всех очередей сервиса.
//
// ВАЖНО: методы AddQueue/RemoveQueue/CountTenantQueues/TenantQueueNames НЕ берут
// блокировку — вызывать под уже захваченным SyncQueues.Lock()/RLock()
// (соответствует существующей дисциплине блокировок в коде).
type QueueStore struct {
sync.RWMutex
Queues map[string]*Queue
tenantQueues map[string]map[string]struct{}
}
var SyncQueues = QueueStore{
Queues: make(map[string]*Queue),
tenantQueues: make(map[string]map[string]struct{}),
}
// splitQueueKey — разбирает tenant-scoped ключ "{accessKey}:{queueName}".
// Легаси-ключи без ":" (yaml-очереди) tenant-независимы → ok=false.
func splitQueueKey(key string) (accessKey, queueName string, ok bool) {
idx := strings.Index(key, ":")
if idx <= 0 || idx >= len(key)-1 {
return "", "", false
}
return key[:idx], key[idx+1:], true
}
// AddQueue — добавляет очередь и обновляет tenant-индекс.
// Вызывать под SyncQueues.Lock().
func (s *QueueStore) AddQueue(key string, q *Queue) {
s.Queues[key] = q
if ak, name, ok := splitQueueKey(key); ok {
if s.tenantQueues[ak] == nil {
s.tenantQueues[ak] = make(map[string]struct{})
}
s.tenantQueues[ak][name] = struct{}{}
}
}
// RemoveQueue — удаляет очередь и обновляет tenant-индекс.
// Вызывать под SyncQueues.Lock().
func (s *QueueStore) RemoveQueue(key string) {
delete(s.Queues, key)
if ak, name, ok := splitQueueKey(key); ok {
if set, exists := s.tenantQueues[ak]; exists {
delete(set, name)
if len(set) == 0 {
delete(s.tenantQueues, ak)
}
}
}
}
// 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 {
return len(s.tenantQueues[accessKey])
}
// TenantQueueNames — копия списка имён очередей тенанта.
// Вызывать под SyncQueues.RLock() или Lock().
func (s *QueueStore) TenantQueueNames(accessKey string) []string {
set := s.tenantQueues[accessKey]
names := make([]string, 0, len(set))
for n := range set {
names = append(names, n)
}
return names
}
// ResetQueueIndex — очищает tenant-индекс (для ResetResources/тестов).
// Вызывать под SyncQueues.Lock().
func (s *QueueStore) ResetQueueIndex() {
s.tenantQueues = make(map[string]map[string]struct{})
}