fix: run MONITOR on a dedicated connection and propagate read errors

command.go

@@ -8039,6 +8039,9 @@ type MonitorCmd struct {
 	ch     chan string
 	status MonitorStatus
 	mu     sync.Mutex
+	// closeConn closes the dedicated connection the MONITOR command runs on.
+	// It is set before readReply spawns the reader goroutine.
+	closeConn func()
 }
 
 func newMonitorCmd(ctx context.Context, ch chan string) *MonitorCmd {
@@ -8061,6 +8064,11 @@ func (cmd *MonitorCmd) String() string {
 func (cmd *MonitorCmd) readReply(rd *proto.Reader) error {
 	ctx, cancel := context.WithCancel(cmd.ctx)
 	go func(ctx context.Context) {
+		defer func() {
+			if cmd.closeConn != nil {
+				cmd.closeConn()
+			}
+		}()
 		for {
 			select {
 			case <-ctx.Done():
@@ -8069,6 +8077,10 @@ func (cmd *MonitorCmd) readReply(rd *proto.Reader) error {
 				err := cmd.readMonitor(rd, cancel)
 				if err != nil {
 					cmd.err = err
+					cancel()
+					// Close the channel so a listener blocked on it is unblocked
+					// and can pick up the error with cmd.Err().
+					close(cmd.ch)
 					return
 				}
 			}
@@ -8079,25 +8091,46 @@ func (cmd *MonitorCmd) readReply(rd *proto.Reader) error {
 
 func (cmd *MonitorCmd) readMonitor(rd *proto.Reader, cancel context.CancelFunc) error {
 	for {
-		cmd.mu.Lock()
-		st := cmd.status
-		pk, _ := rd.Peek(1)
-		cmd.mu.Unlock()
-		if len(pk) != 0 && st == monitorStatusStart {
-			cmd.mu.Lock()
+		if cmd.getStatus() == monitorStatusStop {
+			cancel()
+			return nil
+		}
+		// The reader goroutine is the only reader of this connection, so
+		// Peek can block without holding the mutex; Stop unblocks it by
+		// closing the connection.
+		pk, err := rd.Peek(1)
+		if err != nil {
+			if cmd.getStatus() == monitorStatusStop {
+				// Stop closed the connection to unblock Peek: shut down
+				// cleanly instead of reporting the read error.
+				cancel()
+				return nil
+			}
+			// The connection is no longer usable; without this the loop
+			// would spin forever re-peeking a dead connection.
+			return err
+		}
+		if len(pk) != 0 && cmd.getStatus() == monitorStatusStart {
 			line, err := rd.ReadString()
-			cmd.mu.Unlock()
 			if err != nil {
+				if cmd.getStatus() == monitorStatusStop {
+					// Stop closed the connection while a line was being
+					// read: shut down cleanly instead of reporting the
+					// read error.
+					cancel()
+					return nil
+				}
 				return err
 			}
 			cmd.ch <- line
 		}
-		if st == monitorStatusStop {
-			cancel()
-			break
-		}
 	}
-	return nil
+}
+
+func (cmd *MonitorCmd) getStatus() MonitorStatus {
+	cmd.mu.Lock()
+	defer cmd.mu.Unlock()
+	return cmd.status
 }
 
 func (cmd *MonitorCmd) Start() {
@@ -8110,6 +8143,11 @@ func (cmd *MonitorCmd) Stop() {
 	cmd.mu.Lock()
 	defer cmd.mu.Unlock()
 	cmd.status = monitorStatusStop
+	// The reader goroutine may be blocked in Peek waiting for traffic;
+	// closing the connection unblocks it so it can observe the stop.
+	if cmd.closeConn != nil {
+		cmd.closeConn()
+	}
 }
 
 type VectorScoreSliceCmd struct {

commands.go

@@ -814,9 +814,13 @@ and process all commands sent to a Redis server. This mimics the behavior of
 MONITOR in the redis-cli.
 
 Notes:
-- Using MONITOR blocks the connection to the server for itself. It needs a dedicated connection
-- The user should create a channel of type string
-- This runs concurrently in the background. Trigger via the Start and Stop functions
+  - Using MONITOR blocks the connection to the server for itself. It needs a dedicated connection
+  - The user should create a channel of type string
+  - This runs concurrently in the background. Trigger via the Start and Stop functions
+  - MONITOR runs on a dedicated connection with no read deadline, so an idle server
+    does not break it. If the connection fails, the channel is closed and the error is
+    available via Err(); a clean Stop leaves the channel open and reports no error
+
 See further: Redis MONITOR command: https://redis.io/commands/monitor
 */
 func (c cmdable) Monitor(ctx context.Context, ch chan string) *MonitorCmd {

internal/pool/pool.go

@@ -28,6 +28,10 @@ const (
 	// authentication error during re-authentication.
 	CloseReasonAuthError = "auth_error"
 
+	// CloseReasonMonitor indicates a dedicated MONITOR connection was closed,
+	// either by MonitorCmd.Stop or because the connection failed.
+	CloseReasonMonitor = "monitor_closed"
+
 	// CloseReasonTest is used in tests when closing connections.
 	CloseReasonTest = "test"
 

monitor_conn_test.go

@@ -0,0 +1,153 @@
+package redis_test
+
+import (
+	"bufio"
+	"context"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+)
+
+// fakeMonitorServer implements just enough of the Redis protocol for the
+// MONITOR command: it acknowledges the handshake, replies +OK to MONITOR,
+// streams one monitor line and then closes the connection, simulating the
+// server (or the network) dropping a monitor connection.
+func fakeMonitorServer(ln net.Listener) {
+	for {
+		conn, err := ln.Accept()
+		if err != nil {
+			return
+		}
+		go func(c net.Conn) {
+			defer c.Close()
+			rd := bufio.NewReader(c)
+			for {
+				line, err := rd.ReadString('\n')
+				if err != nil {
+					return
+				}
+				if line[0] == '*' || line[0] == '$' {
+					continue // RESP framing, only react to command verbs
+				}
+				switch {
+				case strings.HasPrefix(strings.ToLower(line), "hello"):
+					c.Write([]byte("-ERR unknown command 'hello'\r\n"))
+				case strings.HasPrefix(strings.ToLower(line), "monitor"):
+					c.Write([]byte("+OK\r\n"))
+					c.Write([]byte("+1700000000.000000 [0 127.0.0.1:1] \"set\" \"foo\" \"bar\"\r\n"))
+					return // deferred Close drops the connection mid-monitor
+				default:
+					c.Write([]byte("+OK\r\n"))
+				}
+			}
+		}(conn)
+	}
+}
+
+// See https://github.com/redis/go-redis/issues/3079: when the connection
+// backing MONITOR dies, the monitor channel must be closed so a listener is
+// not blocked forever, and the error must be available via cmd.Err().
+func TestMonitorConnErrorClosesChannel(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	go fakeMonitorServer(ln)
+
+	client := redis.NewClient(&redis.Options{
+		Addr: ln.Addr().String(),
+	})
+	defer client.Close()
+
+	ch := make(chan string, 100)
+	cmd := client.Monitor(context.Background(), ch)
+	if cmd.Err() != nil {
+		t.Fatal(cmd.Err())
+	}
+	cmd.Start()
+	defer cmd.Stop()
+
+	deadline := time.After(10 * time.Second)
+	for {
+		select {
+		case _, ok := <-ch:
+			if !ok {
+				if cmd.Err() == nil {
+					t.Fatal("channel closed but cmd.Err() is nil")
+				}
+				return // unblocked with an error, as expected
+			}
+		case <-deadline:
+			t.Fatal("monitor channel was not closed after the connection died; listener would block forever")
+		}
+	}
+}
+
+// A clean Stop must not report an error and must not require any server
+// traffic to take effect: closing the dedicated connection unblocks the
+// reader goroutine even when the server is idle.
+func TestMonitorStopWithoutTraffic(t *testing.T) {
+	ln, err := net.Listen("tcp", "127.0.0.1:0")
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer ln.Close()
+	go func() {
+		for {
+			conn, err := ln.Accept()
+			if err != nil {
+				return
+			}
+			go func(c net.Conn) {
+				rd := bufio.NewReader(c)
+				for {
+					line, err := rd.ReadString('\n')
+					if err != nil {
+						c.Close()
+						return
+					}
+					if line[0] == '*' || line[0] == '$' {
+						continue
+					}
+					switch {
+					case strings.HasPrefix(strings.ToLower(line), "hello"):
+						c.Write([]byte("-ERR unknown command 'hello'\r\n"))
+					case strings.HasPrefix(strings.ToLower(line), "monitor"):
+						c.Write([]byte("+OK\r\n"))
+						select {} // keep the connection open, send nothing
+					default:
+						c.Write([]byte("+OK\r\n"))
+					}
+				}
+			}(conn)
+		}
+	}()
+
+	client := redis.NewClient(&redis.Options{
+		Addr: ln.Addr().String(),
+	})
+	defer client.Close()
+
+	ch := make(chan string, 100)
+	cmd := client.Monitor(context.Background(), ch)
+	cmd.Start()
+	time.Sleep(100 * time.Millisecond) // let the reader block in Peek
+
+	done := make(chan struct{})
+	go func() {
+		cmd.Stop()
+		close(done)
+	}()
+	select {
+	case <-done:
+	case <-time.After(5 * time.Second):
+		t.Fatal("Stop() blocked; reader goroutine was not unblocked")
+	}
+	if cmd.Err() != nil {
+		t.Fatalf("clean Stop must not set an error, got: %v", cmd.Err())
+	}
+}

redis.go

@@ -991,6 +991,13 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
 		}
 	}
 
+	// MONITOR takes over its connection indefinitely, so it must not run on
+	// a pooled connection: the pool (or another command) would read from the
+	// same connection the monitor goroutine is reading from.
+	if monitorCmd, ok := cmd.(*MonitorCmd); ok {
+		return false, nil, c.processMonitor(ctx, monitorCmd)
+	}
+
 	var usedConn *pool.Conn
 	var retryTimeout atomic.Uint32
 	if err := c.withConn(ctx, func(ctx context.Context, cn *pool.Conn) error {
@@ -1041,6 +1048,47 @@ func (c *baseClient) _process(ctx context.Context, cmd Cmder, attempt int) (bool
 	return false, usedConn, nil
 }
 
+// processMonitor runs a MONITOR command on a dedicated, non-pooled
+// connection. The monitor's reader goroutine owns the connection until
+// monitoring stops or the connection fails; no read deadline is set, since
+// MONITOR receives no traffic while the server is idle.
+func (c *baseClient) processMonitor(ctx context.Context, cmd *MonitorCmd) error {
+	cn, err := c.connPool.NewConn(ctx)
+	if err != nil {
+		cmd.SetErr(err)
+		return err
+	}
+	// CloseConn (rather than cn.Close) removes the connection from the
+	// pool's bookkeeping and records metrics. Background context: the
+	// connection may outlive ctx and be closed much later by Stop or the
+	// reader goroutine.
+	closeConn := func() {
+		_ = c.connPool.CloseConn(context.Background(), cn, pool.CloseReasonMonitor, pool.MetricStateUsed)
+	}
+	if err := c.initConn(ctx, cn); err != nil {
+		closeConn()
+		cmd.SetErr(err)
+		return err
+	}
+	if err := cn.WithWriter(c.context(ctx), c.opt.WriteTimeout, func(wr *proto.Writer) error {
+		return writeCmd(wr, cmd)
+	}); err != nil {
+		closeConn()
+		cmd.SetErr(err)
+		return err
+	}
+	cmd.closeConn = closeConn
+	// Timeout 0 clears any read deadline armed during the handshake:
+	// MONITOR receives no traffic while the server is idle, so the
+	// connection must not have a read deadline at all.
+	if err := cn.WithReader(c.context(ctx), 0, cmd.readReply); err != nil {
+		closeConn()
+		cmd.SetErr(err)
+		return err
+	}
+	return nil
+}
+
 func (c *baseClient) retryBackoff(attempt int) time.Duration {
 	return internal.RetryBackoff(attempt, c.opt.MinRetryBackoff, c.opt.MaxRetryBackoff)
 }