diff --git a/app/gosqs/change_message_visibility.go b/app/gosqs/change_message_visibility.go index d466889..cee0deb 100644 --- a/app/gosqs/change_message_visibility.go +++ b/app/gosqs/change_message_visibility.go @@ -56,12 +56,13 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo messageFound := false var changedMsgUuid string var msgRemoved bool + var dlqMsg *models.SqsMessage // O(1)-поиск по индексу ReceiptHandle (P1.2) вместо скана всего среза. if i, ok := queue.IndexOfReceipt(receiptHandle); ok { if visibilityTimeout == 0 { // Возврат видимости (+ DLQ-перенос при исчерпании MaxReceiveCount) — // единый путь с батчевой версией. - changedMsgUuid, msgRemoved = queue.ResetVisibilityAndTryDLQ(i, receiptHandle) + changedMsgUuid, msgRemoved, dlqMsg = queue.ResetVisibilityAndTryDLQ(i, receiptHandle) } else { queue.Messages[i].VisibilityTimeout = time.Now().Add(time.Duration(visibilityTimeout) * time.Second) changedMsgUuid = queue.Messages[i].Uuid @@ -71,8 +72,12 @@ func ChangeMessageVisibilityV1(req *http.Request) (int, interfaces.AbstractRespo // Персистим изменения в Redis под Lock (сетевые записи асинхронны). if messageFound { if msgRemoved { - // Сообщение удалено (перемещено в DLQ) — удаляем из Redis + // Сообщение удалено (перемещено в DLQ) — удаляем из Redis исходной + // очереди и сохраняем в DLQ, иначе после рестарта оно пропадёт из DLQ. persistence.DeleteMessagePersist(key, changedMsgUuid) + if dlqMsg != nil { + persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg) + } } else if i, ok := queue.IndexOfUUID(changedMsgUuid); ok { // Сообщение обновлено — перезаписываем в Redis persistence.SaveMessage(key, &queue.Messages[i]) diff --git a/app/gosqs/change_message_visibility_batch.go b/app/gosqs/change_message_visibility_batch.go index 66ad0ff..64a1a4a 100644 --- a/app/gosqs/change_message_visibility_batch.go +++ b/app/gosqs/change_message_visibility_batch.go @@ -86,10 +86,14 @@ func ChangeMessageVisibilityBatchV1(req *http.Request) (int, interfaces.Abstract // Сброс + DLQ-перенос при исчерпании MaxReceiveCount — единый путь // с одиночной версией (раньше батч лишь инкрементировал Retry и не // переносил в DLQ — счётчик и перенос могли расходиться). - uuid, moved := queue.ResetVisibilityAndTryDLQ(i, entry.ReceiptHandle) + uuid, moved, dlqMsg := queue.ResetVisibilityAndTryDLQ(i, entry.ReceiptHandle) if moved { - // Сообщение перенесено в DLQ и удалено из очереди — чистим Redis. + // Сообщение перенесено в DLQ и удалено из очереди — чистим Redis + // исходной и сохраняем в DLQ (иначе пропадёт после рестарта). persistence.DeleteMessagePersist(key, uuid) + if dlqMsg != nil { + persistence.SaveMessage(dlqKeyFor(key, queue.DeadLetterQueue.Name), dlqMsg) + } } else if j, ok := queue.IndexOfUUID(uuid); ok { persistence.SaveMessage(key, &queue.Messages[j]) } diff --git a/app/gosqs/gosqs.go b/app/gosqs/gosqs.go index abe8af7..c032c03 100644 --- a/app/gosqs/gosqs.go +++ b/app/gosqs/gosqs.go @@ -146,10 +146,16 @@ func processQueueTick(key string, queue *models.Queue) { 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) } } diff --git a/app/gosqs/tenant_helpers.go b/app/gosqs/tenant_helpers.go index 6b57557..06ee520 100644 --- a/app/gosqs/tenant_helpers.go +++ b/app/gosqs/tenant_helpers.go @@ -18,8 +18,15 @@ func tenantQueueKey(tenantAccessKey, queueName string) string { return tenantAccessKey + ":" + queueName } -// getTenantFromContext — извлекает тенанта из request context. -// Возвращает nil если тенант не найден (не должно быть — auth middleware должен это поймать раньше). +// dlqKeyFor — ключ DLQ по ключу исходной очереди (формат "{accessKey}:{name}") +// и имени DLQ. Используется для персиста перенесённых в DLQ сообщений. +func dlqKeyFor(srcKey, dlqName string) string { + idx := strings.LastIndex(srcKey, ":") + if idx < 0 { + return srcKey + ":" + dlqName + } + return srcKey[:idx] + ":" + dlqName +} func getTenantFromContext(r *http.Request) *tenant.Tenant { t, _ := r.Context().Value(auth.TenantContextKey).(*tenant.Tenant) return t diff --git a/app/models/models.go b/app/models/models.go index aedf6c3..3e4649e 100644 --- a/app/models/models.go +++ b/app/models/models.go @@ -405,7 +405,17 @@ func (q *Queue) ClearInFlight(i int, receiptHandle string) { // этого retry-счётчик и DLQ-перенос могли расходиться. // Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой. // Возвращает: msgUuid, movedToDLQ. -func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid string, movedToDLQ bool) { +// ResetVisibilityAndTryDLQ — сбрасывает видимость сообщения по индексу и, если +// исчерпан MaxReceiveCount, переносит его в DLQ. При переносе удаляет сообщение +// из исходной очереди (tombstone). +// +// Единый путь для одиночной и батчевой ChangeMessageVisibility: раньше логика +// DLQ была только в одиночной версии, а батч лишь инкрементировал Retry — из-за +// этого retry-счётчик и DLQ-перенос могли расходиться. +// Вызывать под write-блокировкой q. Перенос в DLQ выполняется под её блокировкой. +// Возвращает: msgUuid, movedToDLQ, dlqMsg (копия перенесённого сообщения, чтобы +// вызывающий мог персистить его в DLQ; nil если переноса не было). +func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid string, movedToDLQ bool, dlqMsg *SqsMessage) { q.ensureIndexes() q.ClearInFlight(i, receiptHandle) msg := &q.Messages[i] @@ -420,9 +430,9 @@ func (q *Queue) ResetVisibilityAndTryDLQ(i int, receiptHandle string) (msgUuid s q.DeadLetterQueue.AppendMessage(dlqMsg) q.DeadLetterQueue.UnlockMu() q.RemoveIndex(i) - return msgUuid, true + return msgUuid, true, &dlqMsg } - return msgUuid, false + return msgUuid, false, nil } // InFlightIndexes — снимок индексов in-flight сообщений (для visibility-тика).