fix(command): return queued tx error

.github/workflows/govulncheck.yml

@@ -29,7 +29,7 @@ jobs:
       - name: Set up Go
         uses: actions/setup-go@v7
         with:
-          go-version: "1.26.x"
+          go-version: "1.26.6"
           cache: true
 
       - name: Install govulncheck
@@ -66,4 +66,4 @@ jobs:
             exit 1
           fi
 
-          exit "$status"
+          exit "$status"

csc_test.go

@@ -2367,6 +2367,101 @@ func TestReleaseConnRemovesConnectionAfterPartialPushRead(t *testing.T) {
 	}
 }
 
+func TestReleaseConnRemovesConnectionAfterReaderBufferedPartialPushRead(t *testing.T) {
+	cp := &releaseRecordingPool{}
+	c := &baseClient{
+		opt:           &Options{Addr: "127.0.0.1:6379", Protocol: 3},
+		connPool:      cp,
+		pushProcessor: push.NewProcessor(),
+	}
+
+	cn, cleanup := newReaderBufferedPushConn(t, []byte(">2\r\n$10\r\ninvalidate\r\n*1\r\n$3\r\nfo"))
+	defer cleanup()
+
+	c.releaseConn(context.Background(), cn, nil)
+
+	if cp.removes != 1 || cp.puts != 0 {
+		t.Fatalf("reader-buffered partial push read must remove, not re-pool, the connection: removes=%d puts=%d",
+			cp.removes, cp.puts)
+	}
+}
+
+func TestReleaseConnRemovesConnectionAfterCustomProcessorError(t *testing.T) {
+	cp := &releaseRecordingPool{}
+	proc := erroringProcessor{push.NewProcessor(), proto.NewOOMError("OOM custom processor failure")}
+	c := &baseClient{
+		opt:           &Options{Addr: "127.0.0.1:6379", Protocol: 3},
+		connPool:      cp,
+		pushProcessor: proc,
+	}
+
+	cn, cleanup := newReaderBufferedPushConn(t, invalidateFrame("foo"))
+	defer cleanup()
+
+	c.releaseConn(context.Background(), cn, nil)
+
+	if cp.removes != 1 || cp.puts != 0 {
+		t.Fatalf("custom-processor drain error must remove, not re-pool, the connection: removes=%d puts=%d",
+			cp.removes, cp.puts)
+	}
+}
+
+func TestReleaseConnRemovesConnectionAfterVoidProcessorBufferedPartialPushRead(t *testing.T) {
+	cp := &releaseRecordingPool{}
+	c := &baseClient{
+		opt:           &Options{Addr: "127.0.0.1:6379", Protocol: 3},
+		connPool:      cp,
+		pushProcessor: push.NewVoidProcessor(),
+	}
+
+	cn, cleanup := newReaderBufferedPushConn(t, []byte(">2\r\n$10\r\ninvalidate\r\n*1\r\n$3\r\nfo"))
+	defer cleanup()
+
+	c.releaseConn(context.Background(), cn, nil)
+
+	if cp.removes != 1 || cp.puts != 0 {
+		t.Fatalf("void-processor buffered partial push read must remove, not re-pool, the connection: removes=%d puts=%d",
+			cp.removes, cp.puts)
+	}
+}
+
+func TestReleaseConnRemovesConnectionAfterVoidProcessorPeekError(t *testing.T) {
+	cp := &releaseRecordingPool{}
+	c := &baseClient{
+		opt:           &Options{Addr: "127.0.0.1:6379", Protocol: 3},
+		connPool:      cp,
+		pushProcessor: push.NewVoidProcessor(),
+	}
+
+	// Leading RESP3 attribute can be consumed by PeekReplyType before the
+	// following partial push frame fails; release-time drain must still treat
+	// that as fatal and remove the connection.
+	server, client := newIdleTCPConnPair(t)
+	defer server.Close()
+	defer client.Close()
+
+	cn := pool.NewConn(client)
+	frame := []byte("|1\r\n+meta\r\n>2\r\n$10\r\ninvalidate\r\n*1\r\n$3\r\nfo")
+	if _, err := server.Write(frame); err != nil {
+		t.Fatalf("write attr+partial push: %v", err)
+	}
+	_ = server.Close()
+	deadline := time.Now().Add(time.Second)
+	for !cn.MaybeHasData() && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if !cn.MaybeHasData() {
+		t.Fatal("attr+partial push never became readable")
+	}
+
+	c.releaseConn(context.Background(), cn, nil)
+
+	if cp.removes != 1 || cp.puts != 0 {
+		t.Fatalf("void-processor peek failure must remove, not re-pool, the connection: removes=%d puts=%d",
+			cp.removes, cp.puts)
+	}
+}
+
 // bufferedNetConn models a wrapper such as tls.Conn: bytes may already be
 // buffered inside the wrapper while NetConn's raw socket is empty.
 type bufferedNetConn struct {
@@ -2452,6 +2547,35 @@ func TestDrainPushNotifications_EmptyWrappedProbeStaysShort(t *testing.T) {
 	}
 }
 
+func TestDrainPushNotifications_EmptyWrappedProbeStaysShortWithVoidProcessor(t *testing.T) {
+	oldHardReadCap := cscDrainHardReadCap
+	cscDrainHardReadCap = 200 * time.Millisecond
+	defer func() { cscDrainHardReadCap = oldHardReadCap }()
+
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	c := &baseClient{opt: &Options{Protocol: 3}, pushProcessor: push.NewVoidProcessor()}
+	cn := pool.NewConn(&bufferedNetConn{
+		Conn:     client,
+		buffered: bytes.NewReader(nil),
+	})
+	cn.MarkCscReadPending()
+
+	start := time.Now()
+	processed, err := c.drainPushNotifications(cn)
+	if err != nil {
+		t.Fatalf("empty wrapped probe with void processor: %v", err)
+	}
+	if processed {
+		t.Fatal("empty wrapped probe with void processor must not report processor success")
+	}
+	if elapsed := time.Since(start); elapsed >= 50*time.Millisecond {
+		t.Fatalf("empty wrapped probe with void processor held the connection for %v", elapsed)
+	}
+}
+
 // TestDrainPushNotifications_ConsumesReaderBufferedPush is a regression guard: a
 // push buffered in proto.Reader (no socket data) must still drain — the gate
 // checks HasBufferedData(), not only MaybeHasData().

error.go

@@ -86,6 +86,11 @@ func shouldRetry(err error, retryTimeout bool) bool {
 		return false
 	}
 
+	var txReadErr *txQueuedReadError
+	if errors.As(err, &txReadErr) {
+		return false
+	}
+
 	// Check for EOF errors (works with wrapped errors)
 	if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
 		return true
@@ -190,6 +195,19 @@ func isBadConn(err error, allowTimeout bool, addr string) bool {
 		return false
 	}
 
+	var txReadErr *txQueuedReadError
+	if errors.As(err, &txReadErr) {
+		if txReadErr.forceBad {
+			return true
+		}
+		return isBadConn(txReadErr.readErr, allowTimeout, addr)
+	}
+
+	var forceBadErr *forceBadConnError
+	if errors.As(err, &forceBadErr) {
+		return true
+	}
+
 	// Check for context errors (works with wrapped errors)
 	if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
 		return true