From 938639d76987cda7d05f0006660d8612a2901277 Mon Sep 17 00:00:00 2001
From: Solaris-star <820622658@qq.com>
Date: Wed, 22 Jul 2026 06:32:10 +0800
Subject: [PATCH 1/4] fix: do not close connection on ReceiveWithTimeout
 deadline

---
 redis/conn.go      | 25 +++++++++++++++++++++++--
 redis/conn_test.go | 43 +++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 66 insertions(+), 2 deletions(-)

diff --git a/redis/conn.go b/redis/conn.go
index dc445a06..3d7e1130 100644
--- a/redis/conn.go
+++ b/redis/conn.go
@@ -428,6 +428,13 @@ func (c *conn) fatal(err error) error {
 	return err
 }
 
+// isTimeoutError reports whether err is a timeout from a read/write deadline.
+// Timeouts from ReceiveWithTimeout / DoWithTimeout must not close the connection.
+func isTimeoutError(err error) bool {
+	var ne net.Error
+	return errors.As(err, &ne) && ne.Timeout()
+}
+
 func (c *conn) Err() error {
 	c.mu.Lock()
 	err := c.err
@@ -710,7 +717,8 @@ func (c *conn) Flush() error {
 }
 
 func (c *conn) Receive() (interface{}, error) {
-	return c.ReceiveWithTimeout(c.readTimeout)
+	// Connection-level read timeouts are fatal: the connection may be stuck.
+	return c.receiveWithTimeout(c.readTimeout, true)
 }
 
 func (c *conn) ReceiveContext(ctx context.Context) (interface{}, error) {
@@ -733,7 +741,7 @@ func (c *conn) ReceiveContext(ctx context.Context) (interface{}, error) {
 	go func() {
 		defer close(endch)
 
-		r, e = c.ReceiveWithTimeout(realTimeout)
+		r, e = c.receiveWithTimeout(realTimeout, true)
 	}()
 	select {
 	case <-ctx.Done():
@@ -743,7 +751,14 @@ func (c *conn) ReceiveContext(ctx context.Context) (interface{}, error) {
 	}
 }
 
+// ReceiveWithTimeout receives a single reply with an explicit deadline.
+// Unlike the connection's default DialReadTimeout path, a deadline timeout
+// here does not close the connection so callers can poll (e.g. pub/sub).
 func (c *conn) ReceiveWithTimeout(timeout time.Duration) (reply interface{}, err error) {
+	return c.receiveWithTimeout(timeout, false)
+}
+
+func (c *conn) receiveWithTimeout(timeout time.Duration, closeOnTimeout bool) (reply interface{}, err error) {
 	var deadline time.Time
 	if timeout != 0 {
 		deadline = time.Now().Add(timeout)
@@ -753,6 +768,12 @@ func (c *conn) ReceiveWithTimeout(timeout time.Duration) (reply interface{}, err
 	}
 
 	if reply, err = c.readReply(); err != nil {
+		if !closeOnTimeout && isTimeoutError(err) {
+			// Clear the deadline so a subsequent receive is not immediately
+			// timed out by a stale past deadline. See #676.
+			_ = c.conn.SetReadDeadline(time.Time{})
+			return nil, err
+		}
 		return nil, c.fatal(err)
 	}
 	// When using pub/sub, the number of receives can be greater than the
diff --git a/redis/conn_test.go b/redis/conn_test.go
index 414b7ef3..535f9166 100644
--- a/redis/conn_test.go
+++ b/redis/conn_test.go
@@ -1125,3 +1125,46 @@ func TestWithTimeout(t *testing.T) {
 		}
 	}
 }
+
+
+func TestReceiveWithTimeoutDoesNotCloseConn(t *testing.T) {
+	// Regression for #676: a short ReceiveWithTimeout that hits the deadline
+	// must not permanently close the connection (pub/sub poll loops rely on this).
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	c := redis.NewConn(client, 0, 0)
+	defer c.Close()
+
+	// No data available; short timeout should surface as a timeout error only.
+	_, err := redis.ReceiveWithTimeout(c, 20*time.Millisecond)
+	if err == nil {
+		t.Fatal("expected timeout error, got nil")
+	}
+	if ne, ok := err.(net.Error); !ok || !ne.Timeout() {
+		t.Fatalf("expected net timeout error, got %T %v", err, err)
+	}
+	if c.Err() != nil {
+		t.Fatalf("connection should remain open after timeout, Conn.Err()=%v", c.Err())
+	}
+
+	// Server can still answer a subsequent command.
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		// Consume the PING request then reply OK.
+		buf := make([]byte, 64)
+		_, _ = server.Read(buf)
+		_, _ = server.Write([]byte("+OK\r\n"))
+	}()
+
+	reply, err := c.Do("PING")
+	if err != nil {
+		t.Fatalf("Do after timeout: %v", err)
+	}
+	if s, _ := redis.String(reply, nil); s != "OK" {
+		t.Fatalf("reply = %v, want OK", reply)
+	}
+	<-done
+}

From d8c4cae969b5ff75faf4dc154490152ed375d870 Mon Sep 17 00:00:00 2001
From: Solaris-star <820622658@qq.com>
Date: Thu, 23 Jul 2026 17:04:18 +0800
Subject: [PATCH 2/4] fix: only preserve connection on clean-boundary receive
 timeout

A ReceiveWithTimeout deadline that fires before any bytes of a reply
are consumed leaves the wire at a reply boundary, so the connection can
be safely reused (idle pub/sub polling, #676). Once the first line has
been read, a mid-reply timeout (partial bulk body or array element)
desynchronises the stream; such cases remain fatal and close the
connection to avoid returning stale bytes to a later Receive.

Addresses the wire-state concern raised in review.
---
 redis/conn.go | 68 +++++++++++++++++++++++++++++++++++----------------
 1 file changed, 47 insertions(+), 21 deletions(-)

diff --git a/redis/conn.go b/redis/conn.go
index 3d7e1130..8d4207de 100644
--- a/redis/conn.go
+++ b/redis/conn.go
@@ -633,60 +633,77 @@ var (
 )
 
 func (c *conn) readReply() (interface{}, error) {
+	reply, _, err := c.readReplyBoundary()
+	return reply, err
+}
+
+// readReplyBoundary reads a single reply and additionally reports whether an
+// error (if any) occurred at a clean reply boundary.
+//
+// atBoundary is true only when the error happened before the first line of the
+// reply was fully read. In that case the underlying bufio.Reader still buffers
+// any partially received bytes and the wire is positioned at the start of a
+// reply, so the connection may be safely reused (e.g. an idle pub/sub poll that
+// hits its deadline). Once the first line has been consumed, any later error
+// (partial bulk body, array element) leaves an in-flight reply on the wire;
+// atBoundary is then false and callers must treat the connection as fatal to
+// avoid desynchronising subsequent reads. See #676 and review of #698.
+func (c *conn) readReplyBoundary() (reply interface{}, atBoundary bool, err error) {
 	line, err := c.readLine()
 	if err != nil {
-		return nil, err
+		return nil, true, err
 	}
 	if len(line) == 0 {
-		return nil, protocolError("short response line")
+		return nil, false, protocolError("short response line")
 	}
 	switch line[0] {
 	case '+':
 		switch string(line[1:]) {
 		case "OK":
 			// Avoid allocation for frequent "+OK" response.
-			return okReply, nil
+			return okReply, false, nil
 		case "PONG":
 			// Avoid allocation in PING command benchmarks :)
-			return pongReply, nil
+			return pongReply, false, nil
 		default:
-			return string(line[1:]), nil
+			return string(line[1:]), false, nil
 		}
 	case '-':
-		return Error(line[1:]), nil
+		return Error(line[1:]), false, nil
 	case ':':
-		return parseInt(line[1:])
+		n, err := parseInt(line[1:])
+		return n, false, err
 	case '$':
 		n, err := parseLen(line[1:])
 		if n < 0 || err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		p := make([]byte, n)
 		_, err = io.ReadFull(c.br, p)
 		if err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		if line, err := c.readLine(); err != nil {
-			return nil, err
+			return nil, false, err
 		} else if len(line) != 0 {
-			return nil, protocolError("bad bulk string format")
+			return nil, false, protocolError("bad bulk string format")
 		}
-		return p, nil
+		return p, false, nil
 	case '*':
 		n, err := parseLen(line[1:])
 		if n < 0 || err != nil {
-			return nil, err
+			return nil, false, err
 		}
 		r := make([]interface{}, n)
 		for i := range r {
-			r[i], err = c.readReply()
+			r[i], _, err = c.readReplyBoundary()
 			if err != nil {
-				return nil, err
+				return nil, false, err
 			}
 		}
-		return r, nil
+		return r, false, nil
 	}
-	return nil, protocolError("unexpected response line")
+	return nil, false, protocolError("unexpected response line")
 }
 
 func (c *conn) Send(cmd string, args ...interface{}) error {
@@ -767,13 +784,22 @@ func (c *conn) receiveWithTimeout(timeout time.Duration, closeOnTimeout bool) (r
 		return nil, c.fatal(err)
 	}
 
-	if reply, err = c.readReply(); err != nil {
-		if !closeOnTimeout && isTimeoutError(err) {
-			// Clear the deadline so a subsequent receive is not immediately
-			// timed out by a stale past deadline. See #676.
+	var atBoundary bool
+	if reply, atBoundary, err = c.readReplyBoundary(); err != nil {
+		if !closeOnTimeout && atBoundary && isTimeoutError(err) {
+			// The timeout happened before any bytes of a reply were consumed,
+			// so the wire is still positioned at a reply boundary. Clear the
+			// (now stale) deadline and surface the timeout without closing:
+			// callers such as pub/sub poll loops can safely receive again.
+			// See #676.
 			_ = c.conn.SetReadDeadline(time.Time{})
 			return nil, err
 		}
+		// Either a fatal read error, or a timeout that occurred mid-reply
+		// (partial bulk body / array element). A partially consumed reply
+		// leaves the connection desynchronised, so it must be closed to avoid
+		// returning stale bytes to a later Receive. Addresses the wire-state
+		// concern raised in review of #698.
 		return nil, c.fatal(err)
 	}
 	// When using pub/sub, the number of receives can be greater than the

From ab5a05b9460223271c05eee6499acfa5e0edd33d Mon Sep 17 00:00:00 2001
From: Solaris-star <820622658@qq.com>
Date: Thu, 23 Jul 2026 17:41:23 +0800
Subject: [PATCH 3/4] fix: also close connection on boundary timeout when a
 request is in flight

A clean reply boundary at timeout is only safe to reuse when nothing is
outstanding. If a command was sent but its reply has not yet arrived, the
late reply could be misread as the answer to a later receive. Guard the
connection-preserving path with pending==0 so only truly idle waits (e.g.
pub/sub) keep the connection. Addresses @stevenh's wire-state review.
---
 redis/conn.go      | 44 ++++++++++++++++++++++++++++++------------
 redis/conn_test.go | 48 ++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 80 insertions(+), 12 deletions(-)

diff --git a/redis/conn.go b/redis/conn.go
index 8d4207de..4dab3d55 100644
--- a/redis/conn.go
+++ b/redis/conn.go
@@ -787,19 +787,39 @@ func (c *conn) receiveWithTimeout(timeout time.Duration, closeOnTimeout bool) (r
 	var atBoundary bool
 	if reply, atBoundary, err = c.readReplyBoundary(); err != nil {
 		if !closeOnTimeout && atBoundary && isTimeoutError(err) {
-			// The timeout happened before any bytes of a reply were consumed,
-			// so the wire is still positioned at a reply boundary. Clear the
-			// (now stale) deadline and surface the timeout without closing:
-			// callers such as pub/sub poll loops can safely receive again.
-			// See #676.
-			_ = c.conn.SetReadDeadline(time.Time{})
-			return nil, err
+			// The connection may only be safely reused after a timeout when
+			// BOTH conditions hold:
+			//
+			//  1. atBoundary: the timeout happened before any bytes of a reply
+			//     were consumed, so the wire is positioned at a reply boundary
+			//     and no half-read reply is left buffered.
+			//
+			//  2. c.pending == 0: there is no request whose response is still
+			//     outstanding. If a command has been sent but not yet answered,
+			//     its reply may still arrive later; keeping the connection would
+			//     let that in-flight response be misread as the reply to a
+			//     subsequent receive, corrupting response processing. This is
+			//     the wire-state concern raised by @stevenh in review: a clean
+			//     boundary *now* is not enough if a response could arrive in the
+			//     future. The safe-to-reuse case is an idle poll with nothing in
+			//     flight (e.g. pub/sub waiting for a server push), which is
+			//     exactly what #676 needs.
+			c.mu.Lock()
+			inflight := c.pending
+			c.mu.Unlock()
+			if inflight == 0 {
+				// Clear the (now stale) deadline and surface the timeout
+				// without closing so the caller can receive again. See #676.
+				_ = c.conn.SetReadDeadline(time.Time{})
+				return nil, err
+			}
 		}
-		// Either a fatal read error, or a timeout that occurred mid-reply
-		// (partial bulk body / array element). A partially consumed reply
-		// leaves the connection desynchronised, so it must be closed to avoid
-		// returning stale bytes to a later Receive. Addresses the wire-state
-		// concern raised in review of #698.
+		// Either a fatal read error, a timeout that occurred mid-reply (partial
+		// bulk body / array element), or a boundary timeout while a request is
+		// still in flight. In every one of these cases the connection can no
+		// longer be trusted to stay in sync, so it must be closed to avoid
+		// returning stale or misattributed bytes to a later Receive. Addresses
+		// the wire-state concern raised in review of #698.
 		return nil, c.fatal(err)
 	}
 	// When using pub/sub, the number of receives can be greater than the
diff --git a/redis/conn_test.go b/redis/conn_test.go
index 535f9166..1d347bec 100644
--- a/redis/conn_test.go
+++ b/redis/conn_test.go
@@ -1168,3 +1168,51 @@ func TestReceiveWithTimeoutDoesNotCloseConn(t *testing.T) {
 	}
 	<-done
 }
+
+func TestReceiveWithTimeoutClosesConnWhenRequestInFlight(t *testing.T) {
+	// Wire-state guard raised by @stevenh in review of #698: a clean reply
+	// boundary at timeout is only safe to reuse when nothing is in flight. If a
+	// command has been sent but its reply has not yet arrived, that reply may
+	// still land later; reusing the connection would let it be misread as the
+	// answer to a subsequent receive. In that case the connection MUST be closed
+	// even though the timeout occurred at a boundary.
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	c := redis.NewConn(client, 0, 0)
+	defer c.Close()
+
+	// Drain the outgoing command on the server side but deliberately never
+	// reply, so from the client's perspective a request is outstanding
+	// (pending > 0) with its response still "in flight".
+	drained := make(chan struct{})
+	go func() {
+		defer close(drained)
+		buf := make([]byte, 256)
+		_, _ = server.Read(buf)
+	}()
+
+	if err := c.Send("PING"); err != nil {
+		t.Fatalf("Send: %v", err)
+	}
+	if err := c.Flush(); err != nil {
+		t.Fatalf("Flush: %v", err)
+	}
+	<-drained
+
+	// The reply never comes, so the receive times out at a reply boundary while
+	// the PING response is still outstanding.
+	_, err := redis.ReceiveWithTimeout(c, 20*time.Millisecond)
+	if err == nil {
+		t.Fatal("expected timeout error, got nil")
+	}
+	if ne, ok := err.(net.Error); !ok || !ne.Timeout() {
+		t.Fatalf("expected net timeout error, got %T %v", err, err)
+	}
+	// Because a request was in flight, the connection must be closed to avoid
+	// the late reply corrupting a future receive.
+	if c.Err() == nil {
+		t.Fatal("connection must be closed after boundary timeout with a request in flight")
+	}
+}

From 09cf40ceed041e9e8ce5df3d9c3777858c43e74c Mon Sep 17 00:00:00 2001
From: Solaris-star <820622658@qq.com>
Date: Mon, 27 Jul 2026 08:58:46 +0800
Subject: [PATCH 4/4] fix: treat partial reply-line timeouts as fatal

ReadSlice returns bytes consumed before a timeout together with the error. Preserve that partial slice so ReceiveWithTimeout only considers a timeout to be at a clean RESP boundary when zero bytes were consumed. Add a regression that sends a partial first line and verifies the connection is closed.

Signed-off-by: Solaris-star <820622658@qq.com>
---
 redis/conn.go      | 19 +++++++++----------
 redis/conn_test.go | 32 ++++++++++++++++++++++++++++++--
 2 files changed, 39 insertions(+), 12 deletions(-)

diff --git a/redis/conn.go b/redis/conn.go
index 4dab3d55..3f8ff2ac 100644
--- a/redis/conn.go
+++ b/redis/conn.go
@@ -565,7 +565,10 @@ func (c *conn) readLine() ([]byte, error) {
 		p = buf
 	}
 	if err != nil {
-		return nil, err
+		// ReadSlice returns any bytes consumed before the error. Preserve them
+		// so callers can distinguish an idle timeout from a timeout after a
+		// partial response line.
+		return p, err
 	}
 	i := len(p) - 2
 	if i < 0 || p[i] != '\r' {
@@ -640,18 +643,14 @@ func (c *conn) readReply() (interface{}, error) {
 // readReplyBoundary reads a single reply and additionally reports whether an
 // error (if any) occurred at a clean reply boundary.
 //
-// atBoundary is true only when the error happened before the first line of the
-// reply was fully read. In that case the underlying bufio.Reader still buffers
-// any partially received bytes and the wire is positioned at the start of a
-// reply, so the connection may be safely reused (e.g. an idle pub/sub poll that
-// hits its deadline). Once the first line has been consumed, any later error
-// (partial bulk body, array element) leaves an in-flight reply on the wire;
-// atBoundary is then false and callers must treat the connection as fatal to
-// avoid desynchronising subsequent reads. See #676 and review of #698.
+// atBoundary is true only when the error happened before any bytes of the reply
+// were consumed. Once any part of the first line has been read, or a later read
+// fails (bulk body / array element), callers must treat the connection as fatal
+// to avoid desynchronising subsequent reads. See #676 and review of #698.
 func (c *conn) readReplyBoundary() (reply interface{}, atBoundary bool, err error) {
 	line, err := c.readLine()
 	if err != nil {
-		return nil, true, err
+		return nil, len(line) == 0, err
 	}
 	if len(line) == 0 {
 		return nil, false, protocolError("short response line")
diff --git a/redis/conn_test.go b/redis/conn_test.go
index 1d347bec..80db2fa2 100644
--- a/redis/conn_test.go
+++ b/redis/conn_test.go
@@ -23,10 +23,10 @@ import (
 	"io"
 	"math"
 	"net"
-	"sync"
 	"os"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1126,7 +1126,6 @@ func TestWithTimeout(t *testing.T) {
 	}
 }
 
-
 func TestReceiveWithTimeoutDoesNotCloseConn(t *testing.T) {
 	// Regression for #676: a short ReceiveWithTimeout that hits the deadline
 	// must not permanently close the connection (pub/sub poll loops rely on this).
@@ -1169,6 +1168,35 @@ func TestReceiveWithTimeoutDoesNotCloseConn(t *testing.T) {
 	<-done
 }
 
+func TestReceiveWithTimeoutClosesConnAfterPartialReplyLine(t *testing.T) {
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	c := redis.NewConn(client, 0, 0)
+	defer c.Close()
+
+	written := make(chan struct{})
+	go func() {
+		defer close(written)
+		// A timeout after any bytes of a response line have been consumed does
+		// not leave the connection at a known RESP boundary.
+		_, _ = server.Write([]byte("+PARTIAL"))
+	}()
+
+	_, err := redis.ReceiveWithTimeout(c, 20*time.Millisecond)
+	<-written
+	if err == nil {
+		t.Fatal("expected timeout error, got nil")
+	}
+	if ne, ok := err.(net.Error); !ok || !ne.Timeout() {
+		t.Fatalf("expected net timeout error, got %T %v", err, err)
+	}
+	if c.Err() == nil {
+		t.Fatal("connection must be closed after timeout with a partial reply line")
+	}
+}
+
 func TestReceiveWithTimeoutClosesConnWhenRequestInFlight(t *testing.T) {
 	// Wire-state guard raised by @stevenh in review of #698: a clean reply
 	// boundary at timeout is only safe to reuse when nothing is in flight. If a