fix pipeline Get & GetPut panic on missing keys

(cherry picked from commit 0ce1e6a)

Author: derekperkins

pipeline.go

@@ -18,6 +18,7 @@ import (
 	"bytes"
 	"context"
 	"errors"
+	"fmt"
 	"runtime"
 	"strconv"
 	"sync"
@@ -164,8 +165,15 @@ func (f *FutureGet) Result() (*GetResponse, error) {
 		if cmd.Err() != nil {
 			return nil, processProtocolError(cmd.Err())
 		}
+		value, nilValue, err := pipelineStringValue(cmd)
+		if err != nil {
+			return nil, err
+		}
+		if nilValue {
+			return nil, ErrKeyNotFound
+		}
 		stringCmd := redis.NewStringCmd(context.Background(), cmd.Args()...)
-		stringCmd.SetVal(cmd.(*redis.Cmd).Val().(string))
+		stringCmd.SetVal(value)
 		return f.dp.dm.makeGetResponse(stringCmd)
 	default:
 		return nil, ErrNotReady
@@ -392,8 +400,15 @@ func (f *FutureGetPut) Result() (*GetResponse, error) {
 		if cmd.Err() != nil {
 			return nil, processProtocolError(cmd.Err())
 		}
+		value, nilValue, err := pipelineStringValue(cmd)
+		if err != nil {
+			return nil, err
+		}
+		if nilValue {
+			return nil, nil
+		}
 		stringCmd := redis.NewStringCmd(context.Background(), cmd.Args()...)
-		stringCmd.SetVal(cmd.(*redis.Cmd).Val().(string))
+		stringCmd.SetVal(value)
 		return f.dp.dm.makeGetResponse(stringCmd)
 	default:
 		return nil, ErrNotReady
@@ -645,3 +660,20 @@ func putPipelineCmdsIntoPool(cmds []redis.Cmder) {
 	cmds = cmds[:0]
 	pipelineCmdPool.Put(cmds)
 }
+
+func pipelineStringValue(cmd redis.Cmder) (string, bool, error) {
+	redisCmd, ok := cmd.(*redis.Cmd)
+	if !ok {
+		return "", false, fmt.Errorf("unexpected pipeline command type %T", cmd)
+	}
+
+	value := redisCmd.Val()
+	if value == nil {
+		return "", true, nil
+	}
+	stringValue, ok := value.(string)
+	if !ok {
+		return "", false, fmt.Errorf("unexpected pipeline command value type %T", value)
+	}
+	return stringValue, false, nil
+}

pipeline_test.go

@@ -20,8 +20,10 @@ import (
 	"testing"
 	"time"
 
-	"github.com/olric-data/olric/internal/testutil"
+	"github.com/redis/go-redis/v9"
 	"github.com/stretchr/testify/require"
+
+	"github.com/olric-data/olric/internal/testutil"
 )
 
 func TestDMapPipeline_Put(t *testing.T) {
@@ -107,6 +109,25 @@ func TestDMapPipeline_Get(t *testing.T) {
 	}
 }
 
+func TestFutureGet_Result_NilPipelineValue(t *testing.T) {
+	cmd := redis.NewCmd(context.Background(), "dm.get", "mydmap", "missing", "RW")
+	cmd.SetVal(nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	future := &FutureGet{
+		dp: &DMapPipeline{
+			result: map[uint64][]redis.Cmder{0: {cmd}},
+		},
+		ctx:       ctx,
+		closedCtx: context.Background(),
+	}
+
+	res, err := future.Result()
+	require.Nil(t, res)
+	require.ErrorIs(t, err, ErrKeyNotFound)
+}
+
 func TestDMapPipeline_Delete(t *testing.T) {
 	cluster := newTestOlricCluster(t)
 	db := cluster.addMember(t)
@@ -293,6 +314,25 @@ func TestDMapPipeline_GetPut(t *testing.T) {
 	}
 }
 
+func TestFutureGetPut_Result_NilPipelineValue(t *testing.T) {
+	cmd := redis.NewCmd(context.Background(), "dm.getput", "mydmap", "key", "value", "RW")
+	cmd.SetVal(nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	future := &FutureGetPut{
+		dp: &DMapPipeline{
+			result: map[uint64][]redis.Cmder{0: {cmd}},
+		},
+		ctx:       ctx,
+		closedCtx: context.Background(),
+	}
+
+	res, err := future.Result()
+	require.NoError(t, err)
+	require.Nil(t, res)
+}
+
 func TestDMapPipeline_IncrByFloat(t *testing.T) {
 	cluster := newTestOlricCluster(t)
 	db := cluster.addMember(t)