diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index 85d721d..2bda081 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -79,9 +79,11 @@ func processQueueTick(key string, queue *models.Queue) { if queue.MessageRetentionPeriod > 0 { cutoff := time.Now().Add(-time.Duration(queue.MessageRetentionPeriod) * time.Second) expired := make([]string, 0) // uuid просроченных сообщений для удаления из Redis - kept := queue.Messages[:0] + // Компакция НА МЕСТЕ: читаем значение в i, пишем в n (n <= i) — без + // алиасинга на переиспользуемом срезе и без лишней аллокации. + n := 0 for i := range queue.Messages { - msg := &queue.Messages[i] + msg := queue.Messages[i] if msg.SentTime.Before(cutoff) { expired = append(expired, msg.Uuid) if msg.GroupID != "" { @@ -89,10 +91,11 @@ func processQueueTick(key string, queue *models.Queue) { } continue } - kept = append(kept, *msg) + queue.Messages[n] = msg + n++ } 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)) // Удаляем просроченные из Redis, чтобы не воскресли после рестарта. persistence.DeleteMessagesPersist(key, expired) diff --git a/app/models/models.go b/app/models/models.go index 16ddf31..4ef67f3 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -62,7 +62,11 @@ type Queue struct { MessageRetentionPeriod int // seconds (реализовано в processQueueTick) Messages []SqsMessage DeadLetterQueue *Queue - MaxReceiveCount int + // DeadLetterQueueName — имя DLQ. В Redis-метаданных хранится ТОЛЬКО имя + // (а не вложенный объект): после загрузки указатель DeadLetterQueue + // релинкуется на объект из общей map (см. persistence.LoadAllQueues). + DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"` + MaxReceiveCount int IsFIFO bool FIFOMessages map[string]int FIFOSequenceNumbers map[string]int diff --git a/app/persistence/redis.go b/app/persistence/redis.go index a96c9d0..7ac89ae 100644 --- a/app/persistence/redis.go +++ b/app/persistence/redis.go @@ -18,6 +18,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "sync" "time" @@ -163,11 +164,12 @@ type queueMetaDTO struct { Duplicates map[string]time.Time Tags map[string]string LastPurgeTime time.Time - DeadLetterQueue *queueMetaDTO + DeadLetterQueueName string `json:"dead_letter_queue_name,omitempty"` } -// queueToMetaDTO — строит снимок очереди без сообщений (и без сообщений DLQ), -// НЕ мутируя исходную очередь. Рекурсия ограничена глубиной 1. +// queueToMetaDTO — строит снимок очереди без сообщений, НЕ мутируя исходную очередь. +// DLQ сохраняется как ИМЯ (не вложенный объект): иначе — detached-копия после +// загрузки и неограниченная рекурсия при циклической DLQ. func queueToMetaDTO(q *models.Queue) *queueMetaDTO { if q == nil { return nil @@ -189,9 +191,10 @@ func queueToMetaDTO(q *models.Queue) *queueMetaDTO { Duplicates: q.Duplicates, Tags: q.Tags, LastPurgeTime: q.LastPurgeTime, + DeadLetterQueueName: q.DeadLetterQueueName, } if q.DeadLetterQueue != nil { - dto.DeadLetterQueue = queueToMetaDTO(q.DeadLetterQueue) + dto.DeadLetterQueueName = q.DeadLetterQueue.Name } 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)) return queues, nil } diff --git a/app/router/router.go b/app/router/router.go index 8eecf85..322622f 100644 --- a/app/router/router.go +++ b/app/router/router.go @@ -212,6 +212,9 @@ func countBatchEntries(req *http.Request, action string) int { } else if action == "ChangeMessageVisibilityBatch" { prefix = "ChangeMessageVisibilityBatchRequestEntry." } + // Разбираем form, если ещё не разобран (иначе PostForm пуст для lazy-запросов). + _ = req.ParseForm() + if req.PostForm != nil || req.Form != nil { form := req.Form if form == nil {