fix cluster proxy cleanup routing

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() {