persistence: порядковая нумерация async SaveMessage (старое состояние не перезапишет новое) + очистка при удалении

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