Repository: redis/go-redis

Issue:
Pipeline does not propogate errors to all commands properly
When using a pipeline, previous versions (v8.*.*) used to set error on all commands in a pipeline when it failed. go-redis v9 does not call `setCmdsErr` in some cases.

## Expected Behavior

All commands in the pipeline should have err set properly and `cmd.Err()` should return the error instead of nil for all cmds

## Current Behavior

Commands in a failed pipeline return `"", nil` on calling `Result()`

## Steps to Reproduce

```go
package main

import (
	"context"
	"fmt"

	"github.com/redis/go-redis/v9"
)

func main() {
	fmt.Print("Initializing Redis client... ")
	rdb := redis.NewClient(&redis.Options{
		Addr:       "localhost:6379",
	})

	// closing client here to emulate error
	rdb.Close()

	p := rdb.Pipeline()
	cmd1 := p.Get(context.Background(), "get1")
	cmd2 := p.Get(context.Background(), "get2")

	_, errPipe := p.Exec(context.Background())

	err1 := cmd1.Err()
	err2 := cmd2.Err()

	fmt.Printf("Errors = %s, %s, %s\n", errPipe, err1, err2)
}
```

## Context (Environment)

I was trying to use `cmds.Result()` directly (ignoring return val of `pipe.Exec()`) and checking for errors in the result itself.

(I don't know if this is the intended behavior or a regression. So sorry if it's the former)

I did a bisect on the code and I could narrow it down to https://github.com/redis/go-redis/commit/dd9a200427d7c57e480d61833a7204b13b55acbd . 

```
Output of commit#6327c52e60cabd2acb43dcacab389def4e3e9a5a
Errors = redis: client is closed, redis: client is closed, redis: client is closed

Output of commit#dd9a200427d7c57e480d61833a7204b13b55acbd
Errors = redis: client is closed, %!s(<nil>), %!s(<nil>)
```


Also, this happens on some types of clients only

```
// before
Errors NewClusterClient = redis: client is closed, redis: client is closed, redis: client is closed
Errors NewFailoverClient = redis: client is closed, redis: client is closed, redis: client is closed
Errors NewClient = redis: client is closed, redis: client is closed, redis: client is closed
Errors NewRing = redis: client is closed, redis: client is closed, redis: client is closed

// after
Errors NewClusterClient = redis: client is closed, redis: client is closed, redis: client is closed
Errors NewFailoverClient = redis: client is closed, %!s(<nil>), %!s(<nil>)
Errors NewClient = redis: client is closed, %!s(<nil>), %!s(<nil>)
Errors NewRing = redis: client is closed, redis: client is closed, redis: client is closed
```

Thanks

Hints from issue comments:
@seruman: Observed the same exact behaviour, here's what I could find;


Seems like the closure here (`p == pipelineProcessCmds | txPipelineProcessCmds`) is never called if the client is closed; AFAIU it is expected for processer `p` to set command errors but it never gets called with a closed client.

https://github.com/redis/go-redis/blob/2d8fa02ac2b0cda80410674a4abad00ea35217e8/redis.go#L515-L519

In ClusterClient, this seems to be the path that the errors get set if the client is closed;
https://github.com/redis/go-redis/blob/2d8fa02ac2b0cda80410674a4abad00ea35217e8/osscluster.go#L1221-L1224


@ndyakov: Thank you for reporting this. @seruman feel free to open a PR with the fix if you think you have narrowed this down or write a test that can be used to check the expected behaviour. @sgp: So, I'm going to try to fix this _in anger_ in the next week or so because I've been bitten by this for over a year now and I'm tired of it.

One thing I want to do as a good citizen is to make a unit/integration test that tests the problem and validates the fix, but there are issues with running `make test` in the main branch that are maddening (some tests never run, the docker-compose.yml starts three of the five pods with arguments that fail, etc.). I'm going to open another issue to track this.@sgp: Opened #3490 to track integration / containerized test issues.@tptodorov: I experienced the same problem with `github.com/redis/go-redis/v9 v9.13.0`


```
	pipe := cache.redisClient.TxPipeline()

	// Set the entry data
	pipe.HSet(ctx, key, entryMap)

	// Set TTL if specified
	if ttl > 0 {
		pipe.PExpire(ctx, key, time.Duration(ttl)*time.Millisecond)
	} else {
		pipe.Persist(ctx, key)
	}

	// Execute the transaction
	cmdErrors, err := pipe.Exec(ctx)
```

All errors in cmdErrors contain the last error returned in the pipeline instead of errors for each command.

<img width="763" height="428" alt="Image" src="https://github.com/user-attachments/assets/ddfdbf63-fddc-4333-a97c-1f4024ede1fc" />


--- 

For comparison, here is what I get when I run the same commands from CLI

```
multi
OK
0.0.0.0:64687(TX)> hset abc 122 121
(error) OOM command not allowed when used memory > 'maxmemory'.
0.0.0.0:64687(TX)> exec
(error) EXECABORT Transaction discarded because of previous errors.
```

I need to be able to get to the actual error "OOM command not allowed when used memory > 'maxmemory'."

@ndyakov: @tptodorov thank you for reporting and adding the redis cli example. It looks like the error on queueing the command is ignored if it is a redis error. https://github.com/redis/go-redis/blob/65e1c22065050e7390350482f41728f470fe7994/redis.go#L707

You can check the potential fix in #3509 @ndyakov: Just a note, that those are two separate problems, but both should be fixed in the same PR. 

- The first one reported by @AvyChanna was related to first error which was not supposed to be retried, especially if it is a client error before being able to get a connection, to not be assigned to the commands.

- The second one, reported by @tptodorov, was related to `TxPipeline` which was ignoring errors when queueing commands.

Both are addressed in the PR. 


Solution patch:
diff --git a/osscluster.go b/osscluster.go
index 092260e..7c5a1a9 100644
--- a/osscluster.go
+++ b/osscluster.go
@@ -1712,10 +1712,16 @@ func (c *ClusterClient) txPipelineReadQueued(
 
 	for _, cmd := range cmds {
 		err := statusCmd.readReply(rd)
-		if err == nil || c.checkMovedErr(ctx, cmd, err, failedCmds) || isRedisError(err) {
-			continue
+		if err != nil {
+			if c.checkMovedErr(ctx, cmd, err, failedCmds) {
+				// will be processed later
+				continue
+			}
+			cmd.SetErr(err)
+			if !isRedisError(err) {
+				return err
+			}
 		}
-		return err
 	}
 
 	// Parse number of replies.
diff --git a/pipeline.go b/pipeline.go
index dbbced5..567bf12 100644
--- a/pipeline.go
+++ b/pipeline.go
@@ -30,9 +30,12 @@ type Pipeliner interface {
 	// If a certain Redis command is not yet supported, you can use Do to execute it.
 	Do(ctx context.Context, args ...interface{}) *Cmd
 
-	// Process puts the commands to be executed into the pipeline buffer.
+	// Process queues the cmd for later execution.
 	Process(ctx context.Context, cmd Cmder) error
 
+	// BatchProcess adds multiple commands to be executed into the pipeline buffer.
+	BatchProcess(ctx context.Context, cmd ...Cmder) error
+
 	// Discard discards all commands in the pipeline buffer that have not yet been executed.
 	Discard()
 
@@ -79,7 +82,12 @@ func (c *Pipeline) Do(ctx context.Context, args ...interface{}) *Cmd {
 
 // Process queues the cmd for later execution.
 func (c *Pipeline) Process(ctx context.Context, cmd Cmder) error {
-	c.cmds = append(c.cmds, cmd)
+	return c.BatchProcess(ctx, cmd)
+}
+
+// BatchProcess queues multiple cmds for later execution.
+func (c *Pipeline) BatchProcess(ctx context.Context, cmd ...Cmder) error {
+	c.cmds = append(c.cmds, cmd...)
 	return nil
 }
 
diff --git a/redis.go b/redis.go
index 43ab401..16e2130 100644
--- a/redis.go
+++ b/redis.go
@@ -630,6 +630,7 @@ func (c *baseClient) generalProcessPipeline(
 			return err
 		})
 		if lastErr == nil || !canRetry || !shouldRetry(lastErr, true) {
+			setCmdsErr(cmds, lastErr)
 			return lastErr
 		}
 	}
@@ -703,9 +704,12 @@ func txPipelineReadQueued(rd *proto.Reader, statusCmd *StatusCmd, cmds []Cmder)
 	}
 
 	// Parse +QUEUED.
-	for range cmds {
-		if err := statusCmd.readReply(rd); err != nil && !isRedisError(err) {
-			return err
+	for _, cmd := range cmds {
+		if err := statusCmd.readReply(rd); err != nil {
+			cmd.SetErr(err)
+			if !isRedisError(err) {
+				return err
+			}
 		}
 	}
 



Test patch:
diff --git a/pipeline_test.go b/pipeline_test.go
index d32ab35..15eacb3 100644
--- a/pipeline_test.go
+++ b/pipeline_test.go
@@ -114,6 +114,25 @@ var _ = Describe("pipelining", func() {
 			err := pipe.Do(ctx).Err()
 			Expect(err).To(Equal(errors.New("redis: please enter the command to be executed")))
 		})
+
+		It("should process", func() {
+			err := pipe.Process(ctx, redis.NewCmd(ctx, "asking"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(1))
+		})
+
+		It("should batchProcess", func() {
+			err := pipe.BatchProcess(ctx, redis.NewCmd(ctx, "asking"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(1))
+
+			pipe.Discard()
+			Expect(pipe.Cmds()).To(HaveLen(0))
+
+			err = pipe.BatchProcess(ctx, redis.NewCmd(ctx, "asking"), redis.NewCmd(ctx, "set", "key", "value"))
+			Expect(err).To(BeNil())
+			Expect(pipe.Cmds()).To(HaveLen(2))
+		})
 	}
 
 	Describe("Pipeline", func() {