Repository: redis/go-redis

Issue:
Expose pipeline command results via Cmds() method
## Motivation  
We are building a **transaction package manager** on top of `go-redis` that orchestrates pipelines and Postgres transactions together.  
For this integration, we need a way to **retrieve the list of commands** that were queued in a pipeline.  

At the moment, `pipeline.go` stores the commands in a private field. Because it is unexported, we cannot access or inspect the collected commands outside the package.  

## Proposal  
Introduce a method like:  
```go
func (p *Pipeline) Cmds() []Cmder

Discussion:
@cxljs: @ndyakov Do you think this feature is necessary to add? If so, I'd be happy to implement it.@htemelski-oss: Currently, it's a bit inconvenient to keep track of the commands executed using a pipeline, so I think such a feature will be beneficial for the library.  But let's hear @ndyakov's opinion too.@ndyakov: @RasankRam do you need access to the underlying `[]Cmder` or a clone of it? My concerns is that if we grant access to the underlying `[]Cmder` they may be altered and the pipeline may fails. @RasankRam: @ndyakov I need access to `cmds` so that when a savepoint is released, the commands can be moved to the parent pipeline. I understand your concern that someone might “mess with” the underlying array, e.g. `cmds[0] = "someCmd"`.

I’d be fine with something like (if there are something problems with granting cmds):

```go
func (p *Pipeline) Cmds() []redis.Cmder {
    out := append([]redis.Cmder(nil), p.cmds...)
    return out[:len(out):len(out)]
}
```

If the internal logic of the package makes returning cmds undesirable, I’d also be fine with something like:

```go
func (p *Pipeline) StealCmds() []redis.Cmder {
    cmds := p.cmds
    p.cmds = p.cmds[:0] // or nil
    return cmds
}
```

All I need is to move commands from a child savepoint into the parent savepoint (if it exists, otherwise into the transaction).

<img width="361" height="202" alt="Image" src="https://github.com/user-attachments/assets/1fbe1c0c-68e4-4161-9a0b-f5ef2905e7e5" />@ndyakov: @htemelski-redis will have the final say, after a brief discussion I don't think there is an issue to exposing the Cmder slice. At the end of the day, reordering it or altering it can be considered a feature and any issues that may arise are the users responsibility. We should, however, be careful to not allow data races on this slice. 

@cxljs once @htemelski-redis decides (he will look more indepth at the implementation at the moment), feel free to work on this, thank you.@htemelski-oss: Hello @RasankRam,
The pipeline is already marked as not being thread-safe, and I don't think it is going to be an issue to create a way to expose the underlying `queue` of commands in the pipeline.
Have you thought of a way to append the commands from one pipeline to another?

Would something like that satisfy your use case?
```go
func (c *Pipeline) AppendTo(dst *Pipeline) {
	dst.cmds = append(dst.cmds, c.cmds...)
	c.Discard()
}
```

@ndyakov 
Also, if we are adding mechanisms to modify the underlying queue, what do you think about adding a `Pop` method to remove the last command?@RasankRam: @htemelski-redis Thank you! It will be quite enough for us. Though it seems a bit narrow for our case. But I can’t think of another situation where commands inside TxPipeline would be needed.@RasankRam: @htemelski-redis After giving it some thought, we do need the commands. Since we’ll be bringing the cache into a consistent state by flushing Redis commands, and we’ll be retrieving these commands from the pipeline. The architecture in this case will look like this: commit -> redisPipe.Exec() -failed-> Kafka producer (Kafka receives the set of commands from the pipeline that just need to be flushed).

<img width="462" height="292" alt="Image" src="https://github.com/user-attachments/assets/8fbae337-266d-4ef0-8760-b23b404ed380" />

Then, when the service tries to fetch something from the cache, it will get it from the database and then put the actual data into the cache.@cxljs: > func (c *Pipeline) AppendTo(dst *Pipeline) {
> 	dst.cmds = append(dst.cmds, c.cmds...)
> 	c.Discard()
> }

Pipeline has a similar API:
```go
// Process queues the cmd for later execution.
func (c *Pipeline) Process(ctx context.Context, cmd Cmder) error {
	c.cmds = append(c.cmds, cmd)
	return nil
}
```
@cxljs: In the scenario "commit → redisPipe.Exec() failed → Kafka producer",  @RasankRam  you don't need a new Cmds API, you can just like:
```go
cmds, err := redisPipe.Exec()
if err != nil {
    // send cmds to kafka.
}
```
Is there any scenario where Cmds need to be obtained before executing the pipeline?@RasankRam: @cxljs Regarding redisPipe.Exec, Thank you! I didn’t consider first argument when .Exec(). Then I’m left with one problem: transferring commands when releasing a savepoint into the parent savepoint/transaction@cxljs: > Then I’m left with one problem: transferring commands when releasing a savepoint into the parent savepoint/transaction

Thanks for your feedback @RasankRam . So in this scenario, we still need a new Cmds API.

And I missed a detail earlier. If Exec() is executed, we need to reset the error before sending Cmds to Kafka to avoid propagating the error.
```go
cmds, err := redisPipe.Exec()
if err != nil {
    // reset error.
    for i := range cmds {
        cmds[i].SetErr(nil)
    }
    // send cmds to kafka.
}
```@RasankRam: Thanks for the clarification! For now, we’re using a wrapper around go-redis.Pipeline to store commands:

```go
type pipeline struct {
    redis.Pipeliner
    redisCmds []redis.Cmder
}

func (p *pipeline) Process(ctx context.Context, cmd redis.Cmder) error {
    p.redisCmds = append(p.redisCmds, cmd)
    return p.Pipeliner.Process(ctx, cmd)
}
```

If your MR gets approved (or generally if there’s any solution to my issue, even with a different implementation), we’ll be able to remove one workaround.

No matter the outcome, I want to sincerely thank all the developers and contributors of go-redis for pushing the open-source project forward.

Solution patch:
diff --git a/commands.go b/commands.go
index c035800..e9fd0f2 100644
--- a/commands.go
+++ b/commands.go
@@ -253,6 +253,7 @@ var (
 	_ Cmdable = (*Tx)(nil)
 	_ Cmdable = (*Ring)(nil)
 	_ Cmdable = (*ClusterClient)(nil)
+	_ Cmdable = (*Pipeline)(nil)
 )
 
 type cmdable func(ctx context.Context, cmd Cmder) error
diff --git a/pipeline.go b/pipeline.go
index 1c11420..dbbced5 100644
--- a/pipeline.go
+++ b/pipeline.go
@@ -7,7 +7,7 @@ import (
 
 type pipelineExecer func(context.Context, []Cmder) error
 
-// Pipeliner is an mechanism to realise Redis Pipeline technique.
+// Pipeliner is a mechanism to realise Redis Pipeline technique.
 //
 // Pipelining is a technique to extremely speed up processing by packing
 // operations to batches, send them at once to Redis and read a replies in a
@@ -23,21 +23,24 @@ type pipelineExecer func(context.Context, []Cmder) error
 type Pipeliner interface {
 	StatefulCmdable
 
-	// Len is to obtain the number of commands in the pipeline that have not yet been executed.
+	// Len obtains the number of commands in the pipeline that have not yet been executed.
 	Len() int
 
 	// Do is an API for executing any command.
 	// If a certain Redis command is not yet supported, you can use Do to execute it.
 	Do(ctx context.Context, args ...interface{}) *Cmd
 
-	// Process is to put the commands to be executed into the pipeline buffer.
+	// Process puts the commands to be executed into the pipeline buffer.
 	Process(ctx context.Context, cmd Cmder) error
 
-	// Discard is to discard all commands in the cache that have not yet been executed.
+	// Discard discards all commands in the pipeline buffer that have not yet been executed.
 	Discard()
 
-	// Exec is to send all the commands buffered in the pipeline to the redis-server.
+	// Exec sends all the commands buffered in the pipeline to the redis server.
 	Exec(ctx context.Context) ([]Cmder, error)
+
+	// Cmds returns the list of queued commands.
+	Cmds() []Cmder
 }
 
 var _ Pipeliner = (*Pipeline)(nil)
@@ -119,3 +122,7 @@ func (c *Pipeline) TxPipelined(ctx context.Context, fn func(Pipeliner) error) ([
 func (c *Pipeline) TxPipeline() Pipeliner {
 	return c
 }
+
+func (c *Pipeline) Cmds() []Cmder {
+	return c.cmds
+}



Test patch:
diff --git a/pipeline_test.go b/pipeline_test.go
index 7f73447..d32ab35 100644
--- a/pipeline_test.go
+++ b/pipeline_test.go
@@ -36,6 +36,30 @@ var _ = Describe("pipelining", func() {
 		Expect(get.Val()).To(Equal(""))
 	})
 
+	It("exports queued commands", func() {
+		p := client.Pipeline()
+		cmds := p.Cmds()
+		Expect(cmds).To(BeEmpty())
+
+		p.Set(ctx, "foo", "bar", 0)
+		p.Get(ctx, "foo")
+		cmds = p.Cmds()
+		Expect(cmds).To(HaveLen(p.Len()))
+		Expect(cmds[0].Name()).To(Equal("set"))
+		Expect(cmds[1].Name()).To(Equal("get"))
+
+		cmds, err := p.Exec(ctx)
+		Expect(err).NotTo(HaveOccurred())
+		Expect(cmds).To(HaveLen(2))
+		Expect(cmds[0].Name()).To(Equal("set"))
+		Expect(cmds[0].(*redis.StatusCmd).Val()).To(Equal("OK"))
+		Expect(cmds[1].Name()).To(Equal("get"))
+		Expect(cmds[1].(*redis.StringCmd).Val()).To(Equal("bar"))
+
+		cmds = p.Cmds()
+		Expect(cmds).To(BeEmpty())
+	})
+
 	assertPipeline := func() {
 		It("returns no errors when there are no commands", func() {
 			_, err := pipe.Exec(ctx)