gosqs: персистить перенос в DLQ (SaveMessage в DLQ + DeleteMessagePersist исходной)

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])

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])
 				}

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)
 		}
 	}
 

app/gosqs/tenant_helpers.go

@@ -18,8 +18,15 @@ func tenantQueueKey(tenantAccessKey, queueName string) string {
 return tenantAccessKey + ":" + queueName
 }
 
-// getTenantFromContext — извлекает тенанта из request context.
+// dlqKeyFor — ключ DLQ по ключу исходной очереди (формат "{accessKey}:{name}")
-// Возвращает nil если тенант не найден (не должно быть — auth middleware должен это поймать раньше).
+// и имени 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

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-тика).