From e2754c561dcac8c1232a3531fe0cb0341624417c Mon Sep 17 00:00:00 2001 From: ver0810 Date: Mon, 28 Sep 2026 16:53:13 +0800 Subject: [PATCH] fix(msgtransfer): retry Mongo persistence before acknowledging Kafka --- internal/msgtransfer/init.go | 41 +++++++++++++- internal/msgtransfer/init_test.go | 56 +++++++++++++++++++ .../online_msg_to_mongo_handler.go | 14 +++-- 3 files changed, 103 insertions(+), 8 deletions(-) create mode 100644 internal/msgtransfer/init_test.go diff --git a/internal/msgtransfer/init.go b/internal/msgtransfer/init.go index 2b483fcd7..70202a761 100644 --- a/internal/msgtransfer/init.go +++ b/internal/msgtransfer/init.go @@ -17,6 +17,7 @@ package msgtransfer import ( "context" "fmt" + "time" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/cache/redis" "github.com/openimsdk/open-im-server/v3/pkg/common/storage/database/mgo" @@ -33,6 +34,11 @@ import ( "github.com/openimsdk/tools/log" ) +const ( + mongoPersistenceRetryInitial = time.Second + mongoPersistenceRetryMax = 30 * time.Second +) + type MsgTransfer struct { historyConsumer mq.Consumer historyMongoConsumer mq.Consumer @@ -140,8 +146,8 @@ func (m *MsgTransfer) Start(ctx context.Context) error { go func() { fn := func(msg mq.Message) error { - m.historyMongoHandler.HandleChatWs2Mongo(msg) - return nil + return consumeMongoMessage(m.ctx, msg, m.historyMongoHandler.HandleChatWs2Mongo, + mongoPersistenceRetryInitial, mongoPersistenceRetryMax) } for { if err := m.historyMongoConsumer.Subscribe(m.ctx, fn); err != nil { @@ -161,3 +167,34 @@ func (m *MsgTransfer) Start(ctx context.Context) error { <-m.ctx.Done() return context.Cause(m.ctx) } + +// consumeMongoMessage leaves a Kafka message unacknowledged until its MongoDB +// persistence succeeds. Since the subscription loop is serial, later messages +// cannot call Mark and advance the partition offset past a failed message. +func consumeMongoMessage(ctx context.Context, msg mq.Message, handle func(mq.Message) error, initial, max time.Duration) error { + delay := initial + for { + if err := handle(msg); err == nil { + msg.Mark() + return nil + } else { + log.ZError(ctx, "persist to MongoDB failed; retrying without acknowledging Kafka message", err, "key", msg.Key(), "retryAfter", delay) + } + + timer := time.NewTimer(delay) + select { + case <-ctx.Done(): + if !timer.Stop() { + <-timer.C + } + return context.Cause(ctx) + case <-timer.C: + } + if delay < max { + delay *= 2 + if delay > max { + delay = max + } + } + } +} diff --git a/internal/msgtransfer/init_test.go b/internal/msgtransfer/init_test.go new file mode 100644 index 000000000..29d55488d --- /dev/null +++ b/internal/msgtransfer/init_test.go @@ -0,0 +1,56 @@ +package msgtransfer + +import ( + "context" + "errors" + "testing" + "time" + + "github.com/openimsdk/tools/mq" +) + +type retryTestMessage struct { + marked int +} + +func (m *retryTestMessage) Context() context.Context { return context.Background() } +func (m *retryTestMessage) Key() string { return "retry-test" } +func (m *retryTestMessage) Value() []byte { return nil } +func (m *retryTestMessage) Mark() { m.marked++ } +func (m *retryTestMessage) Commit() {} + +func TestConsumeMongoMessageRetriesBeforeMarking(t *testing.T) { + msg := &retryTestMessage{} + attempts := 0 + err := consumeMongoMessage(context.Background(), msg, func(mq.Message) error { + attempts++ + if attempts < 3 { + return errors.New("MongoDB unavailable") + } + return nil + }, 0, 0) + if err != nil { + t.Fatalf("consumeMongoMessage() error = %v", err) + } + if attempts != 3 { + t.Fatalf("attempts = %d, want 3", attempts) + } + if msg.marked != 1 { + t.Fatalf("Mark calls = %d, want 1 after persistence succeeds", msg.marked) + } +} + +func TestConsumeMongoMessageDoesNotMarkWhenContextEnds(t *testing.T) { + ctx, cancel := context.WithCancelCause(context.Background()) + cancel(errors.New("test shutdown")) + msg := &retryTestMessage{} + err := consumeMongoMessage(ctx, msg, func(mq.Message) error { + return errors.New("MongoDB unavailable") + }, time.Hour, time.Hour) + if err == nil { + t.Fatal("consumeMongoMessage() error = nil, want cancellation cause") + } + if msg.marked != 0 { + t.Fatalf("Mark calls = %d, want 0 when persistence never succeeds", msg.marked) + } +} diff --git a/internal/msgtransfer/online_msg_to_mongo_handler.go b/internal/msgtransfer/online_msg_to_mongo_handler.go index 147bd37b0..174f6c953 100644 --- a/internal/msgtransfer/online_msg_to_mongo_handler.go +++ b/internal/msgtransfer/online_msg_to_mongo_handler.go @@ -15,6 +15,8 @@ package msgtransfer import ( + "fmt" + "github.com/openimsdk/tools/mq" "github.com/openimsdk/open-im-server/v3/pkg/common/prommetrics" @@ -39,7 +41,7 @@ func NewOnlineHistoryMongoConsumerHandler(database controller.MsgTransferDatabas } } -func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) { +func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) error { ctx := val.Context() key := val.Key() msg := val.Value() @@ -47,21 +49,20 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) err := proto.Unmarshal(msg, &msgFromMQ) if err != nil { log.ZError(ctx, "unmarshall failed", err, "key", key, "len", len(msg)) - return + return fmt.Errorf("unmarshal toMongo message: %w", err) } if len(msgFromMQ.MsgData) == 0 { log.ZError(ctx, "msgFromMQ.MsgData is empty", nil, "key", key, "msg", msg) - return + return fmt.Errorf("toMongo message has no message data") } log.ZDebug(ctx, "mongo consumer recv msg", "msgs", msgFromMQ.String()) err = mc.msgTransferDatabase.BatchInsertChat2DB(ctx, msgFromMQ.ConversationID, msgFromMQ.MsgData, msgFromMQ.LastSeq) if err != nil { log.ZError(ctx, "batch data insert to mongo err", err, "msg", msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) prommetrics.MsgInsertMongoFailedCounter.Inc() - } else { - prommetrics.MsgInsertMongoSuccessCounter.Inc() - val.Mark() + return fmt.Errorf("persist to MongoDB: %w", err) } + prommetrics.MsgInsertMongoSuccessCounter.Inc() for _, msgData := range msgFromMQ.MsgData { mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData) @@ -75,4 +76,5 @@ func (mc *OnlineHistoryMongoConsumerHandler) HandleChatWs2Mongo(val mq.Message) // log.ZError(ctx, "remove cache msg from redis err", err, "msg", // msgFromMQ.MsgData, "conversationID", msgFromMQ.ConversationID) //} + return nil }