Pipe commands getting stuck intermittently during scale up and scale down of elasticache clusters

[srikar-jilugu]

We’re using valkey-go v1.0.57 to connect to an ElastiCache cluster with around 6 shards and 4 replicas, with replica auto-scaling enabled. During scale-up or scale-down events, we intermittently observe that pipe.Exec(...) calls hang indefinitely, despite having both connection timeouts and context deadlines set to 2 seconds.

We have faced this issue with pipelined HGetAll and Get commands.

Relevant client configurations:

SendToReplicas: true
Connection timeout: 2s
Ctx Deadline for each command: 2s

We’re trying to understand why this indefinite hang happens during resharding or topology changes.

[rueian]

I tested the same timeout settings against this ElastiCache cluster several times, but still didn't reproduce the issue.

I think it would be helpful if you could provide a sample program for reproducing.

[rueian]

Hi @srikar-jilugu, any updates or more info for us to reproduce the issue?

[srikar-jilugu]

Hi @rueian, apologies for the delayed update, Can you please reopen this issue?

I’ve put together a sample program that should help reproduce the issue. It’s essentially the same code that occasionally encounters the stuck behavior.

You can try running it with a large number of keys and higher throughput, then scale the cluster to trigger the problem.

package main

import (
	"context"
	"errors"
	"fmt"
	"log"
	"net"
	"sync"
	"time"

	"github.com/valkey-io/valkey-go"
	valkeyCompat "github.com/valkey-io/valkey-go/valkeycompat"
)

// boundedGroup provides a simple bounded wait group for controlling concurrency.
type boundedGroup struct {
	limit chan struct{}
	wg    sync.WaitGroup
}

func ChunkArray[T any](slice []T, chunkSize int) [][]T {
	var chunks [][]T
	if chunkSize == 0 {
		chunks = append(chunks, slice)
		return chunks
	}
	for i := 0; i < len(slice); i += chunkSize {
		end := i + chunkSize

		// necessary check to avoid slicing beyond
		// slice capacity
		if end > len(slice) {
			end = len(slice)
		}

		chunks = append(chunks, slice[i:end])
	}

	return chunks
}
func newBoundedGroup(limit int) *boundedGroup {
	if limit <= 0 {
		limit = 1
	}
	return &boundedGroup{limit: make(chan struct{}, limit)}
}

func (b *boundedGroup) Go(fn func()) {
	b.wg.Add(1)
	b.limit <- struct{}{}
	go func() {
		defer b.wg.Done()
		defer func() { <-b.limit }()
		fn()
	}()
}

// The Add method of the BoundedWaitGroup works similarly to the Add method of the regular WaitGroup.
// However, it also acquires a token from the limit channel for each new goroutine that is being added.
func (bwg *boundedGroup) Add(delta int) {
	for i := 0; i < delta; i++ {
		bwg.limit <- struct{}{}
	}
	bwg.wg.Add(delta)
}

// The Done method of the BoundedWaitGroup releases a token back to the limit channel, indicating that a goroutine has finished.
func (bwg *boundedGroup) Done() {
	<-bwg.limit
	bwg.wg.Done()
}

// The Wait method of the BoundedWaitGroup works the same as the regular WaitGroup, blocking until all the added goroutines have finished.
func (bwg *boundedGroup) Wait() {
	bwg.wg.Wait()
}

// fetchHashes concurrently performs HGETALL for the provided keys with a bounded level of concurrency.
// FetchValkeyHashes establishes (or reuses) a valkey client and performs concurrent HGETALL for the provided keys.
// It returns a map[key]hashFields. Errors for individual keys are stored under the special field "__error".
func FetchValkeyHashes(ctx context.Context, client valkeyCompat.Cmdable, keys []string, concurrency int) map[string]map[string]string {
	results := make(map[string]map[string]string, len(keys))
	g := newBoundedGroup(concurrency)

	chunks := ChunkArray(keys, 100)
	for _, k := range chunks {
		g.Add(1)
		go func() {
			defer g.Done()
			resp, err := PipeHGetAll(ctx, client, k)
			if err != nil {
				// store error marker for visibility
				fmt.Println("Error fetching key:", k, "Error:", err)
				return
			}
			fmt.Println("Fetched key:", k, "Response:", resp)

		}()
	}
	g.Wait()
	return results
}

// newValkeyClient initializes a valkey client using VALKEY_ADDR env var (default 127.0.0.1:6379).
func NewValkeyClient(ctx context.Context) (valkeyCompat.Cmdable, error) {

	valkeyClient, err := valkey.NewClient(
		valkey.ClientOption{
			InitAddress: []string{"127.0.0.1:6379"},
			SendToReplicas: func(cmd valkey.Completed) bool {
				return cmd.IsReadOnly()
			},

			Dialer: net.Dialer{
				Timeout: time.Duration(3000) * time.Millisecond,
			},
			BlockingPoolSize:    75,
			BlockingPoolMinSize: 6,
			ConnWriteTimeout:    2000,
		},
	)

	if err != nil {
		log.Panic("Valkey Initialisation Failed:", err)
		return nil, err
	}

	valkeyCompatClient := valkeyCompat.NewAdapter(valkeyClient)

	return valkeyCompatClient, nil
}

func PipeHGetAll(ctx context.Context, client valkeyCompat.Cmdable, keys []string) (map[string]map[string]string, error) {
	responseData := make(map[string]map[string]string, 0)
	redisCmds := make(map[string]*valkeyCompat.StringStringMapCmd, 0)
	pipe := client.Pipeline()
	for i := range keys {
		redisCmds[keys[i]] = pipe.HGetAll(ctx, keys[i])
	}

	_, err := pipe.Exec(ctx)
	if err != nil && !errors.Is(err, valkey.Nil) {
		fmt.Printf("[PipeHGetAll] cmdable error: %s\n", err.Error())
		return responseData, err
	}

	for key, strCmd := range redisCmds {
		resultMap, err := strCmd.Result()
		if err != nil {
			if errors.Is(err, valkey.Nil) {
				responseData[key] = map[string]string{}
				fmt.Printf("[PipeHGetAll] found no documents for key:%v\n", key)
			} else {
				fmt.Printf("[PipeHGetAll] cache lookup error for key:%v error:%v\n", key, err.Error())
			}
			continue
		}
		if len(resultMap) == 0 {
			fmt.Printf("[PipeHGetAll] returned empty map for key:%v\n", key)
			continue
		}
		responseData[key] = resultMap
	}
	return responseData, nil
}

func main() {
	ctx := context.Background()
	client, err := NewValkeyClient(ctx)
	if err != nil {
		log.Panic("Error initializing valkey client:", err)
	}
	keys := []string{"key1", "key2", "key3", "key4", "key5"}
	results := FetchValkeyHashes(ctx, client, keys, 2)
	fmt.Println("Final Results:", results)
}

[rueian]

Hi @srikar-jilugu,

I tried your example with a small fix: ConnWriteTimeout: 2000 * time.Millisecond and 1000 keys with concurrency=10:

	var keys []string
	for i := 0; i < 1000; i++ {
		if err := client.HMSet(ctx, strconv.Itoa(i), strconv.Itoa(i), strconv.Itoa(i)).Err(); err != nil {
			panic(err)
		}
		keys = append(keys, strconv.Itoa(i))
	}
	for {
		results := FetchValkeyHashes(ctx, client, keys, 10)
		fmt.Println("Final Results:", results)
	}

but I still couldn't reproduce the issue. I was using valkey-go 1.0.64 against a Valkey 8.1 elasticache cluster, scaling up and down between 2 and 3 shards with 1 replica on each shard repeatedly.

Is it possible for you to have a goroutine dump when you find it hangs indefinitely next time? (kill -ABRT or print the goroutine dump manually)