refactor: storage (#634)

* feat(store): save copy of bytes in meta store
* fix: return false when push task to MultiFlow fails
* refactor(store): raft storage

Signed-off-by: James Yin <[email protected]>

lib/container/conque/blocking/queue.go

@@ -45,7 +45,7 @@ func (q *Queue[T]) Close() {
 	q.sem.Release()
 }
 
-// Wait esures that all incoming Pushes observe that the queue is closed.
+// Wait ensures that all incoming Pushes observe that the queue is closed.
 func (q *Queue[T]) Wait() {
 	// Make sure no inflight Push.
 	q.mu.Lock()

lib/executor/multi_flow.go

@@ -40,7 +40,7 @@ func (f *flow) Execute(t Task) bool {
 	}
 
 	if f.q.Push(t) {
-		f.mf.q.Push(f)
+		return f.mf.q.Push(f)
 	}
 	return true
 }

server/store/meta/async.go

@@ -111,7 +111,7 @@ func (s *AsyncStore) set(kvs Ranger) error {
 	}
 
 	err := kvs.Range(func(key []byte, value interface{}) error {
-		s.pending.Set(key, value)
+		set(s.pending, key, value)
 		return nil
 	})
 	if err != nil {
@@ -190,12 +190,6 @@ func (s *AsyncStore) commit() {
 	s.version = r.EO
 }
 
-func merge(dst, src *skiplist.SkipList) {
-	for el := src.Front(); el != nil; el = el.Next() {
-		set(dst, el.Key().([]byte), el.Value)
-	}
-}
-
 func RecoverAsyncStore(ctx context.Context, dir string, opts ...walog.Option) (*AsyncStore, error) {
 	committed, snapshot, err := recoverLatestSnapshot(ctx, dir, defaultCodec)
 	if err != nil {

server/store/meta/store.go

@@ -51,11 +51,3 @@ func (s *store) load(key []byte) (interface{}, bool) {
 // func (s *store) delete(key []byte) {
 // 	set(s.committed, key, deletedMark)
 // }
-
-func set(m *skiplist.SkipList, key []byte, value interface{}) {
-	if value == DeletedMark {
-		m.Remove(key)
-	} else {
-		m.Set(key, value)
-	}
-}

server/store/meta/sync.go

@@ -126,11 +126,7 @@ func (s *SyncStore) set(ctx context.Context, kvs Ranger, cb StoreCallback) {
 		// Update state.
 		s.mu.Lock()
 		_ = kvs.Range(func(key []byte, value interface{}) error {
-			if value == DeletedMark {
-				s.committed.Remove(key)
-			} else {
-				s.committed.Set(key, value)
-			}
+			update(s.committed, key, value)
 			return nil
 		})
 		s.version = r.EO
@@ -174,7 +170,7 @@ func RecoverSyncStore(ctx context.Context, dir string, opts ...walog.Option) (*S
 		walog.FromPosition(snapshot),
 		walog.WithRecoveryCallback(func(data []byte, r walog.Range) error {
 			err2 := defaultCodec.Unmarshal(data, func(key []byte, value interface{}) error {
-				set(committed, key, value)
+				rawUpdate(committed, key, value)
 				return nil
 			})
 			if err2 != nil {

server/store/meta/utils.go

@@ -0,0 +1,54 @@
+// Copyright 2022 Linkall Inc.
+//
+// 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 meta
+
+import (
+	// third-party libraries.
+	"github.com/huandu/skiplist"
+)
+
+func update(m *skiplist.SkipList, key []byte, value interface{}) {
+	if value == DeletedMark {
+		m.Remove(key)
+		return
+	}
+
+	set(m, key, value)
+}
+
+func set(m *skiplist.SkipList, key []byte, value interface{}) {
+	switch val := value.(type) {
+	case []byte:
+		// Make a copy to avoid modifying value outside.
+		bs := append([]byte{}, val...)
+		m.Set(key, bs)
+	default:
+		m.Set(key, value)
+	}
+}
+
+func rawUpdate(m *skiplist.SkipList, key []byte, value interface{}) {
+	if value == DeletedMark {
+		m.Remove(key)
+	} else {
+		m.Set(key, value)
+	}
+}
+
+func merge(dst, src *skiplist.SkipList) {
+	for el := src.Front(); el != nil; el = el.Next() {
+		rawUpdate(dst, el.Key().([]byte), el.Value)
+	}
+}

server/store/raft/block/engine.go

@@ -24,6 +24,7 @@ import (
 	// third-party libraries.
 	"google.golang.org/grpc"
 
+	// first-party libraries.
 	raftpb "github.com/vanus-labs/vanus/api/raft"
 	vanus "github.com/vanus-labs/vanus/api/vsr"
 	"github.com/vanus-labs/vanus/pkg/raft"