fix: avoid retrying ClusterClient Watch callback errors

osscluster.go

@@ -2110,10 +2110,18 @@ func (c *ClusterClient) Watch(ctx context.Context, fn func(*Tx) error, keys ...s
 			}
 		}
 
-		err = node.Client.Watch(ctx, fn, keys...)
+		// Track callback errors separately to avoid retrying user failures through cluster retry classification.
+		var fnErr error
+		err = node.Client.Watch(ctx, func(tx *Tx) error {
+			fnErr = fn(tx)
+			return fnErr
+		}, keys...)
 		if err == nil {
 			break
 		}
+		if fnErr != nil {
+			return fnErr
+		}
 
 		moved, ask, addr := isMovedError(err)
 		if moved || ask {

osscluster_test.go

@@ -5,6 +5,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"io"
 	"net"
 	"slices"
 	"strconv"
@@ -1355,6 +1356,27 @@ var _ = Describe("ClusterClient", func() {
 			Expect(info.Val()).Should(ContainSubstring("tcp_port:16601"))
 		})
 
+		It("should not retry Watch callback errors wrapping io.EOF", func() {
+			opt := redisClusterOptions()
+			opt.MaxRedirects = 1
+			opt.MinRetryBackoff = -1
+			opt.MaxRetryBackoff = -1
+
+			client := cluster.newClusterClient(ctx, opt)
+			defer func() {
+				Expect(client.Close()).NotTo(HaveOccurred())
+			}()
+
+			calls := 0
+			err := client.Watch(ctx, func(tx *redis.Tx) error {
+				calls++
+				return fmt.Errorf("external call failed: %w", io.EOF)
+			}, "watch-callback-net-error")
+
+			Expect(errors.Is(err, io.EOF)).To(BeTrue())
+			Expect(calls).To(Equal(1))
+		})
+
 		assertClusterClient()
 	})