fix(search): read the full RESP3 map reply in FTHybridCmd (#3956)

FTHybridCmd.readReply used ReadSlice, which reads a RESP3 map header's
length as an element count and consumes only half its frames, leaving the
connection desynced. Peek the type and read a map with ReadReply, like the
other FT.* parsers.

search_commands.go

@@ -3152,11 +3152,37 @@ func parseFTHybrid(data []interface{}, withCursor bool) (FTHybridResult, *FTHybr
 }
 
 func (cmd *FTHybridCmd) readReply(rd *proto.Reader) (err error) {
-	data, err := rd.ReadSlice()
+	readType, err := rd.PeekReplyType()
 	if err != nil {
 		return err
 	}
 
+	// RESP3 returns a map, RESP2 returns an array. ReadSlice reads a map
+	// header's declared length as an element count, so it consumes only half
+	// the key/value frames and leaves the rest on the connection; ReadReply
+	// consumes the whole map. Flatten it into the key/value list parseFTHybrid
+	// expects, matching AggregateCmd/FTSearchCmd/FTSpellCheckCmd/FTSynDumpCmd.
+	var data []interface{}
+	if readType == proto.RespMap {
+		cmd.rawVal, err = rd.ReadReply()
+		if err != nil {
+			return err
+		}
+		rawMap, ok := cmd.rawVal.(map[interface{}]interface{})
+		if !ok {
+			return fmt.Errorf("unexpected RESP3 response type: %T", cmd.rawVal)
+		}
+		data = make([]interface{}, 0, len(rawMap)*2)
+		for k, v := range rawMap {
+			data = append(data, k, v)
+		}
+	} else {
+		data, err = rd.ReadSlice()
+		if err != nil {
+			return err
+		}
+	}
+
 	result, cursorResult, err := parseFTHybrid(data, cmd.withCursor)
 	if err != nil {
 		return err

search_hybrid_desync_test.go

@@ -0,0 +1,68 @@
+package redis
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"github.com/redis/go-redis/v9/internal/proto"
+)
+
+// FT.HYBRID replies with a RESP3 map on a RESP3 connection, like the other
+// FT.* commands. FTHybridCmd.readReply must consume the whole map: reading it
+// with ReadSlice treats the map header's declared length as an ELEMENT count,
+// so only half the key/value frames are consumed and the rest are left on the
+// connection. readReply then returns nil (success), the connection is pooled
+// desynced, and the next command reads a stray map frame as its own reply.
+func TestFTHybridRESP3MapNoDesync(t *testing.T) {
+	// A 4-pair RESP3 map, followed by the next command's reply on the wire.
+	const reply = "%4\r\n" +
+		"$13\r\ntotal_results\r\n:5\r\n" +
+		"$7\r\nresults\r\n*0\r\n" +
+		"$6\r\nformat\r\n$6\r\nSTRING\r\n" +
+		"$7\r\nwarning\r\n*0\r\n"
+	const next = ":99\r\n"
+
+	rd := proto.NewReader(strings.NewReader(reply + next))
+	cmd := newFTHybridCmd(context.Background(), nil)
+	if err := cmd.readReply(rd); err != nil {
+		t.Fatalf("readReply: %v", err)
+	}
+	if cmd.val.TotalResults != 5 {
+		t.Fatalf("TotalResults = %d, want 5", cmd.val.TotalResults)
+	}
+
+	// The whole map must have been consumed: the next reply on the wire is the
+	// following command's integer 99, not a leftover map frame.
+	got, err := rd.ReadReply()
+	if err != nil {
+		t.Fatalf("reading next reply: %v", err)
+	}
+	if n, ok := got.(int64); !ok || n != 99 {
+		t.Fatalf("connection desynced: next reply = %#v, want int64(99)", got)
+	}
+}
+
+// The RESP2 flat-array form must still parse and stay in sync.
+func TestFTHybridRESP2ArrayStillParses(t *testing.T) {
+	const reply = "*4\r\n" +
+		"$13\r\ntotal_results\r\n:7\r\n" +
+		"$7\r\nresults\r\n*0\r\n"
+	const next = ":42\r\n"
+
+	rd := proto.NewReader(strings.NewReader(reply + next))
+	cmd := newFTHybridCmd(context.Background(), nil)
+	if err := cmd.readReply(rd); err != nil {
+		t.Fatalf("readReply: %v", err)
+	}
+	if cmd.val.TotalResults != 7 {
+		t.Fatalf("TotalResults = %d, want 7", cmd.val.TotalResults)
+	}
+	got, err := rd.ReadReply()
+	if err != nil {
+		t.Fatalf("reading next reply: %v", err)
+	}
+	if n, ok := got.(int64); !ok || n != 42 {
+		t.Fatalf("connection desynced: next reply = %#v, want int64(42)", got)
+	}
+}