fix: pipeline repeatedly sets the error

pipeline_test.go

@@ -60,6 +60,39 @@ var _ = Describe("pipelining", func() {
 Expect(cmds).To(BeEmpty())
 })
 
+It("pipeline: basic exec", func() {
+p := client.Pipeline()
+p.Get(ctx, "key")
+p.Set(ctx, "key", "value", 0)
+p.Get(ctx, "key")
+cmds, err := p.Exec(ctx)
+Expect(err).To(Equal(redis.Nil))
+Expect(cmds).To(HaveLen(3))
+Expect(cmds[0].Err()).To(Equal(redis.Nil))
+Expect(cmds[1].(*redis.StatusCmd).Val()).To(Equal("OK"))
+Expect(cmds[1].Err()).NotTo(HaveOccurred())
+Expect(cmds[2].(*redis.StringCmd).Val()).To(Equal("value"))
+Expect(cmds[2].Err()).NotTo(HaveOccurred())
+})
+
+It("pipeline: exec pipeline when get conn failed", func() {
+p := client.Pipeline()
+p.Get(ctx, "key")
+p.Set(ctx, "key", "value", 0)
+p.Get(ctx, "key")
+
+client.Close()
+
+cmds, err := p.Exec(ctx)
+Expect(err).To(Equal(redis.ErrClosed))
+Expect(cmds).To(HaveLen(3))
+for _, cmd := range cmds {
+Expect(cmd.Err()).To(Equal(redis.ErrClosed))
+}
+
+client = redis.NewClient(redisOptions())
+})
+
 assertPipeline := func() {
 It("returns no errors when there are no commands", func() {
 _, err := pipe.Exec(ctx)

redis.go

@@ -768,7 +768,10 @@ func (c *baseClient) generalProcessPipeline(
 return err
 })
 if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
-setCmdsErr(cmds, lastErr)
+// The error should be set here only when failing to obtain the conn.
+if !isRedisError(lastErr) {
+setCmdsErr(cmds, lastErr)
+}
 return lastErr
 }
 }
@@ -864,7 +867,7 @@ func (c *baseClient) txPipelineReadQueued(ctx context.Context, cn *pool.Conn, rd
 }
 
 // Parse +QUEUED.
- for _, cmd := range cmds {
+for _, cmd := range cmds {
 // To be sure there are no buffered push notifications, we process them before reading the reply
 if err := c.processPendingPushNotificationWithReader(ctx, cn, rd); err != nil {
 internal.Logger.Printf(ctx, "push: error processing pending notifications before reading reply: %v", err)