diff --git a/scripts/IM-租户业务库初始化.sql b/scripts/IM-租户业务库初始化.sql index fceb70d..6f40d2b 100644 --- a/scripts/IM-租户业务库初始化.sql +++ b/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` ( diff --git a/server/store/group_message_runtime.go b/server/store/group_message_runtime.go index e7bd539..22a959a 100644 --- a/server/store/group_message_runtime.go +++ b/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 diff --git a/server/store/group_message_runtime_test.go b/server/store/group_message_runtime_test.go index d7bac37..086a6e8 100644 --- a/server/store/group_message_runtime_test.go +++ b/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() diff --git a/server/store/store.go b/server/store/store.go index 5c6b6ef..e3c982d 100644 --- a/server/store/store.go +++ b/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) {