From efbfb7556a2a40dc90f06d456d2bd5a4a8113f7d Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Fri, 4 Sep 2026 20:14:44 +0800 Subject: [PATCH] Fix P2P message duplicate seq retry --- server/store/store.go | 36 ++++++++++++++++++++++++++++++++---- 1 file changed, 32 insertions(+), 4 deletions(-) diff --git a/server/store/store.go b/server/store/store.go index 108ee78..1706a38 100644 --- a/server/store/store.go +++ b/server/store/store.go @@ -3351,9 +3351,6 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag err = tenantAdp.TenantGroupMessageSave(tenantID, msg) } else { err = tenantAdp.TenantMessageSave(tenantID, msg) - if err == types.ErrDuplicate && clientMsgID == "" { - err = retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg) - } } if err != nil { if isGroupMessage && err == types.ErrDuplicate && clientMsgID != "" { @@ -3369,10 +3366,28 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil } } + if !isGroupMessage && err == types.ErrDuplicate { + if clientMsgID != "" { + duplicate, dupErr := tenantAdp.TenantMessageGetByClientID(tenantID, fromUid, clientMsgID) + if dupErr != nil { + return nil, dupErr + } + if duplicate != nil { + copyPersistedMessage(msg, duplicate) + return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil + } + } + if retryErr := retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg); retryErr != nil { + return nil, retryErr + } + err = nil + } if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" { _ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID) } - return nil, err + if err != nil { + return nil, err + } } if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil { return nil, err @@ -3438,6 +3453,19 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag return &types.MessageSaveResult{SeqId: msg.SeqId, MarkedReadBySender: markedReadBySender}, nil } +func copyPersistedMessage(dst, src *types.Message) { + dst.ObjHeader = src.ObjHeader + dst.DeletedAt = src.DeletedAt + dst.DelId = src.DelId + dst.DeletedFor = src.DeletedFor + dst.SeqId = src.SeqId + dst.Topic = src.Topic + dst.From = src.From + dst.ClientMsgID = src.ClientMsgID + dst.Head = src.Head + dst.Content = src.Content +} + func PrepareMessagePersist(tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) { return Messages.PreparePersist(tenantID, msg, attachmentURLs, readBySender) }