fix: reduce persistence lag and retry audit deadlocks

server/store/store.go

@@ -3682,6 +3682,58 @@ type PreparedMessageBatchItem struct {
 	ReadBySender bool
 }
 
+type preparedMessageBatchTopicUpdate struct {
+	topic   string
+	message *types.Message
+}
+
+type preparedMessageBatchReadUpdate struct {
+	topic   string
+	from    types.Uid
+	message *types.Message
+}
+
+func aggregatePreparedMessageBatchUpdates(items []PreparedMessageBatchItem) ([]preparedMessageBatchTopicUpdate, []preparedMessageBatchReadUpdate) {
+	topics := make(map[string]*types.Message)
+	reads := make(map[string]preparedMessageBatchReadUpdate)
+	for _, item := range items {
+		msg := item.Message
+		if msg == nil {
+			continue
+		}
+		if previous := topics[msg.Topic]; previous == nil || msg.SeqId > previous.SeqId {
+			topics[msg.Topic] = msg
+		}
+		from := types.ParseUid(msg.From)
+		if !item.ReadBySender || from.IsZero() {
+			continue
+		}
+		key := msg.Topic + "\x00" + from.String()
+		if previous, ok := reads[key]; !ok || msg.SeqId > previous.message.SeqId {
+			reads[key] = preparedMessageBatchReadUpdate{topic: msg.Topic, from: from, message: msg}
+		}
+	}
+	topicKeys := make([]string, 0, len(topics))
+	for topic := range topics {
+		topicKeys = append(topicKeys, topic)
+	}
+	sort.Strings(topicKeys)
+	topicUpdates := make([]preparedMessageBatchTopicUpdate, 0, len(topicKeys))
+	for _, topic := range topicKeys {
+		topicUpdates = append(topicUpdates, preparedMessageBatchTopicUpdate{topic: topic, message: topics[topic]})
+	}
+	readKeys := make([]string, 0, len(reads))
+	for key := range reads {
+		readKeys = append(readKeys, key)
+	}
+	sort.Strings(readKeys)
+	readUpdates := make([]preparedMessageBatchReadUpdate, 0, len(readKeys))
+	for _, key := range readKeys {
+		readUpdates = append(readUpdates, reads[key])
+	}
+	return topicUpdates, readUpdates
+}
+
 func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessageBatchItem) error {
 	if len(items) == 0 {
 		return nil
@@ -3725,14 +3777,17 @@ func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessag
 	if err != nil {
 		return err
 	}
-	for _, item := range items {
-		msg := item.Message
-		from := types.ParseUid(msg.From)
-		if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil {
+	topicUpdates, readUpdates := aggregatePreparedMessageBatchUpdates(items)
+	for _, update := range topicUpdates {
+		if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, update.topic, update.message); err != nil {
 			return err
 		}
-		invalidateTopicCache(tenantID, msg.Topic)
-		if isGroup && groupMsgRuntime != nil {
+		invalidateTopicCache(tenantID, update.topic)
+	}
+	if isGroup && groupMsgRuntime != nil {
+		for _, item := range items {
+			msg := item.Message
+			from := types.ParseUid(msg.From)
 			if msg.ClientMsgID != "" {
 				_ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, from, msg.ClientMsgID, msg.SeqId, msg.Uid().String())
 			}
@@ -3741,13 +3796,13 @@ func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessag
 			})
 			_ = groupMsgRuntime.CacheHotMessage(tenantID, msg)
 		}
-		if item.ReadBySender && !from.IsZero() {
-			if updateErr := tenantAdp.TenantSubsUpdate(tenantID, msg.Topic, from, map[string]any{
-				"RecvSeqId": msg.SeqId, "ReadSeqId": msg.SeqId,
-			}); updateErr == nil {
-				invalidateSubscriptionCache(tenantID, msg.Topic, from)
-				DeleteCachedUserUnreadCount(tenantID, from)
-			}
+	}
+	for _, update := range readUpdates {
+		if updateErr := tenantAdp.TenantSubsUpdate(tenantID, update.topic, update.from, map[string]any{
+			"RecvSeqId": update.message.SeqId, "ReadSeqId": update.message.SeqId,
+		}); updateErr == nil {
+			invalidateSubscriptionCache(tenantID, update.topic, update.from)
+			DeleteCachedUserUnreadCount(tenantID, update.from)
 		}
 	}
 	return nil