package gosqs import ( "net/url" "time" "shared-sqs/app/models" "shared-sqs/app/persistence" log "github.com/sirupsen/logrus" ) func init() { models.SyncQueues.Queues = make(map[string]*models.Queue) } // PeriodicTasks — фоновая обработка очередей, выполняется раз в `d` (1 секунда). // // Задачи на каждый такт (для КАЖДОЙ очереди, см. processQueueTick): // 1. Очистка просроченных записей deduplication-окна (FIFO). // 2. Возврат сообщений в видимое состояние по истечении VisibilityTimeout. // 3. Перенос сообщений в DeadLetterQueue при достижении MaxReceiveCount. // // ВАЖНО (фикс производительности): раньше глобальный SyncQueues.Lock удерживался // на время обхода ВСЕХ очередей — при большом числе очередей HTTP-обработчики // блокировались на секунды (наблюдались ReadTimeout и latency до 54 секунд). // Теперь блокировка захватывается ПО ОДНОЙ очереди: между обработкой двух // очередей другие горутины могут вклиниться. func PeriodicTasks(d time.Duration, quit chan bool) { ticker := time.NewTicker(d) for { select { case <-ticker.C: // Снапшот (key, указатель) очередей под коротким RLock. // Указатели остаются валидными после снятия RLock: очереди из map // не удаляются во время итерации, а объекты живут в куче. models.SyncQueues.RLock() 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 _, qi := range queues { func() { // recover: паника в processQueueTick одной очереди не должна // убивать весь процесс; mutex освобождается в defer. defer qi.queue.UnlockMu() defer func() { if r := recover(); r != nil { log.Errorf("gosqs: panic in processQueueTick for queue %q: %v", qi.key, r) } }() qi.queue.LockMu() processQueueTick(qi.key, qi.queue) }() } case <-quit: ticker.Stop() return } } } // processQueueTick — обработка ОДНОЙ очереди за один такт. // // ВЫЗЫВАТЬ под per-queue блокировкой этой очереди (queue.LockMu). // // Логическая схема: // 0. Эвикция сообщений по MessageRetentionPeriod (старше retention удаляются). // 1. Deduplication-окно (FIFO): удаляем записи старше DeduplicationPeriod. // 2. Проходим по сообщениям очереди: // - пустой ReceiptHandle (не in-flight) — пропускаем; // - VisibilityTimeout ещё не истёк — пропускаем; // - иначе: возвращаем видимость (сброс ReceiptHandle, разблокировка // FIFO-группы, обновление ReceiptTime), инкрементируем Retry; // - если Retry >= MaxReceiveCount и настроена DLQ — переносим сообщение // в DLQ и удаляем из исходной очереди. func processQueueTick(key string, queue *models.Queue) { // Ленивая починка индексов (P1.2): очередь могла быть собрана литералом или // загружена из Redis — тогда индексы строятся здесь один раз. queue.EnsureIndexes() // 0. Эвикция просроченных сообщений по MessageRetentionPeriod. // Retention == 0 означает «без истечения» (не эвиктируем). // Удаление — tombstone за O(1); физическая чистка — CompactIfNeeded ниже. if queue.MessageRetentionPeriod > 0 { cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second) expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis for i := range queue.Messages { if queue.IsDead(i) { continue } msg := queue.Messages[i] if !msg.SentTime.Before(cutoff) { continue } expired = append(expired, msg.Uuid) if msg.GroupID != "" { queue.UnlockGroup(msg.GroupID) } if msg.ReceiptHandle != "" { queue.ClearInFlight(i, msg.ReceiptHandle) } queue.RemoveIndex(i) } if len(expired) > 0 { 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)) { log.Debugf("deduplication period for message with deduplicationId [%s] expired", dedupId) delete(queue.Duplicates, dedupId) } } // 2. Возврат видимости истёкших сообщений + перенос в DLQ. // Итерируем ТОЛЬКО по in-flight (receiptIdx), а не по всему срезу — стоимость // пропорциональна числу сообщений «в полёте», а не размеру очереди (P1.2). log.Debugf("Queue [%s] live [%d]", queue.Name, queue.LiveCount()) now := time.Now() for _, i := range queue.InFlightIndexes() { if i < 0 || i >= len(queue.Messages) || queue.IsDead(i) { continue } msg := &queue.Messages[i] rh := msg.ReceiptHandle if rh == "" { continue // уже возвращено в видимость } if msg.VisibilityTimeout.After(now) { continue // таймаут видимости ещё действует } log.Debugf("Making message visible again %s", rh) queue.ClearInFlight(i, rh) queue.UnlockGroup(msg.GroupID) msg.ReceiptHandle = "" msg.ReceiptTime = time.Now().UTC() msg.Retry++ // 3. Перенос в DLQ при превышении MaxReceiveCount. if queue.MaxReceiveCount > 0 && queue.DeadLetterQueue != nil && msg.Retry >= queue.MaxReceiveCount { dlqMsg := *msg msgUuid := msg.Uuid dlqName := queue.DeadLetterQueue.Name queue.DeadLetterQueue.LockMu() queue.DeadLetterQueue.AppendMessage(dlqMsg) queue.DeadLetterQueue.UnlockMu() queue.RemoveIndex(i) // Персист переноса: иначе после рестарта сообщение «воскреснет» // в исходной очереди (не удалено из Redis) и пропадёт из DLQ. _ = persistence.DeleteMessagePersist(key, msgUuid) persistence.SaveMessage(dlqKeyFor(key, dlqName), &dlqMsg) } } // 4. Компакция tombstone-слотов (амортизированно): срез не растёт бесконечно. queue.CompactIfNeeded() } func numberOfHiddenMessagesInQueue(queue *models.Queue) int { num := 0 for i := range queue.Messages { if queue.IsDead(i) { continue } m := &queue.Messages[i] if m.ReceiptHandle != "" || m.DelaySecs > 0 && time.Now().Before(m.SentTime.Add(time.Duration(m.DelaySecs)*time.Second)) { num++ } } return num } func getQueueFromPath(formVal string, theUrl string) string { if formVal != "" { return formVal } u, err := url.Parse(theUrl) if err != nil { return "" } return u.Path }