diff --git a/app/gosqs/create_queue.go b/app/gosqs/create_queue.go index b429ad7..59af599 100644 --- a/app/gosqs/create_queue.go +++ b/app/gosqs/create_queue.go @@ -71,7 +71,7 @@ func CreateQueueV1(req *http.Request) (int, interfaces.AbstractResponseBody) { if req.Header.Get("Content-Type") != "application/x-amz-json-1.0" { provided = utils.ExtractQueueAttributes(req.PostForm) } - if err := setQueueAttributesV1(queue, requestBody.Attributes, provided); err != nil { + if err := setQueueAttributesV1(queue, requestBody.Attributes, provided, t.AccessKey); err != nil { models.SyncQueues.Unlock() return utils.CreateErrorResponseV1(err.Error(), true) } diff --git a/app/gosqs/queue_attributes.go b/app/gosqs/queue_attributes.go index c827cdc..09351a7 100644 --- a/app/gosqs/queue_attributes.go +++ b/app/gosqs/queue_attributes.go @@ -22,9 +22,11 @@ import ( // обнулял все остальные атрибуты очереди. // 3. Значения вне диапазонов AWS → ошибка InvalidParameterValue // (раньше значения молча клэмпились в допустимый диапазон). -// 4. RedrivePolicy: ARN DLQ разбирается, DLQ должна существовать, иначе -// InvalidAttributeValue. -func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided map[string]string) error { +// 4. RedrivePolicy: ARN DLQ разбирается, DLQ ищется по TENANT-SCOPED ключу +// "{accessKey}:{queueName}" (раньше — по голому имени, а ключи в map +// tenant-scoped → DLQ никогда не находилась, RedrivePolicy не работал). +// DLQ должна принадлежать тому же тенанту. +func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided map[string]string, tenantAccessKey string) error { // Шаг 1: whitelist имён атрибутов (единый список — models.AttrNameWhitelist). for name := range provided { if !models.AttrNameWhitelist[name] { @@ -71,7 +73,9 @@ func setQueueAttributesV1(q *models.Queue, attr models.QueueAttributes, provided if attr.RedrivePolicy != (models.RedrivePolicy{}) { arnArray := strings.Split(attr.RedrivePolicy.DeadLetterTargetArn, ":") queueName := arnArray[len(arnArray)-1] - deadLetterQueue, ok := models.SyncQueues.Queues[queueName] + // DLQ ищем по tenant-scoped ключу — как хранятся все очереди. + dlqKey := tenantAccessKey + ":" + queueName + deadLetterQueue, ok := models.SyncQueues.Queues[dlqKey] if !ok { log.Error("Invalid RedrivePolicy Attribute") return fmt.Errorf("InvalidAttributeValue") diff --git a/app/gosqs/set_queue_attributes.go b/app/gosqs/set_queue_attributes.go index cd9ab76..8271932 100644 --- a/app/gosqs/set_queue_attributes.go +++ b/app/gosqs/set_queue_attributes.go @@ -61,7 +61,7 @@ func SetQueueAttributesV1(req *http.Request) (int, interfaces.AbstractResponseBo provided = utils.ExtractQueueAttributes(req.PostForm) } - if err := setQueueAttributesV1(queue, requestBody.Attributes, provided); err != nil { + if err := setQueueAttributesV1(queue, requestBody.Attributes, provided, t.AccessKey); err != nil { return utils.CreateErrorResponseV1(err.Error(), true) } // Сохраняем атрибуты в Redis пока держим Lock (через defer) diff --git a/app/persistence/redis.go b/app/persistence/redis.go index 4e1cd42..c94edf2 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -184,55 +184,49 @@ func SaveMessages(queueKey string, msgs []models.SqsMessage) { }) } -// DeleteMessagePersist — удаляет одно сообщение из Redis асинхронно. -// Вызывать при DeleteMessage (после удаления из in-memory). +// DeleteMessagePersist — удаляет одно сообщение из Redis СИНХРОННО. +// Синхронность обязательна: если удаление потеряется при рестарте пода, +// сообщение «воскреснет» из Redis (факт: удалённые очереди возвращались +// после рестарта, когда удаления были асинхронными). func DeleteMessagePersist(queueKey string, msgUuid string) { if Client == nil || msgUuid == "" { return } hashKey := redisMsgHashPrefix + queueKey - asyncWrite(func() { - ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) - defer cancel() - if err := Client.HDel(ctx, hashKey, msgUuid).Err(); err != nil { - log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err) - } - }) + ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second) + defer cancel() + if err := Client.HDel(ctx, hashKey, msgUuid).Err(); err != nil { + log.Errorf("persistence: HDel message %s/%s: %v", queueKey, msgUuid, err) + } } -// DeleteMessagesPersist — удаляет несколько сообщений из Redis. -// Вызывать при DeleteMessageBatch. +// DeleteMessagesPersist — удаляет несколько сообщений из Redis СИНХРОННО +// (иначе удалённые сообщения воскреснут после рестарта пода). func DeleteMessagesPersist(queueKey string, uuids []string) { if Client == nil || len(uuids) == 0 { return } hashKey := redisMsgHashPrefix + queueKey - // Копируем uuids — вызывающий может переиспользовать slice - ids := make([]string, len(uuids)) - copy(ids, uuids) - asyncWrite(func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := Client.HDel(ctx, hashKey, ids...).Err(); err != nil { - log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(ids), err) - } - }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := Client.HDel(ctx, hashKey, uuids...).Err(); err != nil { + log.Errorf("persistence: HDel messages %s (%d): %v", queueKey, len(uuids), err) + } } -// PurgeMessagesPersist — удаляет ВСЕ сообщения очереди из Redis (для PurgeQueue). -// Удаляет весь HASH ssq:msg:{queueKey}. +// PurgeMessagesPersist — удаляет ВСЕ сообщения очереди из Redis СИНХРОННО. +// Синхронность обязательна: при рестарте пода удалённые сообщения не должны +// воскреснуть (факт: асинхронное удаление очередей приводило к их возврату). func PurgeMessagesPersist(queueKey string) { if Client == nil { return } hashKey := redisMsgHashPrefix + queueKey - asyncWrite(func() { - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - if err := Client.Del(ctx, hashKey).Err(); err != nil { - log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err) - } - }) + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if err := Client.Del(ctx, hashKey).Err(); err != nil { + log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err) + } } // DeleteQueue — удаляет метаданные очереди И все её сообщения из Redis асинхронно. @@ -242,21 +236,22 @@ func DeleteQueue(key string) { } writeSeq := nextQueueWriteSeq(key) msgHashKey := redisMsgHashPrefix + key - asyncWrite(func() { - if !isLatestQueueWriteSeq(key, writeSeq) { - return - } - ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) - defer cancel() - // Удаляем метаданные из общего HASH - if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil { - log.Errorf("persistence: HDel queue meta %q: %v", key, err) - } - // Удаляем весь HASH с сообщениями - if err := Client.Del(ctx, msgHashKey).Err(); err != nil { - log.Errorf("persistence: DEL messages hash %q: %v", key, err) - } - }) + // СИНХРОННО: удаление очереди обязано попасть в Redis до ответа клиенту. + // Асинхронное удаление приводило к «воскрешению» удалённых очередей после + // рестарта пода (факт 2026-08-15: 50 мусорных очередей вернулись из Redis). + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second) + defer cancel() + if !isLatestQueueWriteSeq(key, writeSeq) { + return + } + // Удаляем метаданные из общего HASH + if err := Client.HDel(ctx, redisHashQueues, key).Err(); err != nil { + log.Errorf("persistence: HDel queue meta %q: %v", key, err) + } + // Удаляем весь HASH с сообщениями + if err := Client.Del(ctx, msgHashKey).Err(); err != nil { + log.Errorf("persistence: DEL messages hash %q: %v", key, err) + } } // LoadAllQueues — загружает все очереди из Redis в память при старте сервиса.