feat: batch kafka message persistence

server/db/adapter.go

@@ -260,6 +260,11 @@ type TenantRoutedAdapter interface {
 	TenantAdapterFor(t.TenantID) (TenantBusinessAdapter, TenantAuthAdapter, error)
 }
 
+type TenantMessageBatchAdapter interface {
+	TenantMessageSaveBatch(t.TenantID, []*t.Message) error
+	TenantGroupMessageSaveBatch(t.TenantID, []*t.Message) error
+}
+
 // TenantHTTPInfrastructureAdapter is the optional durable infrastructure used
 // by Kafka dead-letter handling.
 type TenantHTTPInfrastructureAdapter interface {

server/db/mysql/tenant_adapter.go

@@ -2249,6 +2249,41 @@ func (a *adapter) TenantMessageSave(id t.TenantID, msg *t.Message) error {
 	return err
 }
 
+func (a *adapter) TenantMessageSaveBatch(id t.TenantID, messages []*t.Message) error {
+	tenantAdp, err := a.routedTenantAdapter(id)
+	if err != nil {
+		return err
+	}
+	if len(messages) == 0 {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantMessageSaveBatch(id, messages)
+	}
+	ctx, cancel := a.getContext()
+	if cancel != nil {
+		defer cancel()
+	}
+	values := make([]string, 0, len(messages))
+	args := make([]any, 0, len(messages)*13)
+	for _, msg := range messages {
+		if msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() {
+			return t.ErrMalformed
+		}
+		values = append(values, "(?,?,?,?,?,?,?,?,?,?,?,?,?)")
+		args = append(args, id, store.DecodeUid(msg.Uid()), msg.CreatedAt, msg.UpdatedAt, msg.SeqId, msg.Topic,
+			decodeUID(t.ParseUid(msg.From)), sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""},
+			messageVersion(msg.Version), msg.Head, common.ToJSON(msg.Content), msg.ContentNonce, msg.KeyVersion)
+	}
+	query := fmt.Sprintf("INSERT INTO %s(tenant_id,id,createdat,updatedat,seqid,topic,`from`,client_msg_id,version,head,content,content_nonce,key_version) VALUES %s",
+		tenantMessagesTable(id), strings.Join(values, ","))
+	_, err = a.db.ExecContext(ctx, query, args...)
+	if isDupe(err) {
+		return t.ErrDuplicate
+	}
+	return err
+}
+
 func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error {
 	tenantAdp, err := a.routedTenantAdapter(id)
 	if err != nil || msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 {
@@ -2315,6 +2350,69 @@ func (a *adapter) TenantGroupMessageSave(id t.TenantID, msg *t.Message) error {
 	return tx.Commit()
 }
 
+func (a *adapter) TenantGroupMessageSaveBatch(id t.TenantID, messages []*t.Message) error {
+	tenantAdp, err := a.routedTenantAdapter(id)
+	if err != nil {
+		return err
+	}
+	if len(messages) == 0 {
+		return t.ErrMalformed
+	}
+	if tenantAdp != a {
+		return tenantAdp.TenantGroupMessageSaveBatch(id, messages)
+	}
+	if err = a.ensureCurrentGroupMessageShardWindow(); err != nil {
+		return err
+	}
+	ctx, cancel := a.getContext()
+	if cancel != nil {
+		defer cancel()
+	}
+	month := messages[0].ShardMonth
+	table, ok := groupMessageShardTable(id, month)
+	if !ok {
+		return t.ErrMalformed
+	}
+	groupIDs := make(map[string]int64)
+	for _, msg := range messages {
+		if msg == nil || msg.TenantID != id || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() || msg.ShardMonth != month {
+			return t.ErrMalformed
+		}
+		if _, found := groupIDs[msg.Topic]; !found {
+			groupIDs[msg.Topic], err = a.groupIDByTopic(ctx, id, msg.Topic)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	var shardStatus string
+	if err = a.db.QueryRowContext(ctx, "SELECT status FROM im_group_message_shards WHERE tenant_id=? AND shard_month=? AND table_name=?", id, month, table).Scan(&shardStatus); err != nil {
+		return err
+	}
+	if shardStatus != "writable" && shardStatus != "sealing" {
+		return fmt.Errorf("mysql: group message shard %s is %s", month, shardStatus)
+	}
+	values := make([]string, 0, len(messages))
+	args := make([]any, 0, len(messages)*20)
+	for _, msg := range messages {
+		contentJSON := common.ToJSON(msg.Content)
+		values = append(values, "(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)")
+		args = append(args, id, store.DecodeUid(msg.Uid()), msg.Topic, groupIDs[msg.Topic], msg.SeqId,
+			decodeUID(t.ParseUid(msg.From)), sql.NullString{String: msg.ClientMsgID, Valid: msg.ClientMsgID != ""},
+			messageHeadType(msg.Head), "normal", contentJSON, msg.Head, messageHeadJSON(msg.Head, "mentioned_user_ids"),
+			messageHeadBool(msg.Head, "mention_all"), 0, len(contentJSON), messageVersion(msg.Version), msg.CreatedAt,
+			msg.UpdatedAt, msg.ContentNonce, msg.KeyVersion)
+	}
+	query := fmt.Sprintf(`INSERT INTO %s
+		(tenant_id,message_id,topic,group_id,seqid,from_user,client_msg_id,msg_type,status,content,head,mentioned_user_ids,mention_all,
+		 file_count,content_size,version,created_at,updated_at,content_nonce,key_version) VALUES %s`, table, strings.Join(values, ","))
+	_, err = a.db.ExecContext(ctx, query, args...)
+	if isDupe(err) {
+		return t.ErrDuplicate
+	}
+	return err
+}
+
 func (a *adapter) TenantMessageGetByClientID(id t.TenantID, from t.Uid, clientMsgID string) (*t.Message, error) {
 	if !a.tenantScopedDB {
 		tenantAdp, err := a.routedTenantAdapter(id)

server/kafka_admin_commands.go

@@ -141,7 +141,6 @@ func handleKafkaAdminCommand(ctx context.Context, envelope kafkaEnvelope, _ kafk
 			return nil
 		}
 		if kafkaCommandReasonRetryable(reason) {
-			sleepKafkaRetryBackoff(failedRecord.RetryCount)
 			return execErr
 		}
 		if err := publishKafkaAdminCommandResult(ctx, envelope, payload.CommandID, "failed", reason, execErr.Error(), result); err != nil {
@@ -608,7 +607,7 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope
 	if err != nil {
 		return err
 	}
-	return publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{
+	if err = publishKafkaDeadLetter(ctx, "dead_letter", envelope.EventID, kafkaEnvelope{
 		EventID:       deadID,
 		TraceID:       envelope.TraceID,
 		TenantCode:    envelope.TenantCode,
@@ -621,7 +620,12 @@ func markKafkaDeadLetter(ctx context.Context, tenantID types.TenantID, envelope
 		OccurredAt:    now,
 		ParentEventID: envelope.EventID,
 		Payload:       deadPayload,
-	})
+	}); err != nil {
+		return err
+	}
+	// Persist the terminal state before the source consumer commits its offset.
+	// The dead-letter consumer repeats this idempotent upsert for recovery and auditing.
+	return store.HTTPInfrastructure.MarkKafkaConsumeDead(tenantID, dead, now)
 }
 
 func shouldDeadLetterKafkaCommand(reason string, retryCount int) bool {
@@ -643,21 +647,6 @@ func kafkaCommandReasonRetryable(reason string) bool {
 	}
 }
 
-func sleepKafkaRetryBackoff(retryCount int) {
-	if kafkaBus == nil || retryCount <= 0 {
-		return
-	}
-	backoffs := kafkaBus.config.RetryBackoffMS
-	index := retryCount - 1
-	if index >= len(backoffs) {
-		index = len(backoffs) - 1
-	}
-	delay := time.Duration(backoffs[index]) * time.Millisecond
-	if delay > 0 {
-		time.Sleep(delay)
-	}
-}
-
 func parseKafkaCommandUser(raw string) types.Uid {
 	raw = strings.TrimSpace(raw)
 	if raw == "" {

server/kafka_cache_invalidate.go

@@ -70,7 +70,6 @@ func handleKafkaCacheInvalidateEvent(ctx context.Context, envelope kafkaEnvelope
 		if failed.RetryCount >= kafkaBus.config.MaxRetryCount {
 			return markKafkaDeadLetter(ctx, tenantID, envelope, failed, reason, err.Error())
 		}
-		sleepKafkaRetryBackoff(failed.RetryCount)
 		return err
 	}
 	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond))

server/kafka_config_events.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"strings"
 	"time"
 
@@ -12,6 +13,8 @@ import (
 	"github.com/tinode/chat/server/store/types"
 )
 
+var errKafkaConsumeInProgress = errors.New("kafka event is already being processed")
+
 const (
 	configEventTenantUpdated         = "tenant.updated"
 	configEventTenantStatusChanged   = "tenant.status.changed"
@@ -66,7 +69,6 @@ func handleKafkaConfigEvent(ctx context.Context, envelope kafkaEnvelope, _ kafka
 		if failed.RetryCount >= kafkaBus.config.MaxRetryCount {
 			return markKafkaDeadLetter(ctx, tenantID, envelope, failed, reason, err.Error())
 		}
-		sleepKafkaRetryBackoff(failed.RetryCount)
 		return err
 	}
 	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, consumerGroup, time.Now().UTC().Round(time.Millisecond))
@@ -155,7 +157,7 @@ func beginKafkaConsumeForTopicWithGroup(tenantID types.TenantID, envelope kafkaE
 		consumerGroup = kafkaBus.config.ConsumerGroup
 	}
 	deadline := time.Now().UTC().Add(5 * time.Minute).Round(time.Millisecond)
-	return store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{
+	record, claimed, err := store.HTTPInfrastructure.BeginKafkaConsume(tenantID, &types.KafkaConsumeRecord{
 		EventID:       envelope.EventID,
 		ConsumerGroup: consumerGroup,
 		SourceTopic:   kafkaTopic(topicName),
@@ -163,4 +165,11 @@ func beginKafkaConsumeForTopicWithGroup(tenantID types.TenantID, envelope kafkaE
 		BizType:       envelope.BizType,
 		BizID:         envelope.BizID,
 	}, deadline)
+	if err != nil || claimed || record == nil {
+		return record, claimed, err
+	}
+	if record.Status != types.KafkaConsumeSucceeded && record.Status != types.KafkaConsumeDead {
+		return record, false, errKafkaConsumeInProgress
+	}
+	return record, false, nil
 }

server/kafka_message_persist.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"strings"
 	"time"
 
 	"github.com/segmentio/kafka-go"
@@ -24,8 +25,12 @@ type kafkaMessagePersistPayload struct {
 var publishKafkaMessagePersist = publishKafkaEnvelope
 
 func startKafkaMessagePersistConsumer() {
-	startKafkaConsumer("message_persist", handleKafkaMessagePersist)
-	startKafkaConsumer("group_message_persist", handleKafkaGroupMessagePersist)
+	startKafkaBatchConsumer("message_persist", func(ctx context.Context, messages []kafka.Message) error {
+		return handleKafkaMessagePersistBatch(ctx, messages, "message_persist")
+	})
+	startKafkaBatchConsumer("group_message_persist", func(ctx context.Context, messages []kafka.Message) error {
+		return handleKafkaMessagePersistBatch(ctx, messages, "group_message_persist")
+	})
 }
 
 func saveMessageMaybeKafka(ctx context.Context, tenantID types.TenantID, msg *types.Message, attachmentURLs []string, readBySender bool) (*types.MessageSaveResult, error) {
@@ -132,70 +137,155 @@ func enqueueKafkaMessagePersist(ctx context.Context, tenantID types.TenantID, ms
 	return publishKafkaMessagePersist(ctx, kafkaMessagePersistTopicName(msg.Topic), msg.Topic, envelope)
 }
 
-func handleKafkaMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
-	return handleKafkaMessagePersistFromTopic(ctx, envelope, "message_persist")
+type kafkaMessagePersistItem struct {
+	envelope kafkaEnvelope
+	tenantID types.TenantID
+	payload  kafkaMessagePersistPayload
 }
 
-func handleKafkaGroupMessagePersist(ctx context.Context, envelope kafkaEnvelope, _ kafka.Message) error {
-	return handleKafkaMessagePersistFromTopic(ctx, envelope, "group_message_persist")
+func handleKafkaMessagePersistBatch(ctx context.Context, messages []kafka.Message, sourceTopicName string) error {
+	groups := make(map[string][]*kafkaMessagePersistItem)
+	for _, message := range messages {
+		item, terminal, err := prepareKafkaMessagePersistItem(ctx, message, sourceTopicName)
+		if err != nil {
+			return err
+		}
+		if terminal {
+			continue
+		}
+		key := fmt.Sprintf("%d", item.tenantID)
+		if item.payload.MessageKind == "group" {
+			key += ":" + item.payload.Message.ShardMonth
+		}
+		groups[key] = append(groups[key], item)
+	}
+	for _, items := range groups {
+		batch := make([]store.PreparedMessageBatchItem, 0, len(items))
+		for _, item := range items {
+			batch = append(batch, store.PreparedMessageBatchItem{Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender})
+		}
+		if err := store.PersistPreparedMessageBatch(items[0].tenantID, batch); err == nil {
+			for _, item := range items {
+				if err = finishKafkaMessagePersist(ctx, item); err != nil {
+					if err = persistKafkaMessageIndividually(ctx, item); err != nil {
+						return err
+					}
+				}
+			}
+			continue
+		} else {
+			logs.Warn.Printf("kafka message batch persist failed, degrading to singles: tenant=%d kind=%s shard=%s size=%d err=%v",
+				items[0].tenantID, items[0].payload.MessageKind, items[0].payload.Message.ShardMonth, len(items), err)
+		}
+		for _, item := range items {
+			if err := persistKafkaMessageIndividually(ctx, item); err != nil {
+				return err
+			}
+		}
+	}
+	return nil
 }
 
-func handleKafkaMessagePersistFromTopic(ctx context.Context, envelope kafkaEnvelope, sourceTopicName string) error {
+func prepareKafkaMessagePersistItem(ctx context.Context, message kafka.Message, sourceTopicName string) (*kafkaMessagePersistItem, bool, error) {
+	var envelope kafkaEnvelope
+	if err := json.Unmarshal(message.Value, &envelope); err != nil {
+		logs.Err.Printf("kafka message malformed envelope skipped: partition=%d offset=%d err=%v", message.Partition, message.Offset, err)
+		return nil, true, nil
+	}
 	tenantID := envelope.TenantID
 	var err error
 	if tenantID.IsZero() {
 		tenantID, err = kafkaTenantID(envelope.TenantCode)
 		if err != nil {
-			return err
+			return nil, false, err
 		}
 	}
 	consume, claimed, err := beginKafkaConsumeForTopic(tenantID, envelope, sourceTopicName)
-	if err != nil || !claimed {
-		return err
+	if err != nil {
+		return nil, false, err
 	}
-	var payload kafkaMessagePersistPayload
-	if err = json.Unmarshal(envelope.Payload, &payload); err != nil {
-		return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
+	if !claimed {
+		return nil, true, nil
 	}
-	if payload.Message.TenantID == types.ZeroTenantID {
-		payload.Message.TenantID = tenantID
+	item := &kafkaMessagePersistItem{envelope: envelope, tenantID: tenantID}
+	if err = json.Unmarshal(envelope.Payload, &item.payload); err != nil {
+		return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
 	}
-	if payload.Message.ShardMonth == "" {
-		payload.Message.ShardMonth = payload.ShardMonth
+	if item.payload.Message.TenantID == types.ZeroTenantID {
+		item.payload.Message.TenantID = tenantID
 	}
-	if payload.Message.TenantID != tenantID {
-		return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "TENANT_MISMATCH", "message tenant does not match envelope tenant")
+	if item.payload.Message.ShardMonth == "" {
+		item.payload.Message.ShardMonth = item.payload.ShardMonth
 	}
-	if err = validateKafkaMessageKind(payload); err != nil {
-		return markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
+	if item.payload.Message.TenantID != tenantID {
+		return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "TENANT_MISMATCH", "message tenant does not match envelope tenant")
 	}
-	if _, err = store.PersistPreparedMessage(tenantID, &payload.Message, nil, payload.ReadBySender); err != nil && !errors.Is(err, types.ErrDuplicate) {
-		failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup,
-			kafkaCommandErrorReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond))
-		if markErr != nil {
-			return markErr
+	if err = validateKafkaMessageKind(item.payload); err != nil {
+		return nil, true, markKafkaDeadLetter(ctx, tenantID, envelope, consume, "MALFORMED_PAYLOAD", err.Error())
+	}
+	return item, false, nil
+}
+
+func persistKafkaMessageIndividually(ctx context.Context, item *kafkaMessagePersistItem) error {
+	operation := func() error {
+		err := store.PersistPreparedMessageBatch(item.tenantID, []store.PreparedMessageBatchItem{{
+			Message: &item.payload.Message, ReadBySender: item.payload.ReadBySender,
+		}})
+		if errors.Is(err, types.ErrUnsupported) {
+			_, err = store.PersistPreparedMessage(item.tenantID, &item.payload.Message, nil, item.payload.ReadBySender)
 		}
-		if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
-			return markKafkaDeadLetter(ctx, tenantID, envelope, failed, kafkaCommandErrorReason(err), err.Error())
-		}
-		sleepKafkaRetryBackoff(failed.RetryCount)
-		return err
-	}
-	if len(payload.AttachmentURLs) > 0 {
-		if err = enqueueKafkaFileAttachmentsBind(ctx, tenantID, &payload.Message, payload.AttachmentURLs); err != nil {
-			failed, markErr := store.HTTPInfrastructure.MarkKafkaConsumeFailed(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup,
-				"KAFKA_FILE_LIFECYCLE_FAILED", err.Error(), time.Now().UTC().Round(time.Millisecond))
-			if markErr != nil {
-				return markErr
-			}
-			if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
-				return markKafkaDeadLetter(ctx, tenantID, envelope, failed, "KAFKA_FILE_LIFECYCLE_FAILED", err.Error())
-			}
-			sleepKafkaRetryBackoff(failed.RetryCount)
+		if err != nil && !errors.Is(err, types.ErrDuplicate) {
 			return err
 		}
+		return finishKafkaMessagePersist(ctx, item)
 	}
-	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(tenantID, envelope.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
+	markFailed := func(err error) (*types.KafkaConsumeRecord, error) {
+		return store.HTTPInfrastructure.MarkKafkaConsumeFailed(item.tenantID, item.envelope.EventID, kafkaBus.config.ConsumerGroup,
+			kafkaMessagePersistFailureReason(err), err.Error(), time.Now().UTC().Round(time.Millisecond))
+	}
+	markDead := func(failed *types.KafkaConsumeRecord, err error) error {
+		return markKafkaDeadLetter(ctx, item.tenantID, item.envelope, failed, kafkaMessagePersistFailureReason(err), err.Error())
+	}
+	return runKafkaMessageSingleRetries(ctx, kafkaBus.config.MessageRetries, operation, markFailed, markDead, waitKafkaRetry)
+}
+
+func runKafkaMessageSingleRetries(ctx context.Context, retryCount int, operation func() error,
+	markFailed func(error) (*types.KafkaConsumeRecord, error), markDead func(*types.KafkaConsumeRecord, error) error,
+	wait kafkaRetryWaiter) error {
+	var lastErr error
+	var failed *types.KafkaConsumeRecord
+	for attempt := 1; attempt <= retryCount+1; attempt++ {
+		if lastErr = operation(); lastErr == nil {
+			return nil
+		}
+		var err error
+		if failed, err = markFailed(lastErr); err != nil {
+			return err
+		}
+		if attempt <= retryCount {
+			if err = wait(ctx, kafkaRetryBackoff(attempt)); err != nil {
+				return err
+			}
+		}
+	}
+	return markDead(failed, lastErr)
+}
+
+func finishKafkaMessagePersist(ctx context.Context, item *kafkaMessagePersistItem) error {
+	if len(item.payload.AttachmentURLs) > 0 {
+		if err := enqueueKafkaFileAttachmentsBind(ctx, item.tenantID, &item.payload.Message, item.payload.AttachmentURLs); err != nil {
+			return fmt.Errorf("file lifecycle enqueue failed: %w", err)
+		}
+	}
+	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(item.tenantID, item.envelope.EventID,
+		kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))
+}
+
+func kafkaMessagePersistFailureReason(err error) string {
+	if err != nil && strings.Contains(err.Error(), "file lifecycle") {
+		return "KAFKA_FILE_LIFECYCLE_FAILED"
+	}
+	return kafkaCommandErrorReason(err)
 }
 
 func kafkaMessagePersistTopicName(topic string) string {

server/kafka_runtime.go

@@ -51,6 +51,9 @@ type kafkaConfig struct {
 	SASLPassword     string            `json:"sasl_password"`
 	MaxRetryCount    int               `json:"max_retry_count"`
 	RetryBackoffMS   []int             `json:"retry_backoff_ms"`
+	MessageBatchSize int               `json:"message_batch_size"`
+	MessageBatchWait int               `json:"message_batch_wait_ms"`
+	MessageRetries   int               `json:"message_single_retry_count"`
 	DeadRetentionSec int               `json:"dead_letter_retention_seconds"`
 }
 
@@ -180,6 +183,17 @@ func normalizeKafkaConfig(cfg *kafkaConfig) {
 	if len(cfg.RetryBackoffMS) == 0 {
 		cfg.RetryBackoffMS = []int{10000, 30000, 60000, 300000, 900000}
 	}
+	if cfg.MessageBatchSize <= 0 {
+		cfg.MessageBatchSize = 100
+	} else if cfg.MessageBatchSize > 500 {
+		cfg.MessageBatchSize = 500
+	}
+	if cfg.MessageBatchWait <= 0 {
+		cfg.MessageBatchWait = 10
+	}
+	if cfg.MessageRetries <= 0 {
+		cfg.MessageRetries = 3
+	}
 	if cfg.DeadRetentionSec <= 0 {
 		cfg.DeadRetentionSec = 90 * 24 * 3600
 	}
@@ -276,6 +290,11 @@ func publishKafkaEnvelope(ctx context.Context, topicName string, key string, env
 
 type kafkaMessageHandler func(context.Context, kafkaEnvelope, kafka.Message) error
 
+type kafkaMessageCommitter func(context.Context, kafka.Message) error
+type kafkaRetryWaiter func(context.Context, time.Duration) error
+type kafkaBatchHandler func(context.Context, []kafka.Message) error
+type kafkaBatchCommitter func(context.Context, ...kafka.Message) error
+
 func startKafkaConsumer(topicName string, handler kafkaMessageHandler) {
 	startKafkaConsumerWithGroup(topicName, "", handler)
 }
@@ -327,24 +346,157 @@ func startKafkaConsumerWithGroup(topicName, groupID string, handler kafkaMessage
 				logs.Err.Println("Kafka fetch failed:", err)
 				continue
 			}
-			var envelope kafkaEnvelope
-			if err = json.Unmarshal(msg.Value, &envelope); err != nil {
-				logs.Err.Println("Kafka envelope decode failed:", err)
-				_ = reader.CommitMessages(context.Background(), msg)
-				continue
+			commit := func(ctx context.Context, message kafka.Message) error {
+				return reader.CommitMessages(ctx, message)
 			}
-			if err = handler(kafkaBus.ctx, envelope, msg); err != nil {
-				logs.Err.Println("Kafka handler failed:", envelope.EventID, envelope.EventType, err)
-				continue
-			}
-			if err = reader.CommitMessages(context.Background(), msg); err != nil {
-				logs.Err.Println("Kafka commit failed:", err)
+			if err = consumeKafkaMessage(kafkaBus.ctx, msg, handler, commit, waitKafkaRetry); err != nil {
+				if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
+					return
+				}
+				logs.Err.Printf("Kafka message stopped without commit: topic=%s partition=%d offset=%d err=%v", msg.Topic, msg.Partition, msg.Offset, err)
 			}
 		}
 	}()
 	logs.Info.Printf("Kafka consumer started: topic=%s group=%s", topic, groupID)
 }
 
+func startKafkaBatchConsumer(topicName string, handler kafkaBatchHandler) {
+	if kafkaBus == nil || handler == nil {
+		return
+	}
+	topic := kafkaBus.topics[topicName]
+	if topic == "" {
+		topic = topicName
+	}
+	groupID := kafkaBus.config.ConsumerGroup + "." + topicName
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers: kafkaBus.config.Brokers, GroupID: groupID, Topic: topic, StartOffset: kafka.FirstOffset,
+		MinBytes: 1, MaxBytes: 10e6, MaxWait: time.Second, Dialer: kafkaBus.dialer(),
+	})
+	kafkaBus.wg.Add(1)
+	go func() {
+		defer kafkaBus.wg.Done()
+		defer reader.Close()
+		for {
+			messages, err := fetchKafkaMessageBatch(kafkaBus.ctx, reader, kafkaBus.config.MessageBatchSize,
+				time.Duration(kafkaBus.config.MessageBatchWait)*time.Millisecond)
+			if err != nil {
+				if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
+					return
+				}
+				logs.Err.Println("Kafka batch fetch failed:", err)
+				continue
+			}
+			if err = consumeKafkaMessageBatch(kafkaBus.ctx, messages, handler, reader.CommitMessages, waitKafkaRetry); err != nil {
+				if errors.Is(err, context.Canceled) || errors.Is(kafkaBus.ctx.Err(), context.Canceled) {
+					return
+				}
+				logs.Err.Printf("Kafka batch stopped without commit: topic=%s size=%d err=%v", topic, len(messages), err)
+			}
+		}
+	}()
+	logs.Info.Printf("Kafka batch consumer started: topic=%s group=%s size=%d wait_ms=%d", topic, groupID, kafkaBus.config.MessageBatchSize, kafkaBus.config.MessageBatchWait)
+}
+
+func fetchKafkaMessageBatch(ctx context.Context, reader *kafka.Reader, maxSize int, maxWait time.Duration) ([]kafka.Message, error) {
+	first, err := reader.FetchMessage(ctx)
+	if err != nil {
+		return nil, err
+	}
+	messages := []kafka.Message{first}
+	deadline := time.Now().Add(maxWait)
+	for len(messages) < maxSize {
+		remaining := time.Until(deadline)
+		if remaining <= 0 {
+			break
+		}
+		fetchCtx, cancel := context.WithTimeout(ctx, remaining)
+		message, fetchErr := reader.FetchMessage(fetchCtx)
+		cancel()
+		if errors.Is(fetchErr, context.DeadlineExceeded) && ctx.Err() == nil {
+			break
+		}
+		if fetchErr != nil {
+			return nil, fetchErr
+		}
+		messages = append(messages, message)
+	}
+	return messages, nil
+}
+
+func consumeKafkaMessageBatch(ctx context.Context, messages []kafka.Message, handler kafkaBatchHandler, commit kafkaBatchCommitter, wait kafkaRetryWaiter) error {
+	for attempt := 1; ; attempt++ {
+		if err := handler(ctx, messages); err == nil {
+			return commitKafkaMessageBatch(ctx, messages, commit, wait)
+		}
+		if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil {
+			return err
+		}
+	}
+}
+
+func consumeKafkaMessage(ctx context.Context, msg kafka.Message, handler kafkaMessageHandler, commit kafkaMessageCommitter, wait kafkaRetryWaiter) error {
+	var envelope kafkaEnvelope
+	if err := json.Unmarshal(msg.Value, &envelope); err != nil {
+		return commitKafkaMessage(ctx, msg, commit, wait)
+	}
+	for attempt := 1; ; attempt++ {
+		if err := handler(ctx, envelope, msg); err == nil {
+			return commitKafkaMessage(ctx, msg, commit, wait)
+		}
+		if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil {
+			return err
+		}
+	}
+}
+
+func commitKafkaMessage(ctx context.Context, msg kafka.Message, commit kafkaMessageCommitter, wait kafkaRetryWaiter) error {
+	for attempt := 1; ; attempt++ {
+		if err := commit(ctx, msg); err == nil {
+			return nil
+		}
+		if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil {
+			return err
+		}
+	}
+}
+
+func commitKafkaMessageBatch(ctx context.Context, messages []kafka.Message, commit kafkaBatchCommitter, wait kafkaRetryWaiter) error {
+	for attempt := 1; ; attempt++ {
+		if err := commit(ctx, messages...); err == nil {
+			return nil
+		}
+		if err := wait(ctx, kafkaRetryBackoff(attempt)); err != nil {
+			return err
+		}
+	}
+}
+
+func kafkaRetryBackoff(attempt int) time.Duration {
+	if kafkaBus == nil || len(kafkaBus.config.RetryBackoffMS) == 0 {
+		return time.Second
+	}
+	if attempt < 1 {
+		attempt = 1
+	}
+	index := attempt - 1
+	if index >= len(kafkaBus.config.RetryBackoffMS) {
+		index = len(kafkaBus.config.RetryBackoffMS) - 1
+	}
+	return time.Duration(kafkaBus.config.RetryBackoffMS[index]) * time.Millisecond
+}
+
+func waitKafkaRetry(ctx context.Context, delay time.Duration) error {
+	timer := time.NewTimer(delay)
+	defer timer.Stop()
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	case <-timer.C:
+		return nil
+	}
+}
+
 func kafkaBroadcastConsumerGroup(topicName string) string {
 	if kafkaBus == nil {
 		return ""

server/kafka_runtime_batch_test.go

@@ -0,0 +1,143 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+
+	"github.com/segmentio/kafka-go"
+	"github.com/tinode/chat/server/store/types"
+)
+
+func TestConsumeKafkaMessageBatchRetriesHandlerAndCommit(t *testing.T) {
+	handlerCalls := 0
+	commitCalls := 0
+	waitCalls := 0
+	messages := []kafka.Message{{Offset: 1}, {Offset: 2}}
+
+	err := consumeKafkaMessageBatch(context.Background(), messages,
+		func(context.Context, []kafka.Message) error {
+			handlerCalls++
+			if handlerCalls == 1 {
+				return errors.New("batch failed")
+			}
+			return nil
+		},
+		func(_ context.Context, committed ...kafka.Message) error {
+			commitCalls++
+			if len(committed) != len(messages) {
+				t.Fatalf("committed %d messages, want %d", len(committed), len(messages))
+			}
+			if commitCalls == 1 {
+				return errors.New("commit failed")
+			}
+			return nil
+		},
+		func(context.Context, time.Duration) error {
+			waitCalls++
+			return nil
+		})
+
+	if err != nil {
+		t.Fatalf("consumeKafkaMessageBatch() error = %v", err)
+	}
+	if handlerCalls != 2 || commitCalls != 2 || waitCalls != 2 {
+		t.Fatalf("calls handler=%d commit=%d wait=%d, want 2/2/2", handlerCalls, commitCalls, waitCalls)
+	}
+}
+
+func TestConsumeKafkaMalformedMessageRetriesCommit(t *testing.T) {
+	commitCalls := 0
+	waitCalls := 0
+	handlerCalls := 0
+	err := consumeKafkaMessage(context.Background(), kafka.Message{Value: []byte("{")},
+		func(context.Context, kafkaEnvelope, kafka.Message) error {
+			handlerCalls++
+			return nil
+		},
+		func(context.Context, kafka.Message) error {
+			commitCalls++
+			if commitCalls == 1 {
+				return errors.New("commit failed")
+			}
+			return nil
+		},
+		func(context.Context, time.Duration) error {
+			waitCalls++
+			return nil
+		})
+
+	if err != nil {
+		t.Fatalf("consumeKafkaMessage() error = %v", err)
+	}
+	if handlerCalls != 0 || commitCalls != 2 || waitCalls != 1 {
+		t.Fatalf("calls handler=%d commit=%d wait=%d, want 0/2/1", handlerCalls, commitCalls, waitCalls)
+	}
+}
+
+func TestRunKafkaMessageSingleRetriesThenDeadLetters(t *testing.T) {
+	operationCalls := 0
+	markFailedCalls := 0
+	markDeadCalls := 0
+	waitCalls := 0
+
+	err := runKafkaMessageSingleRetries(context.Background(), 3,
+		func() error {
+			operationCalls++
+			return errors.New("write failed")
+		},
+		func(error) (*types.KafkaConsumeRecord, error) {
+			markFailedCalls++
+			return &types.KafkaConsumeRecord{RetryCount: markFailedCalls}, nil
+		},
+		func(record *types.KafkaConsumeRecord, err error) error {
+			markDeadCalls++
+			if record == nil || record.RetryCount != 4 || err == nil {
+				t.Fatalf("unexpected dead-letter input: record=%+v err=%v", record, err)
+			}
+			return nil
+		},
+		func(context.Context, time.Duration) error {
+			waitCalls++
+			return nil
+		})
+
+	if err != nil {
+		t.Fatalf("runKafkaMessageSingleRetries() error = %v", err)
+	}
+	if operationCalls != 4 || markFailedCalls != 4 || waitCalls != 3 || markDeadCalls != 1 {
+		t.Fatalf("calls operation=%d failed=%d wait=%d dead=%d, want 4/4/3/1",
+			operationCalls, markFailedCalls, waitCalls, markDeadCalls)
+	}
+}
+
+func TestRunKafkaMessageSingleRetriesStopsOnSuccess(t *testing.T) {
+	operationCalls := 0
+	markFailedCalls := 0
+	markDeadCalls := 0
+	err := runKafkaMessageSingleRetries(context.Background(), 3,
+		func() error {
+			operationCalls++
+			if operationCalls < 3 {
+				return errors.New("temporary failure")
+			}
+			return nil
+		},
+		func(error) (*types.KafkaConsumeRecord, error) {
+			markFailedCalls++
+			return &types.KafkaConsumeRecord{RetryCount: markFailedCalls}, nil
+		},
+		func(*types.KafkaConsumeRecord, error) error {
+			markDeadCalls++
+			return nil
+		},
+		func(context.Context, time.Duration) error { return nil })
+
+	if err != nil {
+		t.Fatalf("runKafkaMessageSingleRetries() error = %v", err)
+	}
+	if operationCalls != 3 || markFailedCalls != 2 || markDeadCalls != 0 {
+		t.Fatalf("calls operation=%d failed=%d dead=%d, want 3/2/0", operationCalls, markFailedCalls, markDeadCalls)
+	}
+}

server/kafka_workbench_events.go

@@ -56,9 +56,6 @@ func handleKafkaWorkbenchEvent(ctx context.Context, envelope kafkaEnvelope, msg
 		if failed != nil && failed.RetryCount >= kafkaBus.config.MaxRetryCount {
 			return markKafkaDeadLetter(ctx, event.TenantID, envelope, failed, reason, err.Error())
 		}
-		if failed != nil {
-			sleepKafkaRetryBackoff(failed.RetryCount)
-		}
 		return err
 	}
 	return store.HTTPInfrastructure.MarkKafkaConsumeSucceeded(event.TenantID, event.EventID, kafkaBus.config.ConsumerGroup, time.Now().UTC().Round(time.Millisecond))

server/meetingim.conf

@@ -128,6 +128,9 @@
     "sasl_password": "gr58tNnlZOVGW3AmMbL2",
     "max_retry_count": 3,
     "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000],
+    "message_batch_size": 100,
+    "message_batch_wait_ms": 10,
+    "message_single_retry_count": 3,
     "dead_letter_retention_seconds": 7776000,
     "topics": {
       "admin_command": "im.admin.command",

server/store/store.go

@@ -3677,6 +3677,82 @@ func PersistPreparedMessage(tenantID types.TenantID, msg *types.Message, attachm
 	return Messages.PersistPrepared(tenantID, msg, attachmentURLs, readBySender)
 }
 
+type PreparedMessageBatchItem struct {
+	Message      *types.Message
+	ReadBySender bool
+}
+
+func PersistPreparedMessageBatch(tenantID types.TenantID, items []PreparedMessageBatchItem) error {
+	if len(items) == 0 {
+		return nil
+	}
+	tenantAdp, err := tenantBusinessAdapter(tenantID)
+	if err != nil {
+		return err
+	}
+	batchAdp, ok := tenantAdp.(adapter.TenantMessageBatchAdapter)
+	if !ok {
+		return types.ErrUnsupported
+	}
+	if items[0].Message == nil {
+		return types.ErrMalformed
+	}
+	isGroup := types.GetTopicCat(items[0].Message.Topic) == types.TopicCatGrp
+	month := items[0].Message.ShardMonth
+	stored := make([]*types.Message, 0, len(items))
+	for _, item := range items {
+		msg := item.Message
+		if msg == nil || msg.TenantID != tenantID || msg.Topic == "" || msg.SeqId <= 0 || msg.Uid().IsZero() ||
+			(types.GetTopicCat(msg.Topic) == types.TopicCatGrp) != isGroup || (isGroup && msg.ShardMonth != month) {
+			return types.ErrMalformed
+		}
+		clientMsgID, normalizeErr := normalizeClientMsgID(msg.Head)
+		if normalizeErr != nil {
+			return normalizeErr
+		}
+		msg.ClientMsgID = clientMsgID
+		storedMsg, encryptErr := encryptMessageContent(msg)
+		if encryptErr != nil {
+			return encryptErr
+		}
+		stored = append(stored, storedMsg)
+	}
+	if isGroup {
+		err = batchAdp.TenantGroupMessageSaveBatch(tenantID, stored)
+	} else {
+		err = batchAdp.TenantMessageSaveBatch(tenantID, stored)
+	}
+	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 {
+			return err
+		}
+		invalidateTopicCache(tenantID, msg.Topic)
+		if isGroup && groupMsgRuntime != nil {
+			if msg.ClientMsgID != "" {
+				_ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, from, msg.ClientMsgID, msg.SeqId, msg.Uid().String())
+			}
+			_ = groupMsgRuntime.PutShardRoute(tenantID, msg.Topic, GroupMessageShardRoute{
+				Month: msg.ShardMonth, MinSeq: msg.SeqId, MaxSeq: msg.SeqId, Count: 1,
+			})
+			_ = 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)
+			}
+		}
+	}
+	return nil
+}
+
 func NormalizeClientMsgID(head types.KVMap) (string, error) {
 	return normalizeClientMsgID(head)
 }

server/tinode.conf

@@ -128,6 +128,9 @@
     "sasl_password": "gr58tNnlZOVGW3AmMbL2",
     "max_retry_count": 3,
     "retry_backoff_ms": [10000, 30000, 60000, 300000, 900000],
+    "message_batch_size": 100,
+    "message_batch_wait_ms": 10,
+    "message_single_retry_count": 3,
     "dead_letter_retention_seconds": 7776000,
     "topics": {
       "admin_command": "im.admin.command",

server/topic_test.go

@@ -1284,7 +1284,7 @@ func TestFilterHotMessagesForUserSkipsDeleted(t *testing.T) {
 	}
 }
 
-func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) {
+func TestMarkKafkaDeadLetterPublishesAndWritesTerminalState(t *testing.T) {
 	oldKafkaBus := kafkaBus
 	oldPublisher := publishKafkaDeadLetter
 	oldInfrastructure := store.HTTPInfrastructure
@@ -1322,8 +1322,8 @@ func TestMarkKafkaDeadLetterPublishesBeforeDBWrite(t *testing.T) {
 	if err := markKafkaDeadLetter(context.Background(), 1, original, consume, "DB_FAILED", "boom"); err != nil {
 		t.Fatalf("markKafkaDeadLetter returned error: %v", err)
 	}
-	if len(infrastructure.deadLetters) != 0 {
-		t.Fatalf("dead letter was written before dead_letter topic consumption: %#v", infrastructure.deadLetters)
+	if len(infrastructure.deadLetters) != 1 || infrastructure.deadLetters[0].DeadID == "" {
+		t.Fatalf("dead letter terminal state was not written: %#v", infrastructure.deadLetters)
 	}
 	if publishedEnvelope.EventID == "" || publishedEnvelope.TenantID != 1 {
 		t.Fatalf("dead letter event was not published correctly: %#v", publishedEnvelope)