feat(tx): skip redundant UNWATCH in Tx.Close when no WATCH is active (#3854)

Tx.Close issued UNWATCH unconditionally before returning the connection to the pool, adding a round trip to every Watch transaction even when nothing was watched or when EXEC had already discarded the watched keys. Track whether a WATCH is still active on the transaction (watchArmed) and only issue UNWATCH from Close when it is. The flag is set on a successful WATCH and cleared only when EXEC actually ran: a nil error (committed), TxFailedErr (a watched key changed, EXEC returned nil), or an EXECABORT error (a queued command was rejected, EXEC discarded the transaction) — all of which release the watched keys server-side. Any other error may be reported before EXEC executes (for example -LOADING on the MULTI reply) or on a broken connection, so the flag is left set and Close still sends UNWATCH rather than risk leaving a watch on a pooled connection. This removes the extra UNWATCH on the common WATCH/.../EXEC path and on the no-key Watch path, and never returns a connection to the pool with an active WATCH. No exported API changes; the only wire difference is the elided no-op UNWATCH. Tests assert, on the wire (recording hook) and server-side (INFO commandstats), that UNWATCH is elided after a committed EXEC, an aborted EXEC, an EXECABORT, and a no-key Watch, and is still sent on every path where no EXEC cleared the watch: an error before EXEC, a read-only decision, a non-transactional pipeline, an empty TxPipelined, multi-key bail, a WATCH re-armed after EXEC, a panic in fn, a failed WATCH, and a server error reported before EXEC runs. A manual Unwatch case asserts no second UNWATCH from Close, and a PoolSize=1 reuse case asserts no watch leaks onto a pooled connection. Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
2026-10-22 19:49:00 +03:00 · 2026-06-17 10:15:40 +01:00
parent bf57a51fdb
commit 5484b0b86c
2 changed files with 483 additions and 2 deletions
--- a/tx.go
+++ b/tx.go
@@ -2,6 +2,7 @@ package redis

 import (
 	"context"
+	"errors"

 	"github.com/redis/go-redis/v9/internal/pool"
 	"github.com/redis/go-redis/v9/internal/proto"
@@ -19,6 +20,24 @@ type Tx struct {
 	baseClient
 	cmdable
 	statefulCmdable
+
+	// watchArmed reports whether a WATCH issued through Tx.Watch may still be
+	// active on the connection. It is set on a successful WATCH and cleared once
+	// the watched keys are discarded server-side: on UNWATCH (Tx.Unwatch) and on
+	// EXEC, including an aborted EXEC (the TxPipeline closure). Close uses it to
+	// skip an otherwise redundant UNWATCH round trip.
+	//
+	// Only Tx.Watch, Tx.Unwatch and the TxPipeline EXEC closure maintain this
+	// flag. go-redis never issues a standalone DISCARD (Pipeline.Discard is a
+	// client-side buffer reset, not a server command). Issuing a WATCH directly
+	// via Process bypasses tracking, so Close would not release it and the watch
+	// would leak onto the pooled connection; that is the one case where skipping
+	// UNWATCH is unsafe, and using raw Process for WATCH/EXEC/UNWATCH is therefore
+	// unsupported.
+	//
+	// Tx is not safe for concurrent use (see above), so watchArmed is accessed
+	// only from the goroutine that owns the Tx and needs no synchronization.
+	watchArmed bool
 }

 func (c *Client) newTx() *Tx {
@@ -70,7 +89,13 @@ func (c *Client) Watch(ctx context.Context, fn func(*Tx) error, keys ...string)

 // Close closes the transaction, releasing any open resources.
 func (c *Tx) Close(ctx context.Context) error {
-	_ = c.Unwatch(ctx).Err()
+	// UNWATCH is only needed while a WATCH is still active. EXEC discards the
+	// watched keys server-side on both commit and abort, so the common
+	// WATCH/.../EXEC paths leave nothing to release and avoid the extra round
+	// trip.
+	if c.watchArmed {
+		_ = c.Unwatch(ctx).Err()
+	}
 	return c.baseClient.Close()
 }

@@ -84,6 +109,11 @@ func (c *Tx) Watch(ctx context.Context, keys ...string) *StatusCmd {
 	}
 	cmd := NewStatusCmd(ctx, args...)
 	_ = c.Process(ctx, cmd)
+	// A successful WATCH leaves keys watched on the connection that Close must
+	// later release with UNWATCH.
+	if cmd.Err() == nil {
+		c.watchArmed = true
+	}
 	return cmd
 }

@@ -96,6 +126,10 @@ func (c *Tx) Unwatch(ctx context.Context, keys ...string) *StatusCmd {
 	}
 	cmd := NewStatusCmd(ctx, args...)
 	_ = c.Process(ctx, cmd)
+	// The watched keys have been released, so Close need not UNWATCH again.
+	if cmd.Err() == nil {
+		c.watchArmed = false
+	}
 	return cmd
 }

@@ -133,7 +167,20 @@ func (c *Tx) TxPipeline() Pipeliner {
 	pipe := Pipeline{
 		exec: func(ctx context.Context, cmds []Cmder) error {
 			cmds = wrapMultiExec(ctx, cmds)
-			return c.processTxPipelineHook(ctx, cmds)
+			err := c.processTxPipelineHook(ctx, cmds)
+			// EXEC discards the watched keys server-side, so the watch is
+			// cleared only when EXEC actually ran: a nil error (committed),
+			// TxFailedErr (a watched key changed, EXEC returned nil), or an
+			// EXECABORT error (a queued command was rejected, so EXEC discarded
+			// the transaction). All three release the watched keys. Any other
+			// error may be reported before EXEC executes (for example -LOADING on
+			// the MULTI reply) or on a broken connection, so leave watchArmed set
+			// and let Close send UNWATCH rather than risk leaving a watch on a
+			// pooled connection.
+			if err == nil || errors.Is(err, TxFailedErr) || IsExecAbortError(err) {
+				c.watchArmed = false
+			}
+			return err
 		},
 	}
 	pipe.init()
--- a/tx_unwatch_test.go
+++ b/tx_unwatch_test.go
@@ -0,0 +1,434 @@
+package redis_test
+
+import (
+	"context"
+	"errors"
+	"strconv"
+	"strings"
+	"sync"
+
+	. "github.com/bsm/ginkgo/v2"
+	. "github.com/bsm/gomega"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// cmdRecorder captures the command stream a client sends so tests can assert
+// the presence or absence of individual commands. UNWATCH from Tx.Close flows
+// through ProcessHook; the MULTI/EXEC block flows through ProcessPipelineHook.
+type cmdRecorder struct {
+	mu   sync.Mutex
+	sent []string
+}
+
+func (h *cmdRecorder) record(names ...string) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	h.sent = append(h.sent, names...)
+}
+
+func (h *cmdRecorder) has(name string) bool {
+	return h.count(name) > 0
+}
+
+func (h *cmdRecorder) count(name string) int {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	n := 0
+	for _, c := range h.sent {
+		if c == name {
+			n++
+		}
+	}
+	return n
+}
+
+func (h *cmdRecorder) DialHook(next redis.DialHook) redis.DialHook { return next }
+
+func (h *cmdRecorder) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		h.record(strings.ToUpper(cmd.Name()))
+		return next(ctx, cmd)
+	}
+}
+
+func (h *cmdRecorder) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return func(ctx context.Context, cmds []redis.Cmder) error {
+		for _, cmd := range cmds {
+			h.record(strings.ToUpper(cmd.Name()))
+		}
+		return next(ctx, cmds)
+	}
+}
+
+// commandCalls returns the per-command call counts reported by
+// INFO commandstats, keyed by command name (e.g. "unwatch").
+func commandCalls(client *redis.Client) map[string]int64 {
+	info, err := client.Info(ctx, "commandstats").Result()
+	Expect(err).NotTo(HaveOccurred())
+
+	calls := make(map[string]int64)
+	for _, line := range strings.Split(info, "\n") {
+		line = strings.TrimSpace(line)
+		const prefix = "cmdstat_"
+		if !strings.HasPrefix(line, prefix) {
+			continue
+		}
+		name, rest, ok := strings.Cut(line[len(prefix):], ":")
+		if !ok {
+			continue
+		}
+		// rest looks like: calls=3,usec=10,usec_per_call=3.33,...
+		fields := strings.Split(rest, ",")
+		if len(fields) == 0 {
+			continue
+		}
+		kv := strings.SplitN(fields[0], "=", 2)
+		if len(kv) != 2 || kv[0] != "calls" {
+			continue
+		}
+		n, err := strconv.ParseInt(kv[1], 10, 64)
+		Expect(err).NotTo(HaveOccurred())
+		calls[name] = n
+	}
+	return calls
+}
+
+// pipelineErrInjector returns a fixed error for the MULTI/EXEC batch without
+// executing it, simulating a server error reported before EXEC runs (such as
+// -LOADING on the MULTI reply). The watched keys stay armed, so Close must
+// still send UNWATCH.
+type pipelineErrInjector struct{ err error }
+
+func (h *pipelineErrInjector) DialHook(next redis.DialHook) redis.DialHook { return next }
+
+func (h *pipelineErrInjector) ProcessHook(next redis.ProcessHook) redis.ProcessHook { return next }
+
+func (h *pipelineErrInjector) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return func(ctx context.Context, cmds []redis.Cmder) error {
+		for _, cmd := range cmds {
+			if strings.EqualFold(cmd.Name(), "exec") {
+				return h.err // do not call next: EXEC never runs
+			}
+		}
+		return next(ctx, cmds)
+	}
+}
+
+// cmdErrInjector fails a single named command without sending it, simulating a
+// server error on that command (e.g. a WATCH that the server rejects).
+type cmdErrInjector struct {
+	name string
+	err  error
+}
+
+func (h *cmdErrInjector) DialHook(next redis.DialHook) redis.DialHook { return next }
+
+func (h *cmdErrInjector) ProcessHook(next redis.ProcessHook) redis.ProcessHook {
+	return func(ctx context.Context, cmd redis.Cmder) error {
+		if strings.EqualFold(cmd.Name(), h.name) {
+			return h.err // do not call next: the command never runs
+		}
+		return next(ctx, cmd)
+	}
+}
+
+func (h *cmdErrInjector) ProcessPipelineHook(next redis.ProcessPipelineHook) redis.ProcessPipelineHook {
+	return next
+}
+
+var _ = Describe("Tx UNWATCH elision", func() {
+	var client *redis.Client
+
+	BeforeEach(func() {
+		client = redis.NewClient(redisOptions())
+		Expect(client.FlushDB(ctx).Err()).NotTo(HaveOccurred())
+	})
+
+	AfterEach(func() {
+		Expect(client.Close()).NotTo(HaveOccurred())
+	})
+
+	// watchRecorded runs fn inside client.Watch (recording the wire commands)
+	// and returns the recorder plus the error returned by Watch.
+	watchRecorded := func(fn func(*redis.Tx) error, keys ...string) (*cmdRecorder, error) {
+		rec := &cmdRecorder{}
+		client.AddHook(rec)
+		err := client.Watch(ctx, fn, keys...)
+		return rec, err
+	}
+
+	// expectUnwatchSent asserts that fn, which must not run EXEC, leaves the
+	// connection with a WATCH that Close releases via UNWATCH. The error
+	// returned by fn is irrelevant here (a read may legitimately return
+	// redis.Nil), so it is not asserted.
+	expectUnwatchSent := func(fn func(*redis.Tx) error, keys ...string) {
+		rec, _ := watchRecorded(fn, keys...)
+		Expect(rec.has("WATCH")).To(BeTrue(), "WATCH must be sent")
+		Expect(rec.has("EXEC")).To(BeFalse(), "no EXEC should have run")
+		Expect(rec.has("UNWATCH")).To(BeTrue(), "UNWATCH must be sent when no EXEC cleared the watch")
+	}
+
+	// --- the watch is cleared by EXEC: UNWATCH must be elided ---
+
+	It("does not send UNWATCH after a committed EXEC", func() {
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "value", 0)
+				return nil
+			})
+			return err
+		}, "key")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(rec.has("WATCH")).To(BeTrue())
+		Expect(rec.has("EXEC")).To(BeTrue())
+		Expect(rec.has("UNWATCH")).To(BeFalse())
+	})
+
+	It("does not send UNWATCH when no keys are watched", func() {
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			return tx.Ping(ctx).Err()
+		})
+		Expect(err).NotTo(HaveOccurred())
+		Expect(rec.has("PING")).To(BeTrue(), "fn must have actually run")
+		Expect(rec.has("WATCH")).To(BeFalse(), "no WATCH should be sent for a no-key Watch")
+		Expect(rec.has("UNWATCH")).To(BeFalse())
+	})
+
+	It("does not send a second UNWATCH when fn already called Unwatch", func() {
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			return tx.Unwatch(ctx).Err()
+		}, "key")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(rec.count("UNWATCH")).To(Equal(1), "Close must not re-issue UNWATCH after a manual Unwatch")
+	})
+
+	It("does not send UNWATCH when WATCH itself fails", func() {
+		rec := &cmdRecorder{}
+		client.AddHook(rec) // must be added before the injector so it records the short-circuited command
+		client.AddHook(&cmdErrInjector{name: "watch", err: errors.New("watch rejected")})
+		err := client.Watch(ctx, func(tx *redis.Tx) error {
+			return nil
+		}, "key")
+		Expect(err).To(HaveOccurred())
+		Expect(rec.has("WATCH")).To(BeTrue())
+		Expect(rec.has("UNWATCH")).To(BeFalse(), "a failed WATCH armed nothing; Close must not UNWATCH")
+	})
+
+	It("records no UNWATCH server-side after a committed EXEC (commandstats)", Label("NonRedisEnterprise"), func() {
+		Expect(client.Do(ctx, "config", "resetstat").Err()).NotTo(HaveOccurred())
+
+		err := client.Watch(ctx, func(tx *redis.Tx) error {
+			_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "value", 0)
+				return nil
+			})
+			return err
+		}, "key")
+		Expect(err).NotTo(HaveOccurred())
+
+		calls := commandCalls(client)
+		Expect(calls).To(HaveKey("watch"))
+		Expect(calls).To(HaveKey("exec"))
+		Expect(calls).NotTo(HaveKey("unwatch"))
+	})
+
+	// --- the watch is NOT cleared (no EXEC): UNWATCH must still be sent ---
+
+	It("sends UNWATCH when fn returns an error before EXEC", func() {
+		expectUnwatchSent(func(tx *redis.Tx) error {
+			return errors.New("bail")
+		}, "key")
+	})
+
+	It("sends UNWATCH when fn reads and decides not to write", func() {
+		expectUnwatchSent(func(tx *redis.Tx) error {
+			_, err := tx.Get(ctx, "key").Result()
+			if err != nil && err != redis.Nil {
+				return err
+			}
+			return nil // no TxPipelined -> no EXEC
+		}, "key")
+	})
+
+	It("sends UNWATCH when fn uses only a non-transactional pipeline", func() {
+		expectUnwatchSent(func(tx *redis.Tx) error {
+			_, err := tx.Pipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Get(ctx, "key")
+				return nil
+			})
+			return err // Pipelined, not TxPipelined -> no MULTI/EXEC
+		}, "key")
+	})
+
+	It("sends UNWATCH when TxPipelined queues no commands", func() {
+		expectUnwatchSent(func(tx *redis.Tx) error {
+			_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				return nil // nothing queued -> exec closure never runs -> no EXEC
+			})
+			return err
+		}, "key")
+	})
+
+	It("sends UNWATCH when bailing with multiple keys watched", func() {
+		expectUnwatchSent(func(tx *redis.Tx) error {
+			return errors.New("bail")
+		}, "{tag}.a", "{tag}.b", "{tag}.c")
+	})
+
+	It("sends UNWATCH when a WATCH is re-armed after a committed EXEC", func() {
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			if _, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "value", 0)
+				return nil
+			}); err != nil {
+				return err
+			}
+			// EXEC cleared the watch; re-arm it and bail without a second EXEC.
+			return tx.Watch(ctx, "key").Err()
+		}, "key")
+		Expect(err).NotTo(HaveOccurred())
+		Expect(rec.has("EXEC")).To(BeTrue())
+		Expect(rec.has("UNWATCH")).To(BeTrue(), "re-armed WATCH must be released by Close")
+	})
+
+	It("sends UNWATCH when fn panics", func() {
+		rec := &cmdRecorder{}
+		client.AddHook(rec)
+		Expect(func() {
+			_ = client.Watch(ctx, func(tx *redis.Tx) error {
+				panic("boom")
+			}, "key")
+		}).To(Panic())
+		Expect(rec.has("WATCH")).To(BeTrue())
+		Expect(rec.has("UNWATCH")).To(BeTrue(), "deferred Close must UNWATCH during panic unwinding")
+	})
+
+	It("sends UNWATCH when the EXEC pipeline fails before EXEC runs", func() {
+		rec := &cmdRecorder{}
+		client.AddHook(rec) // must be added before the injector so it records the short-circuited batch
+		// Inject a non-TxFailedErr server error for the MULTI/EXEC batch so EXEC
+		// never executes and the watch stays armed (mirrors -LOADING on MULTI).
+		// redis.Nil is a RedisError other than TxFailedErr.
+		client.AddHook(&pipelineErrInjector{err: redis.Nil})
+		err := client.Watch(ctx, func(tx *redis.Tx) error {
+			_, e := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "value", 0)
+				return nil
+			})
+			return e
+		}, "key")
+		Expect(err).To(HaveOccurred())
+		Expect(rec.has("WATCH")).To(BeTrue())
+		Expect(rec.has("UNWATCH")).To(BeTrue(), "a pre-EXEC error must not elide UNWATCH; the watch is still armed")
+	})
+
+	It("records UNWATCH server-side when fn returns before EXEC (commandstats)", Label("NonRedisEnterprise"), func() {
+		Expect(client.Do(ctx, "config", "resetstat").Err()).NotTo(HaveOccurred())
+
+		err := client.Watch(ctx, func(tx *redis.Tx) error {
+			return errors.New("bail")
+		}, "key")
+		Expect(err).To(MatchError("bail"))
+
+		calls := commandCalls(client)
+		Expect(calls).To(HaveKey("watch"))
+		Expect(calls).To(HaveKey("unwatch"))
+		Expect(calls).NotTo(HaveKey("exec"))
+	})
+
+	// --- optimistic abort: EXEC ran (returned TxFailedErr), watch is cleared ---
+
+	It("does not send UNWATCH after an aborted EXEC (optimistic lock)", func() {
+		other := redis.NewClient(redisOptions())
+		defer other.Close()
+		Expect(client.Set(ctx, "key", "0", 0).Err()).NotTo(HaveOccurred())
+
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			// Dirty the watched key from another connection before EXEC.
+			if e := other.Set(ctx, "key", "changed", 0).Err(); e != nil {
+				return e
+			}
+			_, e := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "new", 0)
+				return nil
+			})
+			return e
+		}, "key")
+		Expect(err).To(MatchError(redis.TxFailedErr))
+		Expect(rec.has("EXEC")).To(BeTrue(), "EXEC must have run")
+		Expect(rec.has("UNWATCH")).To(BeFalse(), "aborted EXEC already cleared the watch")
+	})
+
+	It("records no UNWATCH server-side after an aborted EXEC (commandstats)", Label("NonRedisEnterprise"), func() {
+		other := redis.NewClient(redisOptions())
+		defer other.Close()
+		Expect(client.Set(ctx, "key", "0", 0).Err()).NotTo(HaveOccurred())
+		Expect(client.Do(ctx, "config", "resetstat").Err()).NotTo(HaveOccurred())
+
+		err := client.Watch(ctx, func(tx *redis.Tx) error {
+			if e := other.Set(ctx, "key", "changed", 0).Err(); e != nil {
+				return e
+			}
+			_, e := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "key", "new", 0)
+				return nil
+			})
+			return e
+		}, "key")
+		Expect(err).To(MatchError(redis.TxFailedErr))
+
+		calls := commandCalls(client)
+		Expect(calls).To(HaveKey("watch"))
+		Expect(calls).To(HaveKey("exec")) // EXEC ran and aborted
+		Expect(calls).NotTo(HaveKey("unwatch"))
+	})
+
+	It("does not send UNWATCH after an EXECABORT", func() {
+		rec, err := watchRecorded(func(tx *redis.Tx) error {
+			_, e := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				// SET with a missing argument is rejected at queue time, so EXEC
+				// returns EXECABORT and discards the transaction (and the watch).
+				pipe.Do(ctx, "set", "key")
+				return nil
+			})
+			return e
+		}, "key")
+		Expect(err).To(HaveOccurred())
+		Expect(redis.IsExecAbortError(err)).To(BeTrue())
+		Expect(rec.has("EXEC")).To(BeTrue(), "EXEC must have run")
+		Expect(rec.has("UNWATCH")).To(BeFalse(), "EXECABORT already discarded the watch")
+	})
+
+	// --- pool safety: a bailed transaction must not leak a watch ---
+
+	It("does not leak a watch onto a pooled connection", func() {
+		opt := redisOptions()
+		opt.PoolSize = 1
+		opt.MinIdleConns = 0
+		c := redis.NewClient(opt)
+		defer c.Close()
+		Expect(c.Set(ctx, "guard", "0", 0).Err()).NotTo(HaveOccurred())
+
+		errBail := errors.New("bail")
+		err := c.Watch(ctx, func(tx *redis.Tx) error {
+			return errBail
+		}, "guard")
+		Expect(err).To(Equal(errBail))
+
+		// Mutate the watched key on the (single) reused connection.
+		Expect(c.Incr(ctx, "guard").Err()).NotTo(HaveOccurred())
+
+		// An unrelated transaction on the reused connection must not be aborted
+		// by a leftover WATCH on "guard"; a leak would surface as TxFailedErr.
+		err = c.Watch(ctx, func(tx *redis.Tx) error {
+			_, err := tx.TxPipelined(ctx, func(pipe redis.Pipeliner) error {
+				pipe.Set(ctx, "fresh", "ok", 0)
+				return nil
+			})
+			return err
+		})
+		Expect(err).NotTo(HaveOccurred(), "a leaked WATCH would abort this transaction with TxFailedErr")
+		Expect(c.Get(ctx, "fresh").Val()).To(Equal("ok"))
+	})
+})