P1.3: реализован MessageRetentionPeriod (эвикция просроченных сообщений)
processQueueTick теперь эвиктит сообщения старше retention (SentTime + retention < now): удаляет их из очереди и из Redis (DeleteMessagesPersist), разблокируя FIFO-группы. Retention == 0 означает отсутствие истечения. PeriodicTasks итерируется по (key, queue), чтобы передать queueKey для удаления из Redis. Устраняет забивание очередей до OverLimit.
This commit is contained in:
+38
-7
@@ -5,6 +5,7 @@ import (
|
|||||||
"time"
|
"time"
|
||||||
|
|
||||||
"shared-sqs/app/models"
|
"shared-sqs/app/models"
|
||||||
|
"shared-sqs/app/persistence"
|
||||||
|
|
||||||
log "github.com/sirupsen/logrus"
|
log "github.com/sirupsen/logrus"
|
||||||
)
|
)
|
||||||
@@ -30,21 +31,25 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
|
|||||||
for {
|
for {
|
||||||
select {
|
select {
|
||||||
case <-ticker.C:
|
case <-ticker.C:
|
||||||
// Снапшот указателей на очереди под коротким RLock.
|
// Снапшот (key, указатель) очередей под коротким RLock.
|
||||||
// Указатели остаются валидными после снятия RLock: очереди из map
|
// Указатели остаются валидными после снятия RLock: очереди из map
|
||||||
// не удаляются во время итерации, а объекты живут в куче.
|
// не удаляются во время итерации, а объекты живут в куче.
|
||||||
models.SyncQueues.RLock()
|
models.SyncQueues.RLock()
|
||||||
queues := make([]*models.Queue, 0, len(models.SyncQueues.Queues))
|
type qEntry struct {
|
||||||
for _, queue := range models.SyncQueues.Queues {
|
key string
|
||||||
queues = append(queues, queue)
|
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()
|
models.SyncQueues.RUnlock()
|
||||||
|
|
||||||
// Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной
|
// Каждую очередь обрабатываем под ОТДЕЛЬНЫМ Lock — блокировка одной
|
||||||
// очереди не мешает обработке остальных и HTTP-запросам.
|
// очереди не мешает обработке остальных и HTTP-запросам.
|
||||||
for _, queue := range queues {
|
for _, qi := range queues {
|
||||||
models.SyncQueues.Lock()
|
models.SyncQueues.Lock()
|
||||||
processQueueTick(queue)
|
processQueueTick(qi.key, qi.queue)
|
||||||
models.SyncQueues.Unlock()
|
models.SyncQueues.Unlock()
|
||||||
}
|
}
|
||||||
case <-quit:
|
case <-quit:
|
||||||
@@ -59,6 +64,7 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
|
|||||||
// ВЫЗЫВАТЬ ТОЛЬКО под удержанием SyncQueues.Lock.
|
// ВЫЗЫВАТЬ ТОЛЬКО под удержанием SyncQueues.Lock.
|
||||||
//
|
//
|
||||||
// Логическая схема:
|
// Логическая схема:
|
||||||
|
// 0. Эвикция сообщений по MessageRetentionPeriod (старше retention удаляются).
|
||||||
// 1. Deduplication-окно (FIFO): удаляем записи старше DeduplicationPeriod.
|
// 1. Deduplication-окно (FIFO): удаляем записи старше DeduplicationPeriod.
|
||||||
// 2. Проходим по сообщениям очереди:
|
// 2. Проходим по сообщениям очереди:
|
||||||
// - пустой ReceiptHandle (не in-flight) — пропускаем;
|
// - пустой ReceiptHandle (не in-flight) — пропускаем;
|
||||||
@@ -67,7 +73,32 @@ func PeriodicTasks(d time.Duration, quit chan bool) {
|
|||||||
// FIFO-группы, обновление ReceiptTime), инкрементируем Retry;
|
// FIFO-группы, обновление ReceiptTime), инкрементируем Retry;
|
||||||
// - если Retry >= MaxReceiveCount и настроена DLQ — переносим сообщение
|
// - если Retry >= MaxReceiveCount и настроена DLQ — переносим сообщение
|
||||||
// в 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-окна.
|
// 1. Просроченные записи deduplication-окна.
|
||||||
for dedupId, startTime := range queue.Duplicates {
|
for dedupId, startTime := range queue.Duplicates {
|
||||||
if time.Now().After(startTime.Add(models.DeduplicationPeriod)) {
|
if time.Now().After(startTime.Add(models.DeduplicationPeriod)) {
|
||||||
|
|||||||
Reference in New Issue
Block a user