Bump nexus for raft-entires-on-disk + other fixes

go.mod

@@ -10,7 +10,7 @@ require (
 	github.com/dgraph-io/ristretto v0.1.0
 	github.com/dgryski/go-farm v0.0.0-20200201041132-a6ae2369ad13 // indirect
 	github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4
-	github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a
+	github.com/flipkart-incubator/nexus v0.0.0-20220629150345-4b11935eda50
 	github.com/gogo/protobuf v1.3.2
 	github.com/golang/protobuf v1.5.2
 	github.com/gorilla/mux v1.8.0
@@ -25,7 +25,7 @@ require (
 	github.com/spf13/viper v1.10.1
 	github.com/vmihailenco/msgpack/v5 v5.3.4
 	go.uber.org/zap v1.17.0
-	golang.org/x/sys v0.0.0-20220310020820-b874c991c1a5 // indirect
+	golang.org/x/sys v0.0.0-20220405210540-1e041c57c461 // indirect
 	google.golang.org/grpc v1.43.0
 	google.golang.org/protobuf v1.28.0
 	gopkg.in/ini.v1 v1.66.2

go.sum

@@ -92,8 +92,8 @@ github.com/fatih/color v1.9.0/go.mod h1:eQcE1qtQxscV5RaZvpXrrb8Drkc3/DdQ+uUYCNjL
 github.com/fatih/color v1.13.0/go.mod h1:kLAiJbzzSOZDVNGyDpeOxJ47H46qBXwg5ILebYFFOfk=
 github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4 h1:9zPLm5QKcBp5xUOBMWhRZPU+iwfl6xJtbSmbazmgy2g=
 github.com/flipkart-incubator/gorocksdb v0.0.0-20210920082714-1f7dcbb7b2e4/go.mod h1:kvJSXc90Ifw0rxuTxEHKq6UH/7hQ/gd9RKCyD94ctJ0=
-github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a h1:PrHQUQEqNV0tP8j9yvJWJkOwVLTwcsxqvt33dpWxt4Q=
-github.com/flipkart-incubator/nexus v0.0.0-20220316072727-c44c4b25144a/go.mod h1:eEnRO5Yjl4OgxHbLjtmGdx/iQSgUqtLEOEGEGhka8Wo=
+github.com/flipkart-incubator/nexus v0.0.0-20220629150345-4b11935eda50 h1:RTXTdIq0+LRtbCbgBNRqNd77z67Dpoe84w2MgBQfz2w=
+github.com/flipkart-incubator/nexus v0.0.0-20220629150345-4b11935eda50/go.mod h1:cBJAQa9UoAjjY85k7byQ/oKV+f+BLUlacWZx5agM6HM=
 github.com/fsnotify/fsnotify v1.4.7/go.mod h1:jwhsz4b93w/PPRr/qN1Yymfu8t87LnFCMoQvtojpjFo=
 github.com/fsnotify/fsnotify v1.5.1 h1:mZcQUHVQUQWoPXXtuf9yuEXKudkV2sx1E06UadKWpgI=
 github.com/fsnotify/fsnotify v1.5.1/go.mod h1:T3375wBYaZdLLcVNkcVbzGHY7f1l/uK5T5Ai1i3InKU=
@@ -108,7 +108,6 @@ github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/me
 github.com/godbus/dbus/v5 v5.0.4/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
 github.com/gogo/protobuf v1.1.1/go.mod h1:r8qH/GZQm5c6nD/R0oafs1akxWv10x8SbQlK7atdtwQ=
 github.com/gogo/protobuf v1.2.1/go.mod h1:hp+jE20tsWTFYpLwKvXlhS1hjn+gTNwPg2I6zVXpSg4=
-github.com/gogo/protobuf v1.3.1/go.mod h1:SlYgWuQ5SjCEi6WLHjHCa1yvBfUnHcTbrrZtXPKa29o=
 github.com/gogo/protobuf v1.3.2 h1:Ov1cvc58UF3b5XjBnZv7+opcTcQFZebYjWzi34vdm4Q=
 github.com/gogo/protobuf v1.3.2/go.mod h1:P1XiOD3dCwIKUDQYPy72D8LYyHL2YPYrpS2s69NZV8Q=
 github.com/golang/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
@@ -218,7 +217,6 @@ github.com/json-iterator/go v1.1.11/go.mod h1:KdQUCv79m/52Kvf8AW2vK1V8akMuk1QjK/
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/julienschmidt/httprouter v1.2.0/go.mod h1:SYymIcj16QtmaHHD7aYtjjsJG7VTCxuUUipMqKk8s4w=
 github.com/kisielk/errcheck v1.1.0/go.mod h1:EZBBE59ingxPouuu3KfxchcWSUPOHkagtvWXihfKN4Q=
-github.com/kisielk/errcheck v1.2.0/go.mod h1:/BMXB+zMLi60iA8Vv6Ksmxu/1UDYcXs4uQLJ+jE2L00=
 github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8=
 github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck=
 github.com/klauspost/compress v1.12.3 h1:G5AfA94pHPysR56qqrkO2pxEexdDzrpFJ6yt/VqWxVU=

internal/master/ds_service_test.go

@@ -26,6 +26,7 @@ const (
 	clusterSize           = 3
 	logDir                = "/tmp/dkv_test/logs"
 	snapDir               = "/tmp/dkv_test/snap"
+	entDir                = "/tmp/dkv_test/ent"
 	clusterURL            = "http://127.0.0.1:9321,http://127.0.0.1:9322,http://127.0.0.1:9323"
 	clusterStreamingURL   = "http://127.0.0.1:9321,http://127.0.0.1:9322,http://127.0.0.1:9323"
 	replTimeout           = 3 * time.Second
@@ -473,17 +474,14 @@ func initStreamingDKVClients(ids ...int) {
 }
 
 func resetRaftStateDirs(t *testing.T) {
-	if err := exec.Command("rm", "-rf", logDir).Run(); err != nil {
-		t.Fatal(err)
-	}
-	if err := exec.Command("mkdir", "-p", logDir).Run(); err != nil {
-		t.Fatal(err)
-	}
-	if err := exec.Command("rm", "-rf", snapDir).Run(); err != nil {
-		t.Fatal(err)
-	}
-	if err := exec.Command("mkdir", "-p", snapDir).Run(); err != nil {
-		t.Fatal(err)
+	dirs := []string{logDir, snapDir, entDir}
+	for _, dir := range dirs {
+		if err := exec.Command("rm", "-rf", dir).Run(); err != nil {
+			t.Fatal(err)
+		}
+		if err := exec.Command("mkdir", "-p", dir).Run(); err != nil {
+			t.Fatal(err)
+		}
 	}
 }
 
@@ -505,6 +503,7 @@ func newReplicator(kvs storage.KVStore, nodeURL, clusterURL string) nexus_api.Ra
 	replStore := dkv_sync.NewDKVReplStore(kvs)
 	opts := []nexus.Option{
 		nexus.NodeUrl(nodeURL),
+		nexus.EntryDir(entDir),
 		nexus.LogDir(logDir),
 		nexus.SnapDir(snapDir),
 		nexus.ClusterUrl(clusterURL),

internal/sync/repl.go

@@ -24,23 +24,25 @@ func NewDKVReplStore(kvs storage.KVStore) db.Store {
 	return &dkvReplStore{kvs}
 }
 
-func (dr *dkvReplStore) Save(_ db.RaftEntry, req []byte) ([]byte, error) {
+func (dr *dkvReplStore) Save(_ db.RaftEntry, req []byte) (res []byte, err error) {
 	intReq := new(raftpb.InternalRaftRequest)
-	if err := proto.Unmarshal(req, intReq); err != nil {
+	if err = proto.Unmarshal(req, intReq); err != nil {
 		return nil, err
 	}
+
 	switch {
 	case intReq.Put != nil:
-		return dr.put(intReq.Put)
+		res, err = dr.put(intReq.Put)
 	case intReq.MultiPut != nil:
-		return dr.multiPut(intReq.MultiPut)
+		res, err = dr.multiPut(intReq.MultiPut)
 	case intReq.Delete != nil:
-		return dr.delete(intReq.Delete)
+		res, err = dr.delete(intReq.Delete)
 	case intReq.Cas != nil:
-		return dr.cas(intReq.Cas)
+		res, err = dr.cas(intReq.Cas)
 	default:
-		return nil, errors.New("Unknown Save request in dkv")
+		err = errors.New("Unknown Save request in dkv")
 	}
+	return
 }
 
 func (dr *dkvReplStore) Load(req []byte) ([]byte, error) {