From f99f49bd8e57370aadde17e515101c6d60195db4 Mon Sep 17 00:00:00 2001
From: Repinoid <tazet@narod.ru>
Date: Sun, 11 Oct 2026 09:12:42 +0300
Subject: [PATCH] =?UTF-8?q?persistence:=20=D0=BF=D0=BE=D1=80=D1=8F=D0=B4?=
 =?UTF-8?q?=D0=BA=D0=BE=D0=B2=D0=B0=D1=8F=20=D0=BD=D1=83=D0=BC=D0=B5=D1=80?=
 =?UTF-8?q?=D0=B0=D1=86=D0=B8=D1=8F=20async=20SaveMessage=20(=D1=81=D1=82?=
 =?UTF-8?q?=D0=B0=D1=80=D0=BE=D0=B5=20=D1=81=D0=BE=D1=81=D1=82=D0=BE=D1=8F?=
 =?UTF-8?q?=D0=BD=D0=B8=D0=B5=20=D0=BD=D0=B5=20=D0=BF=D0=B5=D1=80=D0=B5?=
 =?UTF-8?q?=D0=B7=D0=B0=D0=BF=D0=B8=D1=88=D0=B5=D1=82=20=D0=BD=D0=BE=D0=B2?=
 =?UTF-8?q?=D0=BE=D0=B5)=20+=20=D0=BE=D1=87=D0=B8=D1=81=D1=82=D0=BA=D0=B0?=
 =?UTF-8?q?=20=D0=BF=D1=80=D0=B8=20=D1=83=D0=B4=D0=B0=D0=BB=D0=B5=D0=BD?=
 =?UTF-8?q?=D0=B8=D0=B8?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 app/persistence/redis.go | 48 ++++++++++++++++++++++++++++++++++++++++
 1 file changed, 48 insertions(+)

diff --git a/app/persistence/redis.go b/app/persistence/redis.go
index 4350ece..3d2d8cf 100644
--- a/app/persistence/redis.go
+++ b/app/persistence/redis.go
@@ -46,6 +46,11 @@ var (
 	seqMu          sync.Mutex
 	queueWriteSeq  = map[string]uint64{}
 	tenantWriteSeq = map[string]uint64{}
+	// msgWriteSeq — порядковая нумерация async-записей ОТДЕЛЬНЫХ сообщений
+	// (ключ "{queueKey}|{uuid}"). Без неё два SaveMessage одного сообщения из
+	// разных горутин могли завершиться вразнобой: старое состояние перезаписало
+	// бы новое. Записи удаляются при удалении сообщения (DeleteMessagePersist и др.).
+	msgWriteSeq = map[string]uint64{}
 )
 
 const (
@@ -260,7 +265,12 @@ func SaveMessage(queueKey string, msg *models.SqsMessage) {
 		return
 	}
 	hashKey := redisMsgHashPrefix + queueKey
+	msgKey := queueKey + "|" + msg.Uuid
+	writeSeq := nextMsgWriteSeq(msgKey)
 	asyncWrite(func() {
+		if !isLatestMsgWriteSeq(msgKey, writeSeq) {
+			return
+		}
 		ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
 		defer cancel()
 		if err := Client.Load().HSet(ctx, hashKey, msg.Uuid, string(data)).Err(); err != nil {
@@ -371,6 +381,7 @@ func DeleteMessagePersist(queueKey string, msgUuid string) error {
 	if err := Client.Load().HDel(ctx, hashKey, msgUuid).Err(); err != nil {
 		return fmt.Errorf("HDel message %s/%s: %w", queueKey, msgUuid, err)
 	}
+	clearMsgWriteSeq(queueKey, msgUuid)
 	return nil
 }
 
@@ -386,6 +397,7 @@ func DeleteMessagesPersist(queueKey string, uuids []string) error {
 	if err := Client.Load().HDel(ctx, hashKey, uuids...).Err(); err != nil {
 		return fmt.Errorf("HDel messages %s (%d): %w", queueKey, len(uuids), err)
 	}
+	clearMsgWriteSeq(queueKey, uuids...)
 	return nil
 }
 
@@ -402,6 +414,7 @@ func PurgeMessagesPersist(queueKey string) {
 	if err := Client.Load().Del(ctx, hashKey).Err(); err != nil {
 		log.Errorf("persistence: DEL messages hash %s: %v", queueKey, err)
 	}
+	clearAllMsgWriteSeq(queueKey)
 }
 
 // DeleteQueue — удаляет метаданные очереди И все её сообщения из Redis асинхронно.
@@ -653,6 +666,41 @@ func isLatestQueueWriteSeq(key string, seq uint64) bool {
 	return queueWriteSeq[key] == seq
 }
 
+func nextMsgWriteSeq(key string) uint64 {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	msgWriteSeq[key]++
+	return msgWriteSeq[key]
+}
+
+func isLatestMsgWriteSeq(key string, seq uint64) bool {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	return msgWriteSeq[key] == seq
+}
+
+// clearMsgWriteSeq — удаляет записи нумерации для удалённых сообщений
+// (иначе msgWriteSeq рос бы бесконечно).
+func clearMsgWriteSeq(queueKey string, uuids ...string) {
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	for _, u := range uuids {
+		delete(msgWriteSeq, queueKey+"|"+u)
+	}
+}
+
+// clearAllMsgWriteSeq — удаляет ВСЕ записи нумерации для очереди (purge).
+func clearAllMsgWriteSeq(queueKey string) {
+	prefix := queueKey + "|"
+	seqMu.Lock()
+	defer seqMu.Unlock()
+	for k := range msgWriteSeq {
+		if strings.HasPrefix(k, prefix) {
+			delete(msgWriteSeq, k)
+		}
+	}
+}
+
 func nextTenantWriteSeq(id string) uint64 {
 	seqMu.Lock()
 	defer seqMu.Unlock()