From d257fe5f11c4678047e0d86210c591181aeb7735 Mon Sep 17 00:00:00 2001
From: Hu Ying <ying.hu@shopee.com>
Date: Fri, 5 Apr 2024 14:57:58 +0800
Subject: [PATCH] Fix for issues #2959 and #2960

---
 osscluster.go | 6 +++++-
 1 file changed, 5 insertions(+), 1 deletion(-)

diff --git a/osscluster.go b/osscluster.go
index 17f98d9dc8..de1e3bf55c 100644
--- a/osscluster.go
+++ b/osscluster.go
@@ -1295,6 +1295,7 @@ func (c *ClusterClient) processPipelineNode(
 	_ = node.Client.withProcessPipelineHook(ctx, cmds, func(ctx context.Context, cmds []Cmder) error {
 		cn, err := node.Client.getConn(ctx)
 		if err != nil {
+			node.MarkAsFailing()
 			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 			setCmdsErr(cmds, err)
 			return err
@@ -1316,6 +1317,9 @@ func (c *ClusterClient) processPipelineNodeConn(
 	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
 		return writeCmds(wr, cmds)
 	}); err != nil {
+		if isBadConn(err, false, node.Client.getAddr()) {
+			node.MarkAsFailing()
+		}
 		if shouldRetry(err, true) {
 			_ = c.mapCmdsByNode(ctx, failedCmds, cmds)
 		}
@@ -1347,7 +1351,7 @@ func (c *ClusterClient) pipelineReadCmds(
 			continue
 		}
 
-		if c.opt.ReadOnly {
+		if c.opt.ReadOnly && isBadConn(err, false, node.Client.getAddr()) {
 			node.MarkAsFailing()
 		}