From 6729f40665fbe0d747b4d651c7023bb70b96bc48 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Wed, 16 Sep 2026 16:21:53 +0800 Subject: [PATCH] fix: sync topic sequence after server data --- server/topic.go | 6 ++++++ server/topic_test.go | 39 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 45 insertions(+) diff --git a/server/topic.go b/server/topic.go index bf87d6e..2e161fd 100644 --- a/server/topic.go +++ b/server/topic.go @@ -680,6 +680,12 @@ func (t *Topic) handleServerMsg(msg *ServerComMessage) { if msg.Pres != nil { t.handlePresence(msg) } else if msg.Info != nil || msg.Data != nil { + if msg.Data != nil && msg.Data.SeqId > t.lastID { + t.lastID = msg.Data.SeqId + if !msg.Data.Timestamp.IsZero() { + t.touched = msg.Data.Timestamp + } + } if msg.Info != nil && msg.Info.What == "group" { // Group management event published by the HTTP or Kafka admin // interface: update in-memory topic state (cached permissions, diff --git a/server/topic_test.go b/server/topic_test.go index d0fab14..438e541 100644 --- a/server/topic_test.go +++ b/server/topic_test.go @@ -1866,6 +1866,45 @@ func TestHandleBroadcastPresMe(t *testing.T) { } } +func TestHandleServerDataAdvancesLastID(t *testing.T) { + topicName := "p2pZAAAAAAAAABAAAAAAAAAAA" + helper := TopicTestHelper{} + helper.setUp(t, 2, types.TopicCatP2P, topicName, true) + defer helper.tearDown() + + timestamp := time.Now().UTC().Round(time.Millisecond) + helper.topic.handleServerMsg(&ServerComMessage{ + TenantID: types.TenantID(1), + RcptTo: topicName, + PreserveTopicName: true, + Data: &MsgServerData{ + Topic: topicName, + From: types.ZeroUid.UserId(), + Timestamp: timestamp, + SeqId: 1, + Head: map[string]any{"x-im-type": "system_notice"}, + Content: map[string]any{"text": "你们已成为好友,现在可以开始聊天了。"}, + }, + }) + helper.finish() + + if helper.topic.lastID != 1 { + t.Fatalf("Topic.lastID: expected server data to advance to 1, got %d", helper.topic.lastID) + } + if !helper.topic.touched.Equal(timestamp) { + t.Fatalf("Topic.touched: expected %s, got %s", timestamp, helper.topic.touched) + } + for i, result := range helper.results { + if len(result.messages) != 1 { + t.Fatalf("session %d expected one server data message, got %d", i, len(result.messages)) + } + msg := result.messages[0].(*ServerComMessage) + if msg.Data == nil || msg.Data.SeqId != 1 { + t.Fatalf("session %d received unexpected message: %#v", i, msg) + } + } +} + func TestHandleBroadcastPresInactiveTopic(t *testing.T) { topicName := "usrMe" numUsers := 1