From db418789512765de27bf7eb55639b95f71f3fffa Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Tue, 6 Oct 2026 18:10:57 +0300
Subject: [PATCH] =?UTF-8?q?P1.3:=20=D1=80=D0=B5=D0=B0=D0=BB=D0=B8=D0=B7?=
 =?UTF-8?q?=D0=BE=D0=B2=D0=B0=D0=BD=20MessageRetentionPeriod=20(=D1=8D?=
 =?UTF-8?q?=D0=B2=D0=B8=D0=BA=D1=86=D0=B8=D1=8F=20=D0=BF=D1=80=D0=BE=D1=81?=
 =?UTF-8?q?=D1=80=D0=BE=D1=87=D0=B5=D0=BD=D0=BD=D1=8B=D1=85=20=D1=81=D0=BE?=
 =?UTF-8?q?=D0=BE=D0=B1=D1=89=D0=B5=D0=BD=D0=B8=D0=B9)?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

processQueueTick теперь эвиктит сообщения старше retention (SentTime + retention < now):
удаляет их из очереди и из Redis (DeleteMessagesPersist), разблокируя FIFO-группы.
Retention == 0 означает отсутствие истечения. PeriodicTasks итерируется по (key, queue),
чтобы передать queueKey для удаления из Redis. Устраняет забивание очередей до OverLimit.
---
 app/gosqs/gosqs.go | 45 ++++++++++++++++++++++++++++++++++++++-------
 1 file changed, 38 insertions(+), 7 deletions(-)

diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go
index 0ef89af..b13eb3f 100644
--- a/app/gosqs/gosqs.go
+++ b/app/gosqs/gosqs.go
@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"shared-sqs/app/models"
+	"shared-sqs/app/persistence"
 
 	log "github.com/sirupsen/logrus"
 )
@@ -30,21 +31,25 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
 	for {
 		select {
 		case <-ticker.C:
-			// Снапшот указателей на очереди под коротким RLock.
+			// Снапшот (key, указатель) очередей под коротким RLock.
 			// Указатели остаются валидными после снятия RLock: очереди из map
 			// не удаляются во время итерации, а объекты живут в куче.
 			models.SyncQueues.RLock()
-			queues := make([]*models.Queue, 0, len(models.SyncQueues.Queues))
-			for _, queue := range models.SyncQueues.Queues {
-				queues = append(queues, queue)
+			type qEntry struct {
+				key   string
+				queue *models.Queue
+			}
+			queues := make([]qEntry, 0, len(models.SyncQueues.Queues))
+			for key, queue := range models.SyncQueues.Queues {
+				queues = append(queues, qEntry{key, queue})
 			}
 			models.SyncQueues.RUnlock()
 
 			// Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной
 			// очереди не мешает обработке остальных и HTTP-запросам.
-			for _, queue := range queues {
+			for _, qi := range queues {
 				models.SyncQueues.Lock()
-				processQueueTick(queue)
+				processQueueTick(qi.key, qi.queue)
 				models.SyncQueues.Unlock()
 			}
 		case <-quit:
@@ -59,6 +64,7 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
 // ВЫЗЫВАТЬ ТОЛЬКО под удержанием SyncQueues.Lock.
 //
 // Логическая схема:
+//  0. Эвикция сообщений по MessageRetentionPeriod (старше retention удаляются).
 //  1. Deduplication-окно (FIFO): удаляем записи старше DeduplicationPeriod.
 //  2. Проходим по сообщениям очереди:
 //     - пустой ReceiptHandle (не in-flight) — пропускаем;
@@ -67,7 +73,32 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
 //     FIFO-группы, обновление ReceiptTime), инкрементируем Retry;
 //     - если Retry >= MaxReceiveCount и настроена DLQ — переносим сообщение
 //     в DLQ и удаляем из исходной очереди.
-func processQueueTick(queue *models.Queue) {
+func processQueueTick(key string, queue *models.Queue) {
+	// 0. Эвикция просроченных сообщений по MessageRetentionPeriod.
+	// Retention == 0 означает «без истечения» (не эвиктируем).
+	if queue.MessageRetentionPeriod > 0 {
+		cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second)
+		expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis
+		kept := queue.Messages[:0]
+		for i := range queue.Messages {
+			msg := &queue.Messages[i]
+			if msg.SentTime.Before(cutoff) {
+				expired = append(expired, msg.Uuid)
+				if msg.GroupID != "" {
+					queue.UnlockGroup(msg.GroupID)
+				}
+				continue
+			}
+			kept = append(kept, *msg)
+		}
+		if len(expired) > 0 {
+			queue.Messages = kept
+			log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired))
+			// Удаляем просроченные из Redis, чтобы не воскресли после рестарта.
+			persistence.DeleteMessagesPersist(key, expired)
+		}
+	}
+
 	// 1. Просроченные записи deduplication-окна.
 	for dedupId, startTime := range queue.Duplicates {
 		if time.Now().After(startTime.Add(models.DeduplicationPeriod)) {