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