feat: trim ERR prefix of redis error message to accommodate `ERR NOSC…

…RIPT` message from kvrocks

message.go

@@ -345,7 +345,12 @@ func (m *RedisMessage) IsMap() bool {
 
 // Error check if message is a redis error response, including nil response
 func (m *RedisMessage) Error() error {
-	if m.typ == '-' || m.typ == '_' || m.typ == '!' {
+	if m.typ == '_' {
+		return (*RedisError)(m)
+	}
+	if m.typ == '-' || m.typ == '!' {
+		// kvrocks: https://github.com/rueian/rueidis/issues/152#issuecomment-1333923750
+		m.string = strings.TrimPrefix(m.string, "ERR ")
 		return (*RedisError)(m)
 	}
 	return nil

message_test.go

@@ -467,6 +467,12 @@ func TestRedisMessage(t *testing.T) {
 			t.Fatal("IsNil fail")
 		}
 	})
+	t.Run("Trim ERR prefix", func(t *testing.T) {
+		// kvrocks: https://github.com/rueian/rueidis/issues/152#issuecomment-1333923750
+		if (&RedisMessage{typ: '-', string: "ERR no_prefix"}).Error().Error() != "no_prefix" {
+			t.Fatal("fail to trim ERR")
+		}
+	})
 	t.Run("ToInt64", func(t *testing.T) {
 		if _, err := (&RedisMessage{typ: '_'}).ToInt64(); err == nil {
 			t.Fatal("ToInt64 not failed as expected")

redis_test.go

@@ -3,6 +3,7 @@ package rueidis
 import (
 	"context"
 	"math/rand"
+	"reflect"
 	"strconv"
 	"sync"
 	"sync/atomic"
@@ -487,6 +488,31 @@ func testPubSub(t *testing.T, client Client) {
 	}
 }
 
+func testLua(t *testing.T, client Client) {
+	script := NewLuaScript("return {KEYS[1],ARGV[1]}")
+
+	keys := 1000
+	para := 4
+	kvs := make(map[string]string, keys)
+	for i := 0; i < keys; i++ {
+		kvs["m"+strconv.Itoa(i)] = strconv.FormatInt(rand.Int63(), 10)
+	}
+
+	t.Logf("testing lua with %d keys and %d parallelism\n", keys, para)
+	jobs, wait := parallel(para)
+	for k, v := range kvs {
+		k := k
+		v := v
+		jobs <- func() {
+			val, err := script.Exec(context.Background(), client, []string{k}, []string{v}).AsStrSlice()
+			if err != nil || !reflect.DeepEqual(val, []string{k, v}) {
+				t.Errorf("unexpected lua response %v %v", val, err)
+			}
+		}
+	}
+	wait()
+}
+
 func run(t *testing.T, client Client, cases ...func(*testing.T, Client)) {
 	wg := sync.WaitGroup{}
 	wg.Add(len(cases))
@@ -509,7 +535,7 @@ func TestSingleClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 	run(t, client, testFlush)
 
 	client.Close()
@@ -529,7 +555,7 @@ func TestSentinelClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 	run(t, client, testFlush)
 
 	client.Close()
@@ -545,7 +571,7 @@ func TestClusterClientIntegration(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 
 	client.Close()
 	time.Sleep(time.Second * 5) // wait background ping exit
@@ -561,7 +587,7 @@ func TestSingleClient5Integration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 
 	client.Close()
 	time.Sleep(time.Second * 5) // wait background ping exit
@@ -577,7 +603,7 @@ func TestCluster5ClientIntegration(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 
 	client.Close()
 	time.Sleep(time.Second * 5) // wait background ping exit
@@ -596,7 +622,7 @@ func TestSentinel5ClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 
 	client.Close()
 	time.Sleep(time.Second * 5) // wait background ping exit
@@ -611,7 +637,7 @@ func TestKeyDBSingleClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub)
+	run(t, client, testSETGETCSC, testMultiSETGETCSC, testBlockingZPOP, testBlockingXREAD, testPubSub, testLua)
 	run(t, client, testFlush)
 
 	client.Close()
@@ -628,7 +654,7 @@ func TestDragonflyDBSingleClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testPubSub)
+	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testPubSub, testLua)
 
 	client.Close()
 	time.Sleep(time.Second * 5) // wait background ping exit
@@ -644,7 +670,7 @@ func TestKvrocksSingleClientIntegration(t *testing.T) {
 		t.Fatal(err)
 	}
 
-	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testPubSub)
+	run(t, client, testSETGETRESP2, testMultiSETGETRESP2, testPubSub, testLua)
 	run(t, client, testFlush)
 
 	client.Close()

rueidis.go

@@ -293,4 +293,4 @@ func dial(dst string, opt *ClientOption) (conn net.Conn, err error) {
 	return conn, err
 }
 
-const redisErrMsgCommandNotAllow = "ERR command is not allowed"
+const redisErrMsgCommandNotAllow = "command is not allowed"

rueidiscompat/adapter_test.go

@@ -155,10 +155,10 @@ func testCluster(resp3 bool) {
 			Expect(kc).To(Equal(int64(1)))
 		})
 		It("ClusterCountFailureReports", func() {
-			Expect(adapter.ClusterCountFailureReports(ctx, "1").Err()).To(MatchError("ERR Unknown node 1"))
+			Expect(adapter.ClusterCountFailureReports(ctx, "1").Err()).To(MatchError("Unknown node 1"))
 		})
 		It("ClusterSlaves", func() {
-			Expect(adapter.ClusterSlaves(ctx, "1").Err()).To(MatchError("ERR Unknown node 1"))
+			Expect(adapter.ClusterSlaves(ctx, "1").Err()).To(MatchError("Unknown node 1"))
 		})
 	})
 }
@@ -211,7 +211,7 @@ func testAdapter(resp3 bool) {
 
 		It("should ClientKill", func() {
 			r := adapter.ClientKill(ctx, "1.1.1.1:1111")
-			Expect(r.Err()).To(MatchError("ERR No such client"))
+			Expect(r.Err()).To(MatchError("No such client"))
 			Expect(r.Val()).To(Equal(""))
 		})
 
@@ -1106,7 +1106,7 @@ func testAdapter(resp3 bool) {
 			Expect(set.Val()).To(Equal("OK"))
 
 			decr = adapter.Decr(ctx, "key")
-			Expect(decr.Err()).To(MatchError("ERR value is not an integer or out of range"))
+			Expect(decr.Err()).To(MatchError("value is not an integer or out of range"))
 			Expect(decr.Val()).To(Equal(int64(0)))
 		})
 
@@ -5840,14 +5840,14 @@ func testAdapterCache(resp3 bool) {
 			time.Sleep(2 * time.Second)
 		})
 		It("Config Rewrite", func() {
-			Expect(adapter.ConfigRewrite(ctx).Err()).To(MatchError("ERR The server is running without a config file"))
+			Expect(adapter.ConfigRewrite(ctx).Err()).To(MatchError("The server is running without a config file"))
 		})
 		It("DebugObject", func() {
-			Expect(adapter.DebugObject(ctx, "non").Err().Error()).To(HavePrefix("ERR DEBUG command not allowed."))
+			Expect(adapter.DebugObject(ctx, "non").Err().Error()).To(HavePrefix("DEBUG command not allowed."))
 		})
 		It("ReadOnly & ReadWrite", func() {
-			Expect(adapter.ReadOnly(ctx).Err()).To(MatchError("ERR This instance has cluster support disabled"))
-			Expect(adapter.ReadWrite(ctx).Err()).To(MatchError("ERR This instance has cluster support disabled"))
+			Expect(adapter.ReadOnly(ctx).Err()).To(MatchError("This instance has cluster support disabled"))
+			Expect(adapter.ReadWrite(ctx).Err()).To(MatchError("This instance has cluster support disabled"))
 		})
 	})