From 75216c311d2ffa66f27c4dc994bbfa7cc3c48c97 Mon Sep 17 00:00:00 2001 From: richjohnrambo Date: Thu, 24 Sep 2026 10:18:07 +0800 Subject: [PATCH] fix cluster proxy cleanup routing --- server/cluster.go | 39 ++++++++++++++++----- server/cluster_gob_test.go | 72 ++++++++++++++++++++++++++++++++++++++ server/topic_proxy.go | 5 +-- 3 files changed, 105 insertions(+), 11 deletions(-) create mode 100644 server/cluster_gob_test.go diff --git a/server/cluster.go b/server/cluster.go index df29d9f..d662eee 100644 --- a/server/cluster.go +++ b/server/cluster.go @@ -36,6 +36,8 @@ const ( clusterRpcCompletionBuffer = 64 ) +var errClusterTopicOwnerUnavailable = errors.New("cluster topic owner unavailable") + // ProxyReqType is the type of proxy requests. type ProxyReqType int @@ -926,9 +928,15 @@ func (c *Cluster) routeToTopicMaster(reqType ProxyReqType, msg *ClientComMessage req := c.makeClusterReq(reqType, msg, topic, sess) // Find the cluster node which owns the topic, then forward to it. - n := c.nodeForTopic(types.TopicKey{TenantID: req.TenantID, Topic: topic}) + key := types.TopicKey{TenantID: req.TenantID, Topic: topic} + if reqType == ProxyReqLeave && !c.isRemoteTopic(key) { + // The proxy topic is stale after rehash/failover. The local session has already been + // detached, and there is no remote master to notify anymore. + return errClusterTopicOwnerUnavailable + } + n := c.nodeForTopic(key) if n == nil { - return errors.New("node for topic not found") + return errClusterTopicOwnerUnavailable } return n.proxyToMasterAsync(req) } @@ -967,9 +975,14 @@ func (c *Cluster) topicProxyGone(tenantID types.TenantID, topicName string) erro } // Find the cluster node which owns the topic, then forward to it. - n := c.nodeForTopic(types.TopicKey{TenantID: tenantID, Topic: topicName}) + key := types.TopicKey{TenantID: tenantID, Topic: topicName} + if !c.isRemoteTopic(key) { + // The proxy topic is stale after rehash/failover. Nothing remote needs to be cleaned. + return errClusterTopicOwnerUnavailable + } + n := c.nodeForTopic(key) if n == nil { - return errors.New("node for topic not found") + return errClusterTopicOwnerUnavailable } req := c.makeClusterReq(ProxyReqLeave, nil, topicName, &Session{tenantID: tenantID}) @@ -1015,11 +1028,7 @@ func clusterInit(configString json.RawMessage, self *string) int { return 1 } - gob.Register([]any{}) - gob.Register(map[string]any{}) - gob.Register(map[string]int{}) - gob.Register(map[string]string{}) - gob.Register(MsgAccessMode{}) + registerClusterGobTypes() if config.NumProxyEventGoRoutines != 0 { logs.Warn.Println("Cluster config: field num_proxy_event_goroutines is deprecated.") @@ -1072,6 +1081,18 @@ func clusterInit(configString json.RawMessage, self *string) int { return workerId } +func registerClusterGobTypes() { + gob.Register([]any{}) + gob.Register(map[string]any{}) + gob.Register(map[string]int{}) + gob.Register(map[string]string{}) + gob.Register(MsgAccessMode{}) + gob.Register(imSystemMessageActor{}) + gob.Register([]imSystemMessageActor{}) + gob.Register(imSystemMessageAction{}) + gob.Register([]imSystemMessageAction{}) +} + // Proxied session is being closed at the Master node. func (sess *Session) closeRPC() { if sess.isMultiplex() { diff --git a/server/cluster_gob_test.go b/server/cluster_gob_test.go new file mode 100644 index 0000000..9b864b3 --- /dev/null +++ b/server/cluster_gob_test.go @@ -0,0 +1,72 @@ +package main + +import ( + "bytes" + "encoding/gob" + "errors" + "testing" + "time" + + "github.com/tinode/chat/server/store/types" +) + +func TestClusterGobRegistersSystemMessagePayloadTypes(t *testing.T) { + registerClusterGobTypes() + + msg := &ClusterRoute{ + SrvMsg: &ServerComMessage{ + TenantID: types.TenantID(10001), + RcptTo: "grpTest", + Data: &MsgServerData{ + Topic: "grpTest", + Timestamp: time.Now().UTC().Round(time.Millisecond), + SeqId: 1, + Head: map[string]any{ + "x-im-type": "system_action", + "x-im-template": "group_member_joined", + }, + Content: map[string]any{ + "text": "成员加入了群聊", + "template": "group_member_joined", + "actors": []imSystemMessageActor{{ + UserID: "usr123", + Click: map[string]any{"type": "user_profile", "target_id": "usr123"}, + }}, + "actions": []imSystemMessageAction{{ + Type: "group_profile", + Label: "查看群聊", + TargetID: "grpTest", + }}, + }, + }, + }, + } + + var buf bytes.Buffer + if err := gob.NewEncoder(&buf).Encode(msg); err != nil { + t.Fatalf("encode cluster route with system message payload: %v", err) + } + + var decoded ClusterRoute + if err := gob.NewDecoder(&buf).Decode(&decoded); err != nil { + t.Fatalf("decode cluster route with system message payload: %v", err) + } +} + +func TestClusterRouteLeaveIgnoresStaleLocalProxyOwner(t *testing.T) { + cluster := &Cluster{ + thisNodeName: "node-a", + nodes: map[string]*ClusterNode{}, + } + cluster.rehash([]string{"node-a"}) + + err := cluster.routeToTopicMaster(ProxyReqLeave, nil, "grpStaleProxy", &Session{tenantID: types.TenantID(10001)}) + if !errors.Is(err, errClusterTopicOwnerUnavailable) { + t.Fatalf("expected stale proxy leave to return owner unavailable, got %v", err) + } + + err = cluster.topicProxyGone(types.TenantID(10001), "grpStaleProxy") + if !errors.Is(err, errClusterTopicOwnerUnavailable) { + t.Fatalf("expected stale proxy shutdown to return owner unavailable, got %v", err) + } +} diff --git a/server/topic_proxy.go b/server/topic_proxy.go index 22b819c..e3f83f2 100644 --- a/server/topic_proxy.go +++ b/server/topic_proxy.go @@ -7,6 +7,7 @@ package main import ( + "errors" "net/http" "time" @@ -101,7 +102,7 @@ func (t *Topic) runProxy(hub *Hub) { s.detachSession(t.name) } - if err := globals.cluster.topicProxyGone(t.tenantID, t.name); err != nil { + if err := globals.cluster.topicProxyGone(t.tenantID, t.name); err != nil && !errors.Is(err, errClusterTopicOwnerUnavailable) { logs.Warn.Printf("proxy topic[%s] shutdown: failed to notify master - %s", t.name, err) } @@ -164,7 +165,7 @@ func (t *Topic) handleProxyLeaveRequest(msg *ClientComMessage, killTimer *time.T } } - if err := globals.cluster.routeToTopicMaster(ProxyReqLeave, msg, t.name, msg.sess); err != nil { + if err := globals.cluster.routeToTopicMaster(ProxyReqLeave, msg, t.name, msg.sess); err != nil && !errors.Is(err, errClusterTopicOwnerUnavailable) { logs.Warn.Printf("proxy topic[%s]: route leave request from proxy to master failed - %s", t.name, err) } if len(t.sessions) == 0 {