From 31840b9756d0033c16227ef91ad6fda8af6427b3 Mon Sep 17 00:00:00 2001 From: Repinoid Date: Wed, 7 Oct 2026 07:18:31 +0300 Subject: [PATCH] =?UTF-8?q?fix:=20DLQ-=D1=80=D0=B5=D0=BB=D0=B8=D0=BD=D0=BA?= =?UTF-8?q?=20=D0=BF=D0=BE=D1=81=D0=BB=D0=B5=20=D0=B7=D0=B0=D0=B3=D1=80?= =?UTF-8?q?=D1=83=D0=B7=D0=BA=D0=B8=20=D0=B8=D0=B7=20Redis=20+=20=D0=B1?= =?UTF-8?q?=D0=B5=D0=B7=D0=BE=D0=BF=D0=B0=D1=81=D0=BD=D0=B0=D1=8F=20=D1=8D?= =?UTF-8?q?=D0=B2=D0=B8=D0=BA=D1=86=D0=B8=D1=8F=20retention=20+=20ParseFor?= =?UTF-8?q?m?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 зелёный. --- app/gosqs/gosqs.go | 11 +++++++---- app/models/models.go | 6 +++++- app/persistence/redis.go | 29 +++++++++++++++++++++++++---- app/router/router.go | 3 +++ 4 files changed, 40 insertions(+), 9 deletions(-) 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 {