v0.1.35: фиксы — синхронные удаления в Redis (воскрешение очередей) + RedrivePolicy tenant-scoped DLQ

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)
 		}

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")

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)

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 в память при старте сервиса.