command_entry_array_desync_test.go

package redis

import (
	"context"
	"strings"
	"testing"

	"github.com/redis/go-redis/v9/internal/proto"
)

// These parsers read a fixed set of fields from each entry array but did not
// reconcile that with the element count the server declares. A server that
// declares a longer array leaves the trailing frames on the wire; readReply
// returns nil and the desynced connection goes back to the pool, so the next
// command reads the leftover frames as its own reply. Each case appends a
// sentinel reply and checks it is still readable after readReply.

func TestEntryArrayDrainsExtraElements(t *testing.T) {
	t.Run("LatencyCmd", func(t *testing.T) {
		// entry declares 5 elements: the 4 fields plus one extra
		rd := proto.NewReader(strings.NewReader(
			"*1\r\n*5\r\n$5\r\nevent\r\n:1000\r\n:5\r\n:9\r\n:7\r\n+PONG\r\n"))
		cmd := NewLatencyCmd(context.Background())
		if err := cmd.readReply(rd); err != nil {
			t.Fatalf("readReply: %v", err)
		}
		assertSentinel(t, rd)
	})

	t.Run("SlowLogCmd", func(t *testing.T) {
		// entry declares 8 elements: the 7 known fields (through CommandArgc)
		// plus one extra that only the drain loop can consume
		rd := proto.NewReader(strings.NewReader(
			"*1\r\n*8\r\n:1\r\n:1000\r\n:50\r\n*1\r\n$3\r\nGET\r\n$5\r\n1.2.3\r\n$4\r\nname\r\n:99\r\n$5\r\nextra\r\n+PONG\r\n"))
		cmd := NewSlowLogCmd(context.Background())
		if err := cmd.readReply(rd); err != nil {
			t.Fatalf("readReply: %v", err)
		}
		if len(cmd.val) != 1 || cmd.val[0].CommandArgc != 99 {
			t.Fatalf("CommandArgc not parsed before drain: %+v", cmd.val)
		}
		assertSentinel(t, rd)
	})

	t.Run("HotKeysCmd", func(t *testing.T) {
		// outer array wraps two elements; only the first is parsed
		rd := proto.NewReader(strings.NewReader(
			"*2\r\n%1\r\n$15\r\ntracking-active\r\n:1\r\n%0\r\n+PONG\r\n"))
		cmd := NewHotKeysCmd(context.Background())
		if err := cmd.readReply(rd); err != nil {
			t.Fatalf("readReply: %v", err)
		}
		assertSentinel(t, rd)
	})
}

// LATENCY GET entries hold 4 fields; a 3-element entry must be rejected rather
// than over-read into the next reply.
func TestLatencyCmdShortEntryRejected(t *testing.T) {
	// The trailing :9 belongs to the next reply; the unpatched parser reads it
	// as the 4th field and returns nil, desyncing the connection.
	rd := proto.NewReader(strings.NewReader(
		"*1\r\n*3\r\n$5\r\nevent\r\n:1000\r\n:5\r\n:9\r\n"))
	cmd := NewLatencyCmd(context.Background())
	if err := cmd.readReply(rd); err == nil {
		t.Fatalf("short latency entry accepted; readReply returned nil (val=%+v)", cmd.val)
	}
}

// GeoSearchLocationCmd must reconcile each entry array against the requested
// WITH flags: extra elements are drained (forwards compatible with servers
// that append fields), while entries shorter than the flags require are
// rejected rather than over-read into the next reply.
func TestGeoSearchLocationEntryLengthReconciled(t *testing.T) {
	opt := &GeoSearchLocationQuery{WithDist: true} // withLen == 2

	t.Run("oversized drained", func(t *testing.T) {
		rd := proto.NewReader(strings.NewReader(
			"*1\r\n*3\r\n$1\r\na\r\n$3\r\n1.5\r\n:9\r\n+PONG\r\n"))
		cmd := NewGeoSearchLocationCmd(context.Background(), opt)
		if err := cmd.readReply(rd); err != nil {
			t.Fatalf("readReply: %v", err)
		}
		if len(cmd.val) != 1 || cmd.val[0].Name != "a" || cmd.val[0].Dist != 1.5 {
			t.Fatalf("unexpected val: %+v", cmd.val)
		}
		assertSentinel(t, rd)
	})

	t.Run("short rejected", func(t *testing.T) {
		rd := proto.NewReader(strings.NewReader(
			"*1\r\n*1\r\n$1\r\na\r\n"))
		cmd := NewGeoSearchLocationCmd(context.Background(), opt)
		if err := cmd.readReply(rd); err == nil {
			t.Fatalf("short geo entry accepted; readReply returned nil (val=%+v)", cmd.val)
		}
	})

	t.Run("valid parses", func(t *testing.T) {
		rd := proto.NewReader(strings.NewReader(
			"*1\r\n*2\r\n$1\r\na\r\n$3\r\n1.5\r\n"))
		cmd := NewGeoSearchLocationCmd(context.Background(), opt)
		if err := cmd.readReply(rd); err != nil {
			t.Fatalf("valid geo entry: %v", err)
		}
		if len(cmd.val) != 1 || cmd.val[0].Name != "a" || cmd.val[0].Dist != 1.5 {
			t.Fatalf("unexpected val: %+v", cmd.val)
		}
	})
}

func assertSentinel(t *testing.T, rd *proto.Reader) {
	t.Helper()
	s, err := rd.ReadString()
	if err != nil {
		t.Fatalf("stream desynced, sentinel unreadable: %v", err)
	}
	if s != "PONG" {
		t.Fatalf("stream desynced, got %q want PONG", s)
	}
}