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