[BUG] MongoDB write failures may be skipped when later Kafka offsets are marked

[www0527-ux]

OpenIM Server Version

main at commit 175a7bb0673eca18e9d1b10bff4f728da6b1b513

Operating System and CPU Architecture

Platform-independent source-level issue (inspected on Windows AMD64)

Deployment Method

Source Code Deployment

Bug Description and Steps to Reproduce

This is related to #3080, which covers the case where the Redis/cache stage succeeds but publishing to the toMongo topic fails. This report concerns a different failure boundary: the message has already reached the toMongo topic, but the MongoDB write fails.

In the current flow:

1. The toMongo consumer is created with auto-commit enabled:
   

   open-im-server/pkg/mqbuild/builder.go

   Line 172 in 175a7bb

   return kafka.NewMConsumerGroupV2(ctx, x.config, groupID, []string{realTopic}, true)

2. The subscription callback invokes HandleChatWs2Mongo, but always returns nil:
   

   open-im-server/internal/msgtransfer/init.go

   Lines 140 to 145 in 175a7bb

   	go func() {
   	fn := func(msg mq.Message) error {
   	m.historyMongoHandler.HandleChatWs2Mongo(msg)
   	return nil
   	}

3. If BatchInsertChat2DB fails, the handler only logs the error and increments a metric. It does not mark the current message, but it also does not return an error or stop consumption:
   

   open-im-server/internal/msgtransfer/online_msg_to_mongo_handler.go

   Lines 56 to 64 in 175a7bb

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

4. Consumption therefore continues. If a later record in the same partition is persisted successfully, val.Mark() marks the later offset. With Kafka/Sarama offset semantics, the next auto-commit may advance past the earlier failed record, so that record will not necessarily be redelivered after a restart or rebalance.

A fault-injection reproduction would be:

1. Send messages A and B with the same Kafka key so they are written to the same partition in order.
2. Make BatchInsertChat2DB fail for A.
3. Restore MongoDB before B is handled, allowing B to succeed and call val.Mark().
4. Wait for the consumer group offset to be committed, then restart the consumer.
5. Observe whether A is redelivered. The committed offset may already be past A.

There is also no visible application-level retry, retry limit, DLQ, or persistence-failure event for this path. Because the sender has already received the earlier acceptance response and Redis message entries expire, a permanently missing MongoDB record can eventually become an unrecoverable history gap.

Additionally, webhookAfterMsgSaveDB is currently invoked after the error branch as well, so it may run even when the database write failed:

open-im-server/internal/msgtransfer/online_msg_to_mongo_handler.go

Lines 66 to 68 in 175a7bb

	for _, msgData := range msgFromMQ.MsgData {
	mc.webhookAfterMsgSaveDB(ctx, &mc.config.WebhooksConfig.AfterMsgSaveDB, msgData)
	}

Expected Behavior

A MongoDB persistence failure should not allow the consumer group offset to advance past the failed record. The message should be retried without blocking the partition forever, or moved to a DLQ after a bounded number of attempts. Database writes should remain idempotent so retries are safe.

Possible approaches include:

• Propagate the persistence error from HandleChatWs2Mongo to the consumer loop.
• Retry per partition and only mark contiguous successfully processed offsets.
• After a bounded retry count, publish the record and error context to a DLQ and alert operators.
• Invoke webhookAfterMsgSaveDB only after a successful database write.

If there is an external compensation mechanism that is not visible in this repository, could you please point to it or document how it prevents this offset from being skipped?

Screenshots Link

Not applicable; this report is based on the source control flow above.