Merge pull request #47 from EchoVault/chore/echovault-test

Chore/echovault test

echovault/echovault.go

@@ -276,6 +276,9 @@ func NewEchoVault(options ...func(echovault *EchoVault)) (*EchoVault, error) {
 	if echovault.config.EvictionPolicy != constants.NoEviction {
 		go func() {
 			ticker := time.NewTicker(echovault.config.EvictionInterval)
+			defer func() {
+				ticker.Stop()
+			}()
 			for {
 				select {
 				case <-ticker.C:
@@ -549,11 +552,9 @@ func (server *EchoVault) rewriteAOF() error {
 	if server.rewriteAOFInProgress.Load() {
 		return errors.New("aof rewrite in progress")
 	}
-	go func() {
-		if err := server.aofEngine.RewriteLog(); err != nil {
-			log.Println(err)
-		}
-	}()
+	if err := server.aofEngine.RewriteLog(); err != nil {
+		return err
+	}
 	return nil
 }
 

echovault/echovault_test.go

@@ -894,61 +894,7 @@ func Test_Standalone(t *testing.T) {
 			wantLastSave int
 		}{
 			{
-				name:    "1. Snapshot with TCP connection",
-				dataDir: path.Join(dataDir, "with_tcp_connection"),
-				values: map[string]string{
-					"key1": "value1",
-					"key2": "value2",
-					"key3": "value3",
-					"key4": "value4",
-				},
-				snapshotFunc: func(mockServer *EchoVault) error {
-					// Start the server's TCP listener
-					go func() {
-						mockServer.Start()
-					}()
-					conn, err := internal.GetConnection("localhost", int(mockServer.config.Port))
-					if err != nil {
-						return err
-					}
-					defer func() {
-						_ = conn.Close()
-					}()
-					client := resp.NewConn(conn)
-					if err = client.WriteArray([]resp.Value{resp.StringValue("SAVE")}); err != nil {
-						return err
-					}
-					res, _, err := client.ReadValue()
-					if err != nil {
-						return err
-					}
-					if !strings.EqualFold(res.String(), "ok") {
-						return fmt.Errorf("expected save response to be \"OK\", got \"%s\"", res.String())
-					}
-					return nil
-				},
-				lastSaveFunc: func(mockServer *EchoVault) (int, error) {
-					conn, err := internal.GetConnection("localhost", int(mockServer.config.Port))
-					if err != nil {
-						return 0, err
-					}
-					defer func() {
-						_ = conn.Close()
-					}()
-					client := resp.NewConn(conn)
-					if err = client.WriteArray([]resp.Value{resp.StringValue("LASTSAVE")}); err != nil {
-						return 0, err
-					}
-					res, _, err := client.ReadValue()
-					if err != nil {
-						return 0, err
-					}
-					return res.Integer(), nil
-				},
-				wantLastSave: int(clock.NewClock().Now().UnixMilli()),
-			},
-			{
-				name:    "2. Snapshot in embedded instance",
+				name:    "1. Snapshot in embedded instance",
 				dataDir: path.Join(dataDir, "embedded_instance"),
 				values: map[string]string{
 					"key5": "value5",
@@ -1047,8 +993,101 @@ func Test_Standalone(t *testing.T) {
 		}
 	})
 
-	t.Run("Test_AOF", func(t *testing.T) {
+	t.Run("Test_AOFRestore", func(t *testing.T) {
 		t.Parallel()
-		// TODO: Implemented AOF persistence and restore.
+
+		ticker := time.NewTicker(50 * time.Millisecond)
+
+		dataDir := path.Join(".", "testdata", "test_aof")
+		t.Cleanup(func() {
+			_ = os.RemoveAll(dataDir)
+			ticker.Stop()
+		})
+
+		// Prepare data for testing.
+		data := map[string]map[string]string{
+			"before-rewrite": {
+				"key1": "value1",
+				"key2": "value2",
+				"key3": "value3",
+				"key4": "value4",
+			},
+			"after-rewrite": {
+				"key3": "value3-updated",
+				"key4": "value4-updated",
+				"key5": "value5",
+				"key6": "value6",
+			},
+			"expected-values": {
+				"key1": "value1",
+				"key2": "value2",
+				"key3": "value3-updated",
+				"key4": "value4-updated",
+				"key5": "value5",
+				"key6": "value6",
+			},
+		}
+
+		conf := DefaultConfig()
+		conf.RestoreAOF = true
+		conf.DataDir = dataDir
+		conf.AOFSyncStrategy = "always"
+
+		mockServer, err := NewEchoVault(WithConfig(conf))
+		if err != nil {
+			t.Error(err)
+			return
+		}
+
+		// Perform write commands from "before-rewrite"
+		for key, value := range data["before-rewrite"] {
+			if _, _, err := mockServer.Set(key, value, SetOptions{}); err != nil {
+				t.Error(err)
+				return
+			}
+		}
+
+		// Yield
+		<-ticker.C
+
+		// Rewrite AOF
+		if _, err := mockServer.RewriteAOF(); err != nil {
+			t.Error(err)
+			return
+		}
+
+		// Perform write commands from "after-rewrite"
+		for key, value := range data["after-rewrite"] {
+			if _, _, err := mockServer.Set(key, value, SetOptions{}); err != nil {
+				t.Error(err)
+				return
+			}
+		}
+
+		// Yield
+		<-ticker.C
+
+		// Shutdown the EchoVault instance
+		mockServer.ShutDown()
+
+		// Start another instance of EchoVault
+		mockServer, err = NewEchoVault(WithConfig(conf))
+		if err != nil {
+			t.Error(err)
+			return
+		}
+
+		// Check if the servers contains the keys and values from "expected-values"
+		for key, value := range data["expected-values"] {
+			res, err := mockServer.Get(key)
+			if err != nil {
+				t.Error(err)
+				return
+			}
+			if res != value {
+				t.Errorf("expected value at key \"%s\" to be \"%s\", got \"%s\"", key, value, res)
+				return
+			}
+		}
 	})
 }

internal/aof/engine.go

@@ -179,7 +179,7 @@ func (engine *Engine) RewriteLog() error {
 	engine.startRewriteFunc()
 	defer engine.finishRewriteFunc()
 
-	// Create AOF preamble
+	// Create AOF preamble.
 	if err := engine.preambleStore.CreatePreamble(); err != nil {
 		return fmt.Errorf("rewrite log -> create preamble error: %+v", err)
 	}

internal/aof/engine_test.go

@@ -29,7 +29,7 @@ import (
 
 func marshalRespCommand(command []string) []byte {
 	return []byte(fmt.Sprintf(
-		"*%d\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n$%d\r\n%s", len(command),
+		"*%d\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n", len(command),
 		len(command[0]), command[0],
 		len(command[1]), command[1],
 		len(command[2]), command[2],
@@ -151,11 +151,11 @@ func Test_AOFEngine(t *testing.T) {
 	}
 
 	if len(wantRestoredState) != len(restoredState) {
-		t.Errorf("expected restored state to be lenght %d, got %d", len(wantRestoredState), len(restoredState))
+		t.Errorf("expected restored state to be length %d, got %d", len(wantRestoredState), len(restoredState))
 		for key, data := range restoredState {
 			want, ok := wantRestoredState[key]
 			if !ok {
-				t.Errorf("could not find key %s in expected state state", key)
+				t.Errorf("could not find key %s in expected state", key)
 			}
 			if want.Value != data.Value {
 				t.Errorf("expected value %v for key %s, got %v", want.Value, key, data.Value)

internal/aof/log/store.go

@@ -105,6 +105,10 @@ func NewAppendStore(options ...func(store *AppendStore)) (*AppendStore, error) {
 	// No need to start this goroutine if sync strategy is anything other than 'everysec'.
 	if strings.EqualFold(store.strategy, "everysec") {
 		go func() {
+			ticker := time.NewTicker(1 * time.Second)
+			defer func() {
+				ticker.Stop()
+			}()
 			for {
 				store.mut.Lock()
 				if err := store.Sync(); err != nil {
@@ -113,7 +117,7 @@ func NewAppendStore(options ...func(store *AppendStore)) (*AppendStore, error) {
 					break
 				}
 				store.mut.Unlock()
-				<-store.clock.After(1 * time.Second)
+				<-ticker.C
 			}
 		}()
 	}
@@ -130,9 +134,7 @@ func (store *AppendStore) Write(command []byte) error {
 	store.mut.Lock()
 	defer store.mut.Unlock()
 
-	// Add new line before writing to AOF file.
-	out := append(command, []byte("\r\n")...)
-	if _, err := store.rw.Write(out); err != nil {
+	if _, err := store.rw.Write(command); err != nil {
 		return err
 	}
 

internal/aof/log/store_test.go

@@ -27,7 +27,7 @@ import (
 
 func marshalRespCommand(command []string) []byte {
 	return []byte(fmt.Sprintf(
-		"*%d\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n$%d\r\n%s", len(command),
+		"*%d\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n$%d\r\n%s\r\n", len(command),
 		len(command[0]), command[0],
 		len(command[1]), command[1],
 		len(command[2]), command[2],
@@ -138,9 +138,6 @@ func Test_AppendStore(t *testing.T) {
 		}()
 
 		ticker := time.NewTicker(200 * time.Millisecond)
-		defer func() {
-			ticker.Stop()
-		}()
 
 		select {
 		case <-done: