From 3a52960d5b6cdce7900a329cfd7317f3f53aead7 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Tue, 6 Oct 2026 18:36:03 +0300 Subject: [PATCH] =?UTF-8?q?P1.4:=20tenant-=D0=B8=D0=BD=D0=B4=D0=B5=D0=BA?= =?UTF-8?q?=D1=81=20=D0=BE=D1=87=D0=B5=D1=80=D0=B5=D0=B4=D0=B5=D0=B9=20(O(?= =?UTF-8?q?1)=20=D0=BF=D0=BE=D0=B4=D1=81=D1=87=D1=91=D1=82/=D1=81=D0=BF?= =?UTF-8?q?=D0=B8=D1=81=D0=BE=D0=BA=20=D0=B2=D0=BC=D0=B5=D1=81=D1=82=D0=BE?= =?UTF-8?q?=20=D1=81=D0=BA=D0=B0=D0=BD=D0=B0)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Введён тип QueueStore с индексом accessKey→{queueName}. Мутации очередей идут через AddQueue/RemoveQueue (create/delete/restore/config/seed/admin), подсчёт и список — через CountTenantQueues/TenantQueueNames. Устранены O(total) скан-по-префиксу в countTenantQueues, deleteTenant, listTenantQueues, createTenantQueue. API SyncQueues (Queues, Lock) сохранён — остальной код не затронут. --- app/admin/admin.go | 72 ++++++++++++++++----------------- app/cmd/goaws.go | 2 +- app/cmd/seed.go | 4 +- app/conf/config.go | 4 +- app/gosqs/create_queue.go | 2 +- app/gosqs/delete_queue.go | 2 +- app/gosqs/tenant_helpers.go | 21 ++++------ app/models/globals.go | 81 +++++++++++++++++++++++++++++++++++-- app/models/helpers.go | 1 + 9 files changed, 127 insertions(+), 62 deletions(-) diff --git a/app/admin/admin.go b/app/admin/admin.go index 0c1eb21..0ac0df2 100644 --- a/app/admin/admin.go +++ b/app/admin/admin.go @@ -405,15 +405,14 @@ func (h *Handler) deleteTenant(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(map[string]string{"error": "tenant not found"}) return } - // Удаляем все очереди тенанта из SyncQueues - prefix := t.AccessKey + ":" - deletedQueueKeys := make([]string, 0) + // Удаляем все очереди тенанта из SyncQueues (через tenant-индекс, без скана всех очередей) models.SyncQueues.Lock() - for key := range models.SyncQueues.Queues { - if strings.HasPrefix(key, prefix) { - deletedQueueKeys = append(deletedQueueKeys, key) - delete(models.SyncQueues.Queues, key) - } + tenantQueueNames := models.SyncQueues.TenantQueueNames(t.AccessKey) + deletedQueueKeys := make([]string, 0, len(tenantQueueNames)) + for _, name := range tenantQueueNames { + key := t.AccessKey + ":" + name + deletedQueueKeys = append(deletedQueueKeys, key) + models.SyncQueues.RemoveQueue(key) } models.SyncQueues.Unlock() for _, queueKey := range deletedQueueKeys { @@ -447,27 +446,28 @@ func (h *Handler) listTenantQueues(w http.ResponseWriter, r *http.Request) { } queues := make([]queueInfo, 0) models.SyncQueues.RLock() - for key, q := range models.SyncQueues.Queues { - if strings.HasPrefix(key, prefix) { - name := strings.TrimPrefix(key, prefix) - visible := 0 - notVisible := 0 - for _, msg := range q.Messages { - if msg.ReceiptHandle != "" { - notVisible++ - } else { - visible++ - } - } - queues = append(queues, queueInfo{ - Name: name, - Messages: visible, - NotVisible: notVisible, - VisibilityTimeout: q.VisibilityTimeout, - MaxMessageSize: q.MaximumMessageSize, - RetentionPeriod: q.MessageRetentionPeriod, - }) + for _, name := range models.SyncQueues.TenantQueueNames(t.AccessKey) { + q := models.SyncQueues.Queues[prefix+name] + if q == nil { + continue } + visible := 0 + notVisible := 0 + for _, msg := range q.Messages { + if msg.ReceiptHandle != "" { + notVisible++ + } else { + visible++ + } + } + queues = append(queues, queueInfo{ + Name: name, + Messages: visible, + NotVisible: notVisible, + VisibilityTimeout: q.VisibilityTimeout, + MaxMessageSize: q.MaximumMessageSize, + RetentionPeriod: q.MessageRetentionPeriod, + }) } models.SyncQueues.RUnlock() w.Header().Set("Content-Type", "application/json") @@ -504,19 +504,14 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) { jsonErr(w, http.StatusConflict, "queue already exists") return } - // Проверяем лимит очередей тенанта - count := 0 - for k := range models.SyncQueues.Queues { - if strings.HasPrefix(k, t.AccessKey+":") { - count++ - } - } + // Проверяем лимит очередей тенанта (через tenant-индекс, O(1)) + count := models.SyncQueues.CountTenantQueues(t.AccessKey) if t.MaxQueues > 0 && count >= t.MaxQueues { models.SyncQueues.Unlock() jsonErr(w, http.StatusForbidden, "queue limit exceeded") return } - models.SyncQueues.Queues[key] = &models.Queue{ + newQueue := &models.Queue{ Name: req.Name, VisibilityTimeout: 30, MaximumMessageSize: 262144, @@ -524,7 +519,8 @@ func (h *Handler) createTenantQueue(w http.ResponseWriter, r *http.Request) { Messages: []models.SqsMessage{}, Duplicates: make(map[string]time.Time), } - persistence.SaveQueue(key, models.SyncQueues.Queues[key]) + models.SyncQueues.AddQueue(key, newQueue) + persistence.SaveQueue(key, newQueue) models.SyncQueues.Unlock() log.Infof("admin: created queue %s for tenant %s", req.Name, t.ID) w.Header().Set("Content-Type", "application/json") @@ -549,7 +545,7 @@ func (h *Handler) deleteTenantQueue(w http.ResponseWriter, r *http.Request) { jsonErr(w, http.StatusNotFound, "queue not found") return } - delete(models.SyncQueues.Queues, key) + models.SyncQueues.RemoveQueue(key) models.SyncQueues.Unlock() persistence.DeleteQueue(key) log.Infof("admin: deleted queue %s for tenant %s", queueName, t.ID) diff --git a/app/cmd/goaws.go b/app/cmd/goaws.go index 1f7a8f2..f2d6282 100644 --- a/app/cmd/goaws.go +++ b/app/cmd/goaws.go @@ -116,7 +116,7 @@ func main() { } else { models.SyncQueues.Lock() for k, q := range queues { - models.SyncQueues.Queues[k] = q + models.SyncQueues.AddQueue(k, q) } models.SyncQueues.Unlock() } diff --git a/app/cmd/seed.go b/app/cmd/seed.go index e318617..091b315 100644 --- a/app/cmd/seed.go +++ b/app/cmd/seed.go @@ -104,7 +104,7 @@ func seedDemoData(store *tenant.TenantStore) { } models.SyncQueues.Lock() - models.SyncQueues.Queues[key] = &models.Queue{ + models.SyncQueues.AddQueue(key, &models.Queue{ Name: q.name, URL: queueURL, Arn: queueARN, @@ -113,7 +113,7 @@ func seedDemoData(store *tenant.TenantStore) { MessageRetentionPeriod: 345600, Messages: msgs, Duplicates: make(map[string]time.Time), - } + }) models.SyncQueues.Unlock() log.Infof("seed: очередь %s (%d сообщений)", q.name, len(q.msgs)) diff --git a/app/conf/config.go b/app/conf/config.go index fa02bea..8fea77c 100644 --- a/app/conf/config.go +++ b/app/conf/config.go @@ -105,7 +105,7 @@ func LoadYamlConfig(filename string, env string) []string { queue.MessageRetentionPeriod = models.CurrentEnvironment.QueueAttributeDefaults.MessageRetentionPeriod } - models.SyncQueues.Queues[queue.Name] = &models.Queue{ + models.SyncQueues.AddQueue(queue.Name, &models.Queue{ Name: queue.Name, VisibilityTimeout: queue.VisibilityTimeout, Arn: queueArn, @@ -116,7 +116,7 @@ func LoadYamlConfig(filename string, env string) []string { IsFIFO: utils.HasFIFOQueueName(queue.Name), EnableDuplicates: models.CurrentEnvironment.EnableDuplicates, Duplicates: make(map[string]time.Time), - } + }) } // Второй проход — устанавливаем RedrivePolicy, чтобы DLQ были доступны независимо от порядка diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index 59af599..a6f4a1f 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -75,7 +75,7 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { models.SyncQueues.Unlock() return utils.CreateErrorResponseV1(err.Error(), true) } - models.SyncQueues.Queues[key] = queue + models.SyncQueues.AddQueue(key, queue) } // Сохраняем очередь в Redis пока держим Lock — консистентный снапшот persistence.SaveQueue(key, models.SyncQueues.Queues[key]) diff --git a/app/gosqs/delete_queue.go b/app/gosqs/delete_queue.go index ec29c1a..b4bb432 100644 --- a/app/gosqs/delete_queue.go +++ b/app/gosqs/delete_queue.go @@ -50,7 +50,7 @@ func DeleteQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { log.Warnf("Delete Queue: %s does not exist (tenant: %s)", queueName, t.ID) return utils.CreateErrorResponseV1("QueueNotFound", true) } - delete(models.SyncQueues.Queues, key) + models.SyncQueues.RemoveQueue(key) // Удаляем из Redis асинхронно persistence.DeleteQueue(key) diff --git a/app/gosqs/tenant_helpers.go b/app/gosqs/tenant_helpers.go index f1f6337..8aa422f 100644 --- a/app/gosqs/tenant_helpers.go +++ b/app/gosqs/tenant_helpers.go @@ -4,11 +4,11 @@ package gosqs import ( -"net/http" + "net/http" -"shared-sqs/app/auth" -"shared-sqs/app/models" -"shared-sqs/app/tenant" + "shared-sqs/app/auth" + "shared-sqs/app/models" + "shared-sqs/app/tenant" ) // tenantQueueKey — внутренний ключ очереди в SyncQueues в формате "{accessKey}:{queueName}". @@ -41,16 +41,9 @@ func tenantQueueARN(t *tenant.Tenant, queueName string) string { return "arn:aws:sqs:" + models.CurrentEnvironment.Region + ":" + t.ID + ":" + queueName } -// countTenantQueues — считает количество очередей тенанта в SyncQueues. +// countTenantQueues — считает количество очередей тенанта (через tenant-индекс, O(1)). // Используется для проверки лимита MaxQueues. -// Вызывать под SyncQueues.RLock(). +// Вызывать под SyncQueues.RLock()/Lock(). func countTenantQueues(tenantAccessKey string) int { -prefix := tenantAccessKey + ":" -count := 0 -for key := range models.SyncQueues.Queues { -if len(key) > len(prefix) && key[:len(prefix)] == prefix { -count++ -} -} -return count + return models.SyncQueues.CountTenantQueues(tenantAccessKey) } diff --git a/app/models/globals.go b/app/models/globals.go index 8d0c050..3236505 100644 --- a/app/models/globals.go +++ b/app/models/globals.go @@ -1,6 +1,7 @@ package models import ( + "strings" "sync" ) @@ -17,7 +18,81 @@ var CurrentEnvironment = Environment{ var LogMessages bool var LogFile string -var SyncQueues = struct { +// 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 -}{Queues: make(map[string]*Queue)} + 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) + } + } + } +} + +// 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{}) +} diff --git a/app/models/helpers.go b/app/models/helpers.go index cf5c91d..3183ed0 100644 --- a/app/models/helpers.go +++ b/app/models/helpers.go @@ -16,6 +16,7 @@ func ResetApp() { func ResetResources() { SyncQueues.Lock() SyncQueues.Queues = make(map[string]*Queue) + SyncQueues.ResetQueueIndex() SyncQueues.Unlock() }