Some probability exists that the pipeline only returns part of the results for certain commands.

[xuanskyer]

The issue occurs when using the Pipeline method to retrieve multiple content caches in a batch manner from Redis version 7. Some probability exists that the pipeline only returns part of the results for certain commands.

Here is an example of code that demonstrates the issue:

// some codes...  
redisResults := make([]*redisv7.StringStringMapCmd, len(contentIds))  
	pipe := redisClient.Pipeline()  
	for i, id := range contentIds {  
		redisResults[i] = pipe.HGetAll(id)  
	}  
	if _, err := pipe.Exec(); err != nil {  
		logger.Log().Error("err", zap.Error(err), zap.Any("contentIds", contentIds))  
		return nil, nil, errors.New("err exec")  
	}

	metaList = make([]ContentMeta, 0, len(contentIds))  
	missIds = make([]string, 0, len(contentIds))  
	errList := make([]string, 0)  
	for idx, result := range redisResults {  
		if val, err := result.Result(); err == nil && len(val) > 0 {  
			metaList = append(metaList, val)  
		} else {  
			errStr := fmt.Sprintf("id: %d, len(val): %d, content_id: %s, result: %s ", idx, len(val), contentIds[idx], result.String())  
			if err != nil {  
				errStr += fmt.Sprintf("err: %s", err.Error())  
			}  
			errList = append(errList, errStr)  
			missIds = append(missIds, contentIds[idx])  
		}  
	}  
	if len(errList) > 0 {  
		logger.Log().Error("Pipeline_err: ", zap.Error(err), zap.Any("contentIds", contentIds), zap.Any("errList", errList))  
	}  

Looking at the code log above, it was found that sometimes the following issues occur:

For example, when contentIds are: ["123", "456", "789"] (these three corresponding Redis cache keys do indeed exist), sometimes only 2 or 1 results are returned.

What is the reason for this?

Is it possible that it was due to the timeout?

[ljluestc]

package main

import (
	"fmt"
	"time"

	"github.com/go-redis/redis/v7"
	"go.uber.org/zap"
)

type ContentMeta map[string]string // Assuming ContentMeta is a map for simplicity

func fetchContentMeta(redisClient *redis.Client, contentIds []string, logger *zap.Logger) ([]ContentMeta, []string, error) {
	// Ensure client has reasonable timeouts
	if redisClient.Options().ReadTimeout == 0 {
		redisClient = redisClient.WithTimeout(5 * time.Second) // Adjust timeout as needed
	}

	// Prepare pipeline
	pipe := redisClient.Pipeline()
	redisResults := make([]*redis.StringStringMapCmd, len(contentIds))
	for i, id := range contentIds {
		redisResults[i] = pipe.HGetAll(id)
	}

	// Execute pipeline
	_, err := pipe.Exec()
	if err != nil && err != redis.Nil {
		logger.Error("Pipeline execution failed", zap.Error(err), zap.Any("contentIds", contentIds))
		return nil, nil, fmt.Errorf("pipeline exec error: %w", err)
	}

	// Process results
	metaList := make([]ContentMeta, 0, len(contentIds))
	missIds := make([]string, 0, len(contentIds))
	errList := make([]string, 0)

	for idx, result := range redisResults {
		val, err := result.Result()
		if err == redis.Nil || (err == nil && len(val) == 0) {
			// Key doesn’t exist or is empty
			errStr := fmt.Sprintf("id: %d, content_id: %s, result: %s", idx, contentIds[idx], result.String())
			errList = append(errList, errStr)
			missIds = append(missIds, contentIds[idx])
		} else if err != nil {
			// Command-level error
			errStr := fmt.Sprintf("id: %d, content_id: %s, result: %s, err: %s", idx, contentIds[idx], result.String(), err.Error())
			errList = append(errList, errStr)
			missIds = append(missIds, contentIds[idx])
		} else {
			// Success case
			metaList = append(metaList, val)
		}
	}

	// Log errors if any
	if len(errList) > 0 {
		logger.Error("Pipeline partial failure",
			zap.Any("contentIds", contentIds),
			zap.Any("errList", errList),
			zap.Int("missCount", len(missIds)))
	}

	return metaList, missIds, nil
}

// Example usage
func main() {
	// Initialize Redis client
	redisClient := redis.NewClient(&redis.Options{
		Addr:         "localhost:6379",
		DialTimeout:  5 * time.Second,  // Connection timeout
		ReadTimeout:  5 * time.Second,  // Read timeout
		WriteTimeout: 5 * time.Second,  // Write timeout
		PoolSize:     10,               // Connection pool size
	})

	// Initialize logger (simplified)
	logger, _ := zap.NewProduction()

	// Test content IDs
	contentIds := []string{"123", "456", "789"}
	metaList, missIds, err := fetchContentMeta(redisClient, contentIds, logger)
	if err != nil {
		logger.Error("Failed to fetch content", zap.Error(err))
		return
	}

	fmt.Printf("MetaList: %+v\n", metaList)
	fmt.Printf("MissIds: %+v\n", missIds)
}
​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​​

[ndyakov]

Please retest with lates v9 release and reopen if needed.