fix: DLQ-релинк после загрузки из Redis + безопасная эвикция retention + ParseForm
1. DLQ больше не сериализуется вложенным объектом (была detached-копия после загрузки + неограниченная рекурсия при цикле): в метаданных хранится только имя (dead_letter_queue_name), после LoadAllQueues указатель релинкуется на объект из map. Закрывает потерянные переносы в DLQ и чужой мьютекс. 2. Retention-эвикция: компакция НА МЕСТЕ (чтение value в i, запись в n<=i) вместо kept := Messages[:0] — устранён алиасинг на переиспользуемом срезе. 3. countBatchEntries: явный req.ParseForm() — корректный msg_count для батчей. Тесты: +DLQ-перенос, +эвикция с сохранением порядка; go test -race зелёный.
This commit is contained in:
+7
-4
@@ -79,9 +79,11 @@ func processQueueTick(key string, queue *models.Queue) {
|
|||||||
if queue.MessageRetentionPeriod > 0 {
|
if queue.MessageRetentionPeriod > 0 {
|
||||||
cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second)
|
cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second)
|
||||||
expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis
|
expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis
|
||||||
kept := queue.Messages[:0]
|
// Компакция НА МЕСТЕ: читаем значение в i, пишем в n (n <= i) — без
|
||||||
|
// алиасинга на переиспользуемом срезе и без лишней аллокации.
|
||||||
|
n := 0
|
||||||
for i := range queue.Messages {
|
for i := range queue.Messages {
|
||||||
msg := &queue.Messages[i]
|
msg := queue.Messages[i]
|
||||||
if msg.SentTime.Before(cutoff) {
|
if msg.SentTime.Before(cutoff) {
|
||||||
expired = append(expired, msg.Uuid)
|
expired = append(expired, msg.Uuid)
|
||||||
if msg.GroupID != "" {
|
if msg.GroupID != "" {
|
||||||
@@ -89,10 +91,11 @@ func processQueueTick(key string, queue *models.Queue) {
|
|||||||
}
|
}
|
||||||
continue
|
continue
|
||||||
}
|
}
|
||||||
kept = append(kept, *msg)
|
queue.Messages[n] = msg
|
||||||
|
n++
|
||||||
}
|
}
|
||||||
if len(expired) > 0 {
|
if len(expired) > 0 {
|
||||||
queue.Messages = kept
|
queue.Messages = queue.Messages[:n]
|
||||||
log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired))
|
log.Infof("Queue [%s]: evicted %d messages by MessageRetentionPeriod", queue.Name, len(expired))
|
||||||
// Удаляем просроченные из Redis, чтобы не воскресли после рестарта.
|
// Удаляем просроченные из Redis, чтобы не воскресли после рестарта.
|
||||||
persistence.DeleteMessagesPersist(key, expired)
|
persistence.DeleteMessagesPersist(key, expired)
|
||||||
|
|||||||
@@ -62,6 +62,10 @@ type Queue struct {
|
|||||||
MessageRetentionPeriod int // seconds (реализовано в processQueueTick)
|
MessageRetentionPeriod int // seconds (реализовано в processQueueTick)
|
||||||
Messages []SqsMessage
|
Messages []SqsMessage
|
||||||
DeadLetterQueue *Queue
|
DeadLetterQueue *Queue
|
||||||
|
// DeadLetterQueueName — имя DLQ. В Redis-метаданных хранится ТОЛЬКО имя
|
||||||
|
// (а не вложенный объект): после загрузки указатель DeadLetterQueue
|
||||||
|
// релинкуется на объект из общей map (см. persistence.LoadAllQueues).
|
||||||
|
DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"`
|
||||||
MaxReceiveCount int
|
MaxReceiveCount int
|
||||||
IsFIFO bool
|
IsFIFO bool
|
||||||
FIFOMessages map[string]int
|
FIFOMessages map[string]int
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ import (
|
|||||||
"context"
|
"context"
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"strings"
|
||||||
"sync"
|
"sync"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
@@ -163,11 +164,12 @@ type queueMetaDTO struct {
|
|||||||
Duplicates map[string]time.Time
|
Duplicates map[string]time.Time
|
||||||
Tags map[string]string
|
Tags map[string]string
|
||||||
LastPurgeTime time.Time
|
LastPurgeTime time.Time
|
||||||
DeadLetterQueue *queueMetaDTO
|
DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"`
|
||||||
}
|
}
|
||||||
|
|
||||||
// queueToMetaDTO — строит снимок очереди без сообщений (и без сообщений DLQ),
|
// queueToMetaDTO — строит снимок очереди без сообщений, НЕ мутируя исходную очередь.
|
||||||
// НЕ мутируя исходную очередь. Рекурсия ограничена глубиной 1.
|
// DLQ сохраняется как ИМЯ (не вложенный объект): иначе — detached-копия после
|
||||||
|
// загрузки и неограниченная рекурсия при циклической DLQ.
|
||||||
func queueToMetaDTO(q *models.Queue) *queueMetaDTO {
|
func queueToMetaDTO(q *models.Queue) *queueMetaDTO {
|
||||||
if q == nil {
|
if q == nil {
|
||||||
return nil
|
return nil
|
||||||
@@ -189,9 +191,10 @@ func queueToMetaDTO(q *models.Queue) *queueMetaDTO {
|
|||||||
Duplicates: q.Duplicates,
|
Duplicates: q.Duplicates,
|
||||||
Tags: q.Tags,
|
Tags: q.Tags,
|
||||||
LastPurgeTime: q.LastPurgeTime,
|
LastPurgeTime: q.LastPurgeTime,
|
||||||
|
DeadLetterQueueName: q.DeadLetterQueueName,
|
||||||
}
|
}
|
||||||
if q.DeadLetterQueue != nil {
|
if q.DeadLetterQueue != nil {
|
||||||
dto.DeadLetterQueue = queueToMetaDTO(q.DeadLetterQueue)
|
dto.DeadLetterQueueName = q.DeadLetterQueue.Name
|
||||||
}
|
}
|
||||||
return dto
|
return dto
|
||||||
}
|
}
|
||||||
@@ -481,6 +484,24 @@ func LoadAllQueues() (map[string]*models.Queue, error) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// 4. Релинк DLQ: в метаданных хранится только имя DLQ — связываем указатель
|
||||||
|
// с реальным объектом из map. Иначе DeadLetterQueue оказался бы detached-копией
|
||||||
|
// (переносы в DLQ писали бы в неё, а не в живую очередь; мьютекс — чужой).
|
||||||
|
for k, q := range queues {
|
||||||
|
if q.DeadLetterQueueName == "" {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
idx := strings.Index(k, ":")
|
||||||
|
if idx <= 0 {
|
||||||
|
continue
|
||||||
|
}
|
||||||
|
if dlq, ok := queues[k[:idx+1]+q.DeadLetterQueueName]; ok {
|
||||||
|
q.DeadLetterQueue = dlq
|
||||||
|
} else {
|
||||||
|
log.Warnf("persistence: DLQ %q для очереди %q не найдена при загрузке", q.DeadLetterQueueName, k)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
log.Infof("persistence: загружено %d очередей из Redis (миграций: %d)", len(queues), len(migrateKeys))
|
log.Infof("persistence: загружено %d очередей из Redis (миграций: %d)", len(queues), len(migrateKeys))
|
||||||
return queues, nil
|
return queues, nil
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -212,6 +212,9 @@ func countBatchEntries(req *http.Request, action string) int {
|
|||||||
} else if action == "ChangeMessageVisibilityBatch" {
|
} else if action == "ChangeMessageVisibilityBatch" {
|
||||||
prefix = "ChangeMessageVisibilityBatchRequestEntry."
|
prefix = "ChangeMessageVisibilityBatchRequestEntry."
|
||||||
}
|
}
|
||||||
|
// Разбираем form, если ещё не разобран (иначе PostForm пуст для lazy-запросов).
|
||||||
|
_ = req.ParseForm()
|
||||||
|
|
||||||
if req.PostForm != nil || req.Form != nil {
|
if req.PostForm != nil || req.Form != nil {
|
||||||
form := req.Form
|
form := req.Form
|
||||||
if form == nil {
|
if form == nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user