From fa72d06d80f7895057adabe9e45f5b20c6c55b34 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Thu, 8 Jun 2023 19:30:16 +0800 Subject: [PATCH 1/6] del list cache --- pkg/common/db/cache/msg.go | 22 ++++++++++++++++------ 1 file changed, 16 insertions(+), 6 deletions(-) diff --git a/pkg/common/db/cache/msg.go b/pkg/common/db/cache/msg.go index e13861b80..be44f36f4 100644 --- a/pkg/common/db/cache/msg.go +++ b/pkg/common/db/cache/msg.go @@ -383,13 +383,23 @@ func (c *msgCache) DelUserDeleteMsgsList(ctx context.Context, conversationID str log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) continue } - for _, userID := range delUsers { - if err := c.rdb.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err(); err != nil { - log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq, "userID", userID) + if len(delUsers) > 0 { + pipe := c.rdb.Pipeline() + var failedFlag bool + for _, userID := range delUsers { + err = pipe.SRem(ctx, c.getUserDelList(conversationID, userID), seq).Err() + if err != nil { + failedFlag = true + log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq, "userID", userID) + } } - - if err := c.rdb.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil { - log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) + if !failedFlag { + if err := pipe.Del(ctx, c.getMessageDelUserListKey(conversationID, seq)).Err(); err != nil { + log.ZWarn(ctx, "DelUserDeleteMsgsList failed", err, "conversationID", conversationID, "seq", seq) + } + } + if _, err := pipe.Exec(ctx); err != nil { + log.ZError(ctx, "pipe exec failed", err, "conversationID", conversationID, "seq", seq) } } } From 331d90f198e9bf84144df07fc780edffa28ca0a8 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Thu, 8 Jun 2023 19:52:11 +0800 Subject: [PATCH 2/6] fix del --- pkg/common/db/controller/msg.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index 5f4ad098c..2ca70f882 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -681,7 +681,7 @@ func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID st for _, msg := range msgs { cacheSeqs = append(cacheSeqs, msg.Seq) } - if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil { + if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil && errs.Unwrap(err) != redis.Nil { return err } From 89a1bf47426e2295f09bb46091b46197cb71eef9 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Thu, 8 Jun 2023 19:55:08 +0800 Subject: [PATCH 3/6] del --- pkg/common/db/controller/msg.go | 18 ++++++++++-------- 1 file changed, 10 insertions(+), 8 deletions(-) diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index 2ca70f882..ad2c501a9 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -672,17 +672,19 @@ func (db *commonMsgDatabase) DeleteMsgsPhysicalBySeqs(ctx context.Context, conve } func (db *commonMsgDatabase) DeleteUserMsgsBySeqs(ctx context.Context, userID string, conversationID string, seqs []int64) error { - msgs, _, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs) - if err != nil { + cachedMsgs, _, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs) + if err != nil && errs.Unwrap(err) != redis.Nil { log.ZWarn(ctx, "DeleteUserMsgsBySeqs", err, "conversationID", conversationID, "seqs", seqs) return err } - var cacheSeqs []int64 - for _, msg := range msgs { - cacheSeqs = append(cacheSeqs, msg.Seq) - } - if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil && errs.Unwrap(err) != redis.Nil { - return err + if len(cachedMsgs) > 0 { + var cacheSeqs []int64 + for _, msg := range cachedMsgs { + cacheSeqs = append(cacheSeqs, msg.Seq) + } + if err := db.cache.UserDeleteMsgs(ctx, conversationID, cacheSeqs, userID); err != nil { + return err + } } for docID, seqs := range db.msg.GetDocIDSeqsMap(conversationID, seqs) { From 03d910028b58a318d163ffd5a4e4b1a659d458e0 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Thu, 8 Jun 2023 20:29:10 +0800 Subject: [PATCH 4/6] del --- pkg/common/db/controller/msg.go | 22 +++++++++++++++++----- 1 file changed, 17 insertions(+), 5 deletions(-) diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index ad2c501a9..b6df02822 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -465,20 +465,23 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin maxSeq = userMaxSeq } } + newBegin := begin + newEnd := end if begin < minSeq { - begin = minSeq + newBegin = minSeq } if end > maxSeq { - end = maxSeq + newEnd = maxSeq } var seqs []int64 - for i := end; i > end-num; i-- { - if i >= begin { + for i := newEnd; i > newEnd-num; i-- { + if i >= newBegin { seqs = append([]int64{i}, seqs...) } else { break } } + log.ZDebug(ctx, "GetMsgBySeqsRange", "first seqs", seqs) cachedMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs) if err != nil { if err != redis.Nil { @@ -500,10 +503,19 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin successMsgs = append(successMsgs, msg) } } + for i := 0; i < len(delSeqs); i++ { + newSeq := newBegin - 1 + if newSeq >= begin { + log.ZDebug(ctx, "seq del in cache, a new seq in range append", "new seq", newSeq) + failedSeqs = append(failedSeqs, newSeq) + } else { + break + } + } } log.ZDebug(ctx, "get msgs from cache", "successMsgs", successMsgs) if len(failedSeqs) != 0 { - log.ZDebug(ctx, "msgs not exist in redis", err, "seqs", seqs) + log.ZDebug(ctx, "msgs not exist in redis", err, "seqs", failedSeqs) } // get from cache or db prome.Add(prome.MsgPullFromRedisSuccessCounter, len(successMsgs)) From 2a78974cc10936ebae0a93874365e10439c65cf0 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Fri, 9 Jun 2023 10:58:12 +0800 Subject: [PATCH 5/6] both stdout and file --- pkg/common/log/zap.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/common/log/zap.go b/pkg/common/log/zap.go index 9f6fe633a..407fc8f12 100644 --- a/pkg/common/log/zap.go +++ b/pkg/common/log/zap.go @@ -119,7 +119,7 @@ func (l *ZapLogger) cores(isStdout bool, isJson bool, logLocation string, rotate return nil, err } var cores []zapcore.Core - if logLocation != "" && !isStdout { + if logLocation != "" { cores = []zapcore.Core{ zapcore.NewCore(fileEncoder, writer, zap.NewAtomicLevelAt(l.level)), } From 202a90ab657e60d7db5a4bed3beae7693af412c0 Mon Sep 17 00:00:00 2001 From: wangchuxiao Date: Fri, 9 Jun 2023 13:04:18 +0800 Subject: [PATCH 6/6] fix del seq --- pkg/common/db/controller/msg.go | 24 ++++++++++++++---------- 1 file changed, 14 insertions(+), 10 deletions(-) diff --git a/pkg/common/db/controller/msg.go b/pkg/common/db/controller/msg.go index b6df02822..5c9f4b570 100644 --- a/pkg/common/db/controller/msg.go +++ b/pkg/common/db/controller/msg.go @@ -465,28 +465,32 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin maxSeq = userMaxSeq } } - newBegin := begin - newEnd := end + if begin < minSeq { - newBegin = minSeq + begin = minSeq } if end > maxSeq { - newEnd = maxSeq + end = maxSeq + } + if end < begin { + return 0, 0, nil, errs.ErrArgs.Wrap("seq end < begin") } var seqs []int64 - for i := newEnd; i > newEnd-num; i-- { - if i >= newBegin { + for i := end; i > end-num; i-- { + if i >= begin { seqs = append([]int64{i}, seqs...) } else { break } } - log.ZDebug(ctx, "GetMsgBySeqsRange", "first seqs", seqs) + newBegin := seqs[0] + newEnd := seqs[len(seqs)-1] + log.ZDebug(ctx, "GetMsgBySeqsRange", "first seqs", seqs, "newBegin", newBegin, "newEnd", newEnd) cachedMsgs, failedSeqs, err := db.cache.GetMessagesBySeq(ctx, conversationID, seqs) if err != nil { if err != redis.Nil { prome.Add(prome.MsgPullFromRedisFailedCounter, len(failedSeqs)) - log.ZError(ctx, "get message from redis exception", err, conversationID, seqs) + log.ZError(ctx, "get message from redis exception", err, "conversationID", conversationID, "seqs", seqs) } } var successMsgs []*sdkws.MsgData @@ -503,8 +507,8 @@ func (db *commonMsgDatabase) GetMsgBySeqsRange(ctx context.Context, userID strin successMsgs = append(successMsgs, msg) } } - for i := 0; i < len(delSeqs); i++ { - newSeq := newBegin - 1 + for i := 1; i <= len(delSeqs); i++ { + newSeq := newBegin - int64(i) if newSeq >= begin { log.ZDebug(ctx, "seq del in cache, a new seq in range append", "new seq", newSeq) failedSeqs = append(failedSeqs, newSeq)