Fixed BUG

Author: dobyte

cluster/gate/proxy.go

@@ -84,7 +84,6 @@ func (p *proxy) deliver(ctx context.Context, cid, uid int64, buf buffer.Buffer)
 		Route:  message.Route,
 		Buffer: buf,
 	}); err != nil {
-		buf.Release()
 		switch {
 		case errors.Is(err, errors.ErrNotFoundRoute), errors.Is(err, errors.ErrNotFoundEndpoint):
 			log.Warnf("deliver message failed, cid: %d uid: %d seq: %d route: %d err: %v", cid, uid, message.Seq, message.Route, err)

internal/link/node.go

@@ -167,8 +167,9 @@ func (l *NodeLinker) FetchNodeList(ctx context.Context, states ...cluster.State)
 // Deliver 投递消息给节点处理
 func (l *NodeLinker) Deliver(ctx context.Context, args *DeliverArgs) error {
 	var (
-		err error
-		buf buffer.Buffer
+		err       error
+		buf       buffer.Buffer
+		isDeliver bool
 	)
 
 	switch b := args.Buffer.(type) {
@@ -187,15 +188,24 @@ func (l *NodeLinker) Deliver(ctx context.Context, args *DeliverArgs) error {
 	if args.NID != "" {
 		client, err := l.doBuildClient(args.NID)
 		if err != nil {
+			buf.Release()
 			return err
 		}
 
 		return client.Deliver(ctx, args.CID, args.UID, buf)
 	} else {
 		if _, err = l.doRPC(ctx, args.Route, args.UID, func(ctx context.Context, client *node.Client) (bool, any, error) {
+			isDeliver = true
+
 			return false, nil, client.Deliver(ctx, args.CID, args.UID, buf)
-		}); err != nil && !errors.Is(err, errors.ErrNotFoundUserLocation) {
-			return err
+		}); err != nil {
+			if !isDeliver {
+				buf.Release()
+			}
+
+			if !errors.Is(err, errors.ErrNotFoundUserLocation) {
+				return err
+			}
 		}
 
 		return nil

internal/transporter/internal/client/conn.go

@@ -121,16 +121,19 @@ func (c *Conn) handshake(conn net.Conn) error {
 	)
 
 	buf := protocol.EncodeHandshakeReq(seq, c.cli.opts.InsKind, c.cli.opts.InsID)
-	defer buf.Release()
 
 	c.pending.store(seq, call)
 
 	if _, err := conn.Write(buf.Bytes()); err != nil {
+		buf.Release()
+
 		close(call)
 
 		c.pending.delete(seq)
 
 		return err
+	} else {
+		buf.Release()
 	}
 
 	ctx, cancel := context.WithTimeout(context.Background(), defaultTimeout)
@@ -139,7 +142,11 @@ func (c *Conn) handshake(conn net.Conn) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-call:
+	case buf := <-call:
+		if buf != nil {
+			buf.Release()
+		}
+
 		return nil
 	}
 }

network/tcp/server_conn.go

@@ -339,8 +339,8 @@ func (c *serverConn) read() {
 
 			isHeartbeat, err := packet.CheckHeartbeat(buf.Bytes())
 			if err != nil {
-				log.Errorf("check heartbeat message error: %v", err)
 				buf.Release()
+				log.Errorf("check heartbeat message error: %v", err)
 				continue
 			}