Added test for forward command

echovault/cluster_test.go

@@ -0,0 +1,124 @@
+// Copyright 2024 Kelvin Clement Mwinuka
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package echovault
+
+import (
+	"github.com/tidwall/resp"
+	"math"
+	"strings"
+	"testing"
+)
+
+func Test_raftApplyDeleteKey(t *testing.T) {
+	nodes, err := makeCluster(5)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	// Prepare the write data for the cluster
+	tests := []struct {
+		key   string
+		value string
+	}{
+		{
+			key:   "key1",
+			value: "value1",
+		},
+		{
+			key:   "key2",
+			value: "value2",
+		},
+		{
+			key:   "key3",
+			value: "value3",
+		},
+	}
+
+	// Write all the data to the cluster leader.
+	for i, test := range tests {
+		node := nodes[0]
+		if err := node.client.WriteArray([]resp.Value{
+			resp.StringValue("SET"),
+			resp.StringValue(test.key),
+			resp.StringValue(test.value),
+		}); err != nil {
+			t.Errorf("could not write data to leader node (test %d): %v", i, err)
+		}
+		// Read response and make sure we received "ok" response.
+		rd, _, err := node.client.ReadValue()
+		if err != nil {
+			t.Errorf("could not read response from leader node (test %d): %v", i, err)
+		}
+		if !strings.EqualFold(rd.String(), "ok") {
+			t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
+		}
+	}
+
+	// Check if the data has been replicated on a quorum (majority of the cluster).
+	quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
+	for i, test := range tests {
+		count := 0
+		for j := 0; j < len(nodes); j++ {
+			node := nodes[j]
+			if err := node.client.WriteArray([]resp.Value{
+				resp.StringValue("GET"),
+				resp.StringValue(test.key),
+			}); err != nil {
+				t.Errorf("could not write data to follower node %d (test %d): %v", j, i, err)
+			}
+			rd, _, err := node.client.ReadValue()
+			if err != nil {
+				t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
+			}
+			if rd.String() == test.value {
+				count += 1 // If the expected value is found, increment the count.
+			}
+		}
+		// Fail if count is less than quorum.
+		if count < quorum {
+			t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
+		}
+		// Delete key across raft cluster.
+		if err = nodes[0].server.raftApplyDeleteKey(nodes[0].server.context, test.key); err != nil {
+			t.Error(err)
+		}
+	}
+
+	// Check if the data is absent in quorum (majority of the cluster).
+	for i, test := range tests {
+		count := 0
+		for j := 0; j < len(nodes); j++ {
+			node := nodes[j]
+			if err := node.client.WriteArray([]resp.Value{
+				resp.StringValue("GET"),
+				resp.StringValue(test.key),
+			}); err != nil {
+				t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
+			}
+			rd, _, err := node.client.ReadValue()
+			if err != nil {
+				t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
+			}
+			if rd.IsNull() {
+				count += 1 // If the expected value is found, increment the count.
+			}
+		}
+		// Fail if count is less than quorum.
+		if count < quorum {
+			t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
+		}
+	}
+}

echovault/echovault_test.go

@@ -1,7 +1,22 @@
+// Copyright 2024 Kelvin Clement Mwinuka
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//      http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package echovault
 
 import (
 	"bufio"
+	"context"
 	"crypto/tls"
 	"crypto/x509"
 	"fmt"
@@ -15,6 +30,7 @@ import (
 	"strings"
 	"sync"
 	"testing"
+	"time"
 )
 
 type ClientServerPair struct {
@@ -62,6 +78,8 @@ func setupServer(
 	setupLock.Lock()
 	defer setupLock.Unlock()
 
+	ctx := context.Background()
+
 	config := DefaultConfig()
 	config.DataDir = "./testdata"
 	config.ForwardCommand = true
@@ -73,7 +91,11 @@ func setupServer(
 	config.RaftBindPort = uint16(raftPort)
 	config.MemberListBindPort = uint16(mlPort)
 	config.BootstrapCluster = bootstrapCluster
-	return NewEchoVault(WithConfig(config))
+
+	return NewEchoVault(
+		WithContext(ctx),
+		WithConfig(config),
+	)
 }
 
 func makeCluster(size int) ([]ClientServerPair, error) {
@@ -230,6 +252,200 @@ func Test_ClusterReplication(t *testing.T) {
 	}
 }
 
+func Test_ClusterDeleteKey(t *testing.T) {
+	nodes, err := makeCluster(5)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	// Prepare the write data for the cluster
+	tests := []struct {
+		key   string
+		value string
+	}{
+		{
+			key:   "key1",
+			value: "value1",
+		},
+		{
+			key:   "key2",
+			value: "value2",
+		},
+		{
+			key:   "key3",
+			value: "value3",
+		},
+	}
+
+	// Write all the data to the cluster leader
+	for i, test := range tests {
+		node := nodes[0]
+		if err := node.client.WriteArray([]resp.Value{
+			resp.StringValue("SET"),
+			resp.StringValue(test.key),
+			resp.StringValue(test.value),
+		}); err != nil {
+			t.Errorf("could not write command to leader node (test %d): %v", i, err)
+		}
+		// Read response and make sure we received "ok" response.
+		rd, _, err := node.client.ReadValue()
+		if err != nil {
+			t.Errorf("could not read response from leader node (test %d): %v", i, err)
+		}
+		if !strings.EqualFold(rd.String(), "ok") {
+			t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
+		}
+	}
+
+	quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
+
+	// Check if the data has been replicated on a quorum (majority of the cluster).
+	for i, test := range tests {
+		count := 0
+		for j := 0; j < len(nodes); j++ {
+			node := nodes[j]
+			if err := node.client.WriteArray([]resp.Value{
+				resp.StringValue("GET"),
+				resp.StringValue(test.key),
+			}); err != nil {
+				t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
+			}
+			rd, _, err := node.client.ReadValue()
+			if err != nil {
+				t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
+			}
+			if rd.String() == test.value {
+				count += 1 // If the expected value is found, increment the count.
+			}
+		}
+		// Fail if count is less than quorum.
+		if count < quorum {
+			t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
+			return
+		}
+	}
+
+	// Delete the key on the leader node
+	for i, test := range tests {
+		node := nodes[0]
+		if err := node.client.WriteArray([]resp.Value{
+			resp.StringValue("DEL"),
+			resp.StringValue(test.key),
+		}); err != nil {
+			t.Errorf("could not write command to leader node (test %d): %v", i, err)
+		}
+		// Read response and make sure we received "ok" response.
+		rd, _, err := node.client.ReadValue()
+		if err != nil {
+			t.Errorf("could not read response from leader node (test %d): %v", i, err)
+		}
+		if rd.Integer() != 1 {
+			t.Errorf("expected response for test %d to be 1, got %d", i, rd.Integer())
+		}
+	}
+
+	// Check if the data is absent in quorum (majority of the cluster).
+	for i, test := range tests {
+		count := 0
+		for j := 0; j < len(nodes); j++ {
+			node := nodes[j]
+			if err := node.client.WriteArray([]resp.Value{
+				resp.StringValue("GET"),
+				resp.StringValue(test.key),
+			}); err != nil {
+				t.Errorf("could not write command to follower node %d (test %d): %v", j, i, err)
+			}
+			rd, _, err := node.client.ReadValue()
+			if err != nil {
+				t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
+			}
+			if rd.IsNull() {
+				count += 1 // If the expected value is found, increment the count.
+			}
+		}
+		// Fail if count is less than quorum.
+		if count < quorum {
+			t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
+		}
+	}
+}
+
+func Test_CommandForwarded(t *testing.T) {
+	nodes, err := makeCluster(5)
+	if err != nil {
+		t.Error(err)
+		return
+	}
+
+	// Prepare the write data for the cluster
+	tests := []struct {
+		key   string
+		value string
+	}{
+		{
+			key:   "key1",
+			value: "value1",
+		},
+		{
+			key:   "key2",
+			value: "value2",
+		},
+		{
+			key:   "key3",
+			value: "value3",
+		},
+	}
+
+	// Write all the data a random cluster follower.
+	for i, test := range tests {
+		// Send write command to follower node.
+		node := nodes[1]
+		if err := node.client.WriteArray([]resp.Value{
+			resp.StringValue("SET"),
+			resp.StringValue(test.key),
+			resp.StringValue(test.value),
+		}); err != nil {
+			t.Errorf("could not write data to leader node (test %d): %v", i, err)
+		}
+		// Read response and make sure we received "ok" response.
+		rd, _, err := node.client.ReadValue()
+		if err != nil {
+			t.Errorf("could not read response from leader node (test %d): %v", i, err)
+		}
+		if !strings.EqualFold(rd.String(), "ok") {
+			t.Errorf("expected response for test %d to be \"OK\", got %s", i, rd.String())
+		}
+	}
+	<-time.After(250 * time.Millisecond) // Short yield to allow change to take effect.
+
+	// Check if the data has been replicated on a quorum (majority of the cluster).
+	quorum := int(math.Ceil(float64(len(nodes)/2)) + 1)
+	for i, test := range tests {
+		count := 0
+		for j := 0; j < len(nodes); j++ {
+			node := nodes[j]
+			if err := node.client.WriteArray([]resp.Value{
+				resp.StringValue("GET"),
+				resp.StringValue(test.key),
+			}); err != nil {
+				t.Errorf("could not write data to follower node %d (test %d): %v", j, i, err)
+			}
+			rd, _, err := node.client.ReadValue()
+			if err != nil {
+				t.Errorf("could not read data from follower node %d (test %d): %v", j, i, err)
+			}
+			if rd.String() == test.value {
+				count += 1 // If the expected value is found, increment the count.
+			}
+		}
+		// Fail if count is less than quorum.
+		if count < quorum {
+			t.Errorf("could not find value %s at key %s in cluster quorum", test.value, test.key)
+		}
+	}
+}
+
 func Test_TLS(t *testing.T) {
 	port, err := internal.GetFreePort()
 	if err != nil {

internal/memberlist/delegate.go

@@ -66,9 +66,8 @@ func (delegate *Delegate) NodeMeta(limit int) []byte {
 // NotifyMsg implements Delegate interface
 func (delegate *Delegate) NotifyMsg(msgBytes []byte) {
 	var msg BroadcastMessage
-
 	if err := json.Unmarshal(msgBytes, &msg); err != nil {
-		fmt.Print(err)
+		log.Printf("notifymsg: %v", err)
 		return
 	}