fix: do not close connection on ReceiveWithTimeout deadline

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
@@ -558,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' {
@@ -626,60 +636,73 @@ 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 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, err
+		return nil, len(line) == 0, 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 {
@@ -710,7 +733,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 +757,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 +767,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)
@@ -752,7 +783,42 @@ func (c *conn) ReceiveWithTimeout(timeout time.Duration) (reply interface{}, err
 		return nil, c.fatal(err)
 	}
 
-	if reply, err = c.readReply(); err != nil {
+	var atBoundary bool
+	if reply, atBoundary, err = c.readReplyBoundary(); err != nil {
+		if !closeOnTimeout && atBoundary && isTimeoutError(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, 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

redis/conn_test.go

@@ -23,10 +23,10 @@ import (
 	"io"
 	"math"
 	"net"
-	"sync"
 	"os"
 	"reflect"
 	"strings"
+	"sync"
 	"testing"
 	"time"
 
@@ -1125,3 +1125,122 @@ 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
+}
+
+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
+	// 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")
+	}
+}