fix: harden message seq allocation against stale redis replicas

- use sentinel failover client for group message runtime when
  sentinel_master_name/sentinel_addrs are configured, so seq counter
  reads never hit a stale replica after failover
- re-seed missing redis seq counter from DB topic seqid instead of
  trusting the caller in-memory baseline
- on unique-key conflict during async persist, re-allocate seq from DB
  latest, bump redis counter and fix the idempotency record so resends
  with the same client mid return the final seq
- add uk(topic, seqid) unique keys for im_messages in tenant DDL

scripts/IM-租户业务库初始化.sql

@@ -1109,7 +1109,8 @@ CREATE TABLE IF NOT EXISTS `im_messages` (
     `head` json DEFAULT NULL,
     `content` json DEFAULT NULL,
     PRIMARY KEY (`id`),
-    UNIQUE KEY `messages_sender_client_mid` (`from`,`client_msg_id`)
+    UNIQUE KEY `messages_sender_client_mid` (`from`,`client_msg_id`),
+    UNIQUE KEY `uk_im_messages_topic_seq` (`topic`,`seqid`)
     ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci;
 
 CREATE TABLE IF NOT EXISTS `im_monthly_reports` (

server/store/group_message_runtime.go

@@ -32,6 +32,14 @@ type redisConfig struct {
 	CacheTTL    int    `json:"cache_ttl_seconds"`
 	ShortTTL    int    `json:"short_state_ttl_seconds"`
 	GroupFanout bool   `json:"group_fanout_enabled"`
+
+	// Sentinel-aware connection. When set, the runtime uses a failover client
+	// instead of the single Addr. Required whenever Redis runs behind Sentinel:
+	// a plain client pinned to one address will hit replicas after failover,
+	// where writes fail with READONLY and reads return stale values (a stale
+	// seq counter can hand out already-persisted seq values).
+	SentinelMasterName string   `json:"sentinel_master_name"`
+	SentinelAddrs      []string `json:"sentinel_addrs"`
 }
 
 type groupMessageUniqueRecord struct {
@@ -77,6 +85,8 @@ type groupMessageRuntime interface {
 	Close() error
 	Claim(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) (*groupMessageUniqueRecord, bool, error)
 	NextSeq(tenantID types.TenantID, topic string, baseline int64) (int64, error)
+	PeekSeq(tenantID types.TenantID, topic string) (int64, error)
+	BumpSeqFloor(tenantID types.TenantID, topic string, seq int64) error
 	MarkPending(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64) error
 	MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error
 	MarkFailed(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string) error
@@ -126,10 +136,50 @@ func ReserveMessageSeq(tenantID types.TenantID, topic string, baseline int64) (i
 	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" {
 		return 0, false, nil
 	}
-	seq, err := groupMsgRuntime.NextSeq(tenantID, topic, baseline)
+	seq, err := nextMessageSeqWithFloor(tenantID, topic, baseline)
 	return seq, true, err
 }
 
+// TopicLastSeqID returns the seq last persisted for the topic in the DB, or 0
+// when the topic is unknown. It is used as a floor when re-seeding the Redis
+// seq counter.
+func TopicLastSeqID(tenantID types.TenantID, topic string) int64 {
+	if Topics == nil || !tenantID.IsValid() || topic == "" {
+		return 0
+	}
+	stopic, err := Topics.Get(tenantID, topic)
+	if err != nil || stopic == nil {
+		return 0
+	}
+	return stopic.SeqId
+}
+
+// nextMessageSeqWithFloor allocates the next seq for the topic. When the Redis
+// counter key is missing (Redis restart, eviction or flush) it re-seeds from
+// the DB topic seq instead of trusting the caller's in-memory baseline, which
+// may be stale and would otherwise hand out already-persisted seq values.
+func nextMessageSeqWithFloor(tenantID types.TenantID, topic string, baseline int64) (int64, error) {
+	if baseline < 0 {
+		baseline = 0
+	}
+	if current, err := groupMsgRuntime.PeekSeq(tenantID, topic); err == nil && current <= 0 {
+		if floor := TopicLastSeqID(tenantID, topic); floor > baseline {
+			baseline = floor
+		}
+	}
+	return groupMsgRuntime.NextSeq(tenantID, topic, baseline)
+}
+
+// BumpMessageSeqFloor pushes the Redis seq counter forward to seq (never
+// backward). Used after the async persister re-allocates a seq following a
+// uniqueness conflict so subsequent allocations do not collide again.
+func BumpMessageSeqFloor(tenantID types.TenantID, topic string, seq int64) error {
+	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || seq <= 0 {
+		return nil
+	}
+	return groupMsgRuntime.BumpSeqFloor(tenantID, topic, seq)
+}
+
 func MarkMessageUniqueDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
 	if groupMsgRuntime == nil || !tenantID.IsValid() || topic == "" || from.IsZero() || clientMsgID == "" {
 		return nil
@@ -407,15 +457,31 @@ func initGroupMessageRuntime(raw json.RawMessage) error {
 	dialTimeout := timeoutOrDefault(cfg.DialMS, 2*time.Second)
 	readTimeout := timeoutOrDefault(cfg.ReadMS, 2*time.Second)
 	writeTimeout := timeoutOrDefault(cfg.WriteMS, 2*time.Second)
-	client := redis.NewClient(&redis.Options{
-		Addr:         cfg.Addr,
-		Username:     cfg.Username,
-		Password:     cfg.Password,
-		DB:           cfg.DB,
-		DialTimeout:  dialTimeout,
-		ReadTimeout:  readTimeout,
-		WriteTimeout: writeTimeout,
-	})
+
+	var client redis.UniversalClient
+	if cfg.SentinelMasterName != "" && len(cfg.SentinelAddrs) > 0 {
+		client = redis.NewFailoverClient(&redis.FailoverOptions{
+			MasterName:       cfg.SentinelMasterName,
+			SentinelAddrs:    cfg.SentinelAddrs,
+			SentinelPassword: cfg.Password,
+			Username:         cfg.Username,
+			Password:         cfg.Password,
+			DB:               cfg.DB,
+			DialTimeout:      dialTimeout,
+			ReadTimeout:      readTimeout,
+			WriteTimeout:     writeTimeout,
+		})
+	} else {
+		client = redis.NewClient(&redis.Options{
+			Addr:         cfg.Addr,
+			Username:     cfg.Username,
+			Password:     cfg.Password,
+			DB:           cfg.DB,
+			DialTimeout:  dialTimeout,
+			ReadTimeout:  readTimeout,
+			WriteTimeout: writeTimeout,
+		})
+	}
 	ctx, cancel := context.WithTimeout(context.Background(), dialTimeout)
 	defer cancel()
 	if err := client.Ping(ctx).Err(); err != nil {
@@ -441,7 +507,7 @@ func timeoutOrDefault(ms int, fallback time.Duration) time.Duration {
 }
 
 type redisGroupMessageRuntime struct {
-	client *redis.Client
+	client redis.UniversalClient
 	prefix string
 	cfg    redisConfig
 }
@@ -509,6 +575,35 @@ return redis.call("INCR", KEYS[1])
 	return seq, nil
 }
 
+func (r *redisGroupMessageRuntime) PeekSeq(tenantID types.TenantID, topic string) (int64, error) {
+	value, err := r.client.Get(context.Background(), r.seqKey(tenantID, topic)).Result()
+	if err == redis.Nil {
+		return 0, nil
+	}
+	if err != nil {
+		return 0, err
+	}
+	seq, err := strconv.ParseInt(value, 10, 64)
+	if err != nil {
+		return 0, err
+	}
+	return seq, nil
+}
+
+func (r *redisGroupMessageRuntime) BumpSeqFloor(tenantID types.TenantID, topic string, seq int64) error {
+	if seq <= 0 {
+		return nil
+	}
+	_, err := r.client.Eval(context.Background(), `
+local current = redis.call("GET", KEYS[1])
+if not current or tonumber(current) < tonumber(ARGV[1]) then
+  redis.call("SET", KEYS[1], ARGV[1])
+end
+return redis.call("GET", KEYS[1])
+`, []string{r.seqKey(tenantID, topic)}, seq).Int64()
+	return err
+}
+
 func (r *redisGroupMessageRuntime) MarkDone(tenantID types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
 	if clientMsgID == "" {
 		return nil

server/store/group_message_runtime_test.go

@@ -43,6 +43,17 @@ func (r *testGroupMessageRuntime) NextSeq(_ types.TenantID, topic string, baseli
 	return r.seqs[topic], nil
 }
 
+func (r *testGroupMessageRuntime) PeekSeq(_ types.TenantID, topic string) (int64, error) {
+	return r.seqs[topic], nil
+}
+
+func (r *testGroupMessageRuntime) BumpSeqFloor(_ types.TenantID, topic string, seq int64) error {
+	if r.seqs[topic] < seq {
+		r.seqs[topic] = seq
+	}
+	return nil
+}
+
 func (r *testGroupMessageRuntime) MarkDone(_ types.TenantID, topic string, from types.Uid, clientMsgID string, seq int64, msgID string) error {
 	key := topic + ":" + from.UserId() + ":" + clientMsgID
 	r.claims[key] = &groupMessageUniqueRecord{Status: "done", Seq: seq, MessageID: msgID, CreatedAt: time.Now().UTC()}
@@ -246,6 +257,37 @@ func TestMessageRuntimeSeqAndClientMidIdempotency(t *testing.T) {
 	}
 }
 
+func TestBumpMessageSeqFloorNeverMovesBackward(t *testing.T) {
+	oldRuntime := groupMsgRuntime
+	runtime := newTestGroupMessageRuntime()
+	groupMsgRuntime = runtime
+	t.Cleanup(func() { groupMsgRuntime = oldRuntime })
+
+	tenantID := types.TenantID(7)
+	topic := types.Uid(1).P2PName(types.Uid(2))
+
+	seq, ready, err := ReserveMessageSeq(tenantID, topic, 0)
+	if err != nil || !ready || seq != 1 {
+		t.Fatalf("unexpected first seq: seq=%d ready=%v err=%v", seq, ready, err)
+	}
+	if err = BumpMessageSeqFloor(tenantID, topic, 5); err != nil {
+		t.Fatalf("bump floor: %v", err)
+	}
+	if current, err := runtime.PeekSeq(tenantID, topic); err != nil || current != 5 {
+		t.Fatalf("floor was not applied: current=%d err=%v", current, err)
+	}
+	if err = BumpMessageSeqFloor(tenantID, topic, 3); err != nil {
+		t.Fatalf("lower bump: %v", err)
+	}
+	if current, err := runtime.PeekSeq(tenantID, topic); err != nil || current != 5 {
+		t.Fatalf("floor must never move backward: current=%d err=%v", current, err)
+	}
+	// The next allocation must continue above the bumped floor.
+	if seq, _, err := ReserveMessageSeq(tenantID, topic, 0); err != nil || seq != 6 {
+		t.Fatalf("unexpected seq after bump: seq=%d err=%v", seq, err)
+	}
+}
+
 func TestInvalidateGroupMemberPagesUsesTopicKeyPrefix(t *testing.T) {
 	oldRuntime := groupMsgRuntime
 	runtime := newTestGroupMessageRuntime()

server/store/store.go

@@ -3597,7 +3597,7 @@ func (messagesMapper) PreparePersist(tenantID types.TenantID, msg *types.Message
 				return nil, types.ErrDuplicate
 			}
 		}
-		seq, err := groupMsgRuntime.NextSeq(tenantID, msg.Topic, msg.SeqId-1)
+		seq, err := nextMessageSeqWithFloor(tenantID, msg.Topic, msg.SeqId-1)
 		if err != nil {
 			if clientMsgID != "" {
 				_ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID)
@@ -3661,21 +3661,21 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag
 		err = tenantAdp.TenantMessageSave(tenantID, msg)
 	}
 	if err != nil {
-		if isGroupMessage && err == types.ErrDuplicate && clientMsgID != "" {
-			duplicate, dupErr := tenantAdp.TenantGroupMessageGetByClientID(tenantID, msg.Topic, fromUid, clientMsgID)
-			if dupErr != nil {
-				return nil, dupErr
-			}
-			if duplicate != nil {
-				msg.SeqId = duplicate.SeqId
-				if groupMsgRuntime != nil {
-					_ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, fromUid, clientMsgID, duplicate.SeqId, duplicate.Uid().String())
+		if err == types.ErrDuplicate {
+			if isGroupMessage && clientMsgID != "" {
+				duplicate, dupErr := tenantAdp.TenantGroupMessageGetByClientID(tenantID, msg.Topic, fromUid, clientMsgID)
+				if dupErr != nil {
+					return nil, dupErr
+				}
+				if duplicate != nil {
+					msg.SeqId = duplicate.SeqId
+					if groupMsgRuntime != nil {
+						_ = groupMsgRuntime.MarkDone(tenantID, msg.Topic, fromUid, clientMsgID, duplicate.SeqId, duplicate.Uid().String())
+					}
+					return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil
 				}
-				return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil
 			}
-		}
-		if !isGroupMessage && err == types.ErrDuplicate {
-			if clientMsgID != "" {
+			if !isGroupMessage && clientMsgID != "" {
 				duplicate, dupErr := tenantAdp.TenantMessageGetByClientID(tenantID, fromUid, clientMsgID)
 				if dupErr != nil {
 					return nil, dupErr
@@ -3685,12 +3685,14 @@ func (messagesMapper) PersistPrepared(tenantID types.TenantID, msg *types.Messag
 					return &types.MessageSaveResult{SeqId: duplicate.SeqId, Duplicate: true}, nil
 				}
 			}
-			if retryErr := retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg); retryErr != nil {
+			if retryErr := retryTenantMessageSaveAfterSeqConflict(tenantAdp, tenantID, msg, isGroupMessage); retryErr != nil {
+				if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" {
+					_ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID)
+				}
 				return nil, retryErr
 			}
 			err = nil
-		}
-		if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" {
+		} else if isGroupMessage && groupMsgRuntime != nil && clientMsgID != "" {
 			_ = groupMsgRuntime.MarkFailed(tenantID, msg.Topic, fromUid, clientMsgID)
 		}
 		if err != nil {
@@ -3786,19 +3788,52 @@ func NormalizeClientMsgID(head types.KVMap) (string, error) {
 	return normalizeClientMsgID(head)
 }
 
-func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, msg *types.Message) error {
-	latest, err := tenantAdp.TenantMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1})
+func retryTenantMessageSaveAfterSeqConflict(tenantAdp adapter.TenantBusinessAdapter, tenantID types.TenantID, msg *types.Message, isGroupMessage bool) error {
+	var latest []types.Message
+	var err error
+	if isGroupMessage {
+		latest, err = tenantAdp.TenantGroupMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1})
+	} else {
+		latest, err = tenantAdp.TenantMessageGetAll(tenantID, msg.Topic, types.ZeroUid, &types.QueryOpt{Before: 1 << 62, Limit: 1})
+	}
 	if err != nil {
 		return err
 	}
 	if len(latest) == 0 || latest[0].SeqId < msg.SeqId {
 		return types.ErrDuplicate
 	}
+	prevSeq := msg.SeqId
 	msg.SeqId = latest[0].SeqId + 1
 	if err = tenantAdp.TenantTopicUpdateOnMessage(tenantID, msg.Topic, msg); err != nil {
 		return err
 	}
-	return tenantAdp.TenantMessageSave(tenantID, msg)
+	if isGroupMessage {
+		err = tenantAdp.TenantGroupMessageSave(tenantID, msg)
+	} else {
+		err = tenantAdp.TenantMessageSave(tenantID, msg)
+	}
+	if err != nil {
+		return err
+	}
+	// The re-allocated seq differs from the one handed out before the insert
+	// failed. Sync the Redis allocator and the idempotency record with the
+	// final seq, otherwise subsequent publishes keep colliding with the
+	// persisted row and resends of the same client mid return the abandoned
+	// seq forever.
+	BumpMessageSeqFloor(tenantID, msg.Topic, msg.SeqId)
+	if isGroupMessage {
+		// Drop the pending hot-cache entry the producer wrote for the
+		// abandoned seq so hot-range reads cannot surface a phantom message.
+		if err := InvalidateHotMessages(tenantID, msg.Topic, []types.Range{{Low: prevSeq, Hi: prevSeq + 1}}); err != nil {
+			logs.Warn.Printf("topic[%s]: failed to invalidate abandoned hot message seq=%d: %v", msg.Topic, prevSeq, err)
+		}
+	}
+	if msg.ClientMsgID != "" {
+		if fromUid := types.ParseUid(msg.From); !fromUid.IsZero() {
+			_ = MarkMessageUniqueDone(tenantID, msg.Topic, fromUid, msg.ClientMsgID, msg.SeqId, msg.Uid().String())
+		}
+	}
+	return nil
 }
 
 func normalizeClientMsgID(head types.KVMap) (string, error) {