fix(store): unexpected compaction after restart (#603)

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

Author: ifplusor

internal/store/raft/storage/compaction.go

@@ -255,23 +255,14 @@ func (w *WAL) removeNode(ctx context.Context, nodeID vanus.ID) error {
 }
 
 func (w *WAL) recoverNode(nodeID vanus.ID, offset int64) {
-	task := adminTask{
-		nodeID: nodeID,
-		offset: offset,
-	}
-
-	w.closeMu.RLock()
-	select {
-	case <-w.closeC:
-	default:
-		w.compactC <- task.recoverNode
+	w.nodes[nodeID] = true
+	if offset != 0 {
+		w.barrier.Set(offset, nodeID)
 	}
-	w.closeMu.RUnlock()
 }
 
 type adminTask struct {
 	nodeID vanus.ID
-	offset int64
 	ch     chan<- error
 }
 
@@ -309,13 +300,6 @@ func (t *adminTask) removeNode(w *WAL, cCtx *compactContext) {
 	})
 }
 
-func (t *adminTask) recoverNode(w *WAL, _ *compactContext) {
-	w.nodes[t.nodeID] = true
-	if t.offset != 0 {
-		w.barrier.Set(t.offset, t.nodeID)
-	}
-}
-
 type compactInfo struct {
 	index, term uint64
 }

internal/store/raft/storage/compaction_test.go

@@ -53,7 +53,7 @@ func TestStorage_Compact(t *testing.T) {
 
 		rawWAL, err := walog.Open(ctx, walDir, walog.WithFileSize(int64(fileSize)))
 		So(err, ShouldBeNil)
-		wal := newWAL(rawWAL, stateStore)
+		wal := newWAL(rawWAL, stateStore, true)
 
 		s1, _ := NewStorage(ctx, nodeID1, wal, stateStore, hintStore, nil)
 		s2, _ := NewStorage(ctx, nodeID2, wal, stateStore, hintStore, nil)

internal/store/raft/storage/recovery.go

@@ -123,7 +123,7 @@ func Recover(
 	if err != nil {
 		return nil, nil, err
 	}
-	wal2 := newWAL(wal, stateStore)
+	wal2 := newWAL(wal, stateStore, false)
 
 	// convert, and set wal
 	storages := make(map[vanus.ID]*Storage, len(sb.storages))
@@ -140,6 +140,9 @@ func Recover(
 		storages[nodeID] = storage
 	}
 
+	// Start compaction after recover nodes.
+	wal2.startCompaction()
+
 	return storages, wal2, nil
 }
 

internal/store/raft/storage/snapshot_storage_test.go

@@ -61,7 +61,7 @@ func TestStorage_SnapshotStorage(t *testing.T) {
 
 		rawWAL, err := walog.Open(ctx, walDir, walog.WithFileSize(int64(fileSize)))
 		So(err, ShouldBeNil)
-		wal := newWAL(rawWAL, stateStore)
+		wal := newWAL(rawWAL, stateStore, true)
 		defer wal.Close()
 
 		snapOp := NewMockSnapshotOperator(ctrl)

internal/store/raft/storage/storage_test.go

@@ -86,7 +86,7 @@ func TestStorage(t *testing.T) {
 		Convey("create raft log", func() {
 			rawWAL, err := walog.Open(ctx, raftDir, walog.WithFileSize(int64(fileSize)))
 			So(err, ShouldBeNil)
-			wal := newWAL(rawWAL, stateStore)
+			wal := newWAL(rawWAL, stateStore, true)
 			defer func() {
 				wal.Close()
 				wal.Wait()

internal/store/raft/storage/wal.go

@@ -45,7 +45,7 @@ type WAL struct {
 	doneC   chan struct{}
 }
 
-func newWAL(wal *walog.WAL, stateStore *meta.SyncStore) *WAL {
+func newWAL(wal *walog.WAL, stateStore *meta.SyncStore, startCompaction bool) *WAL {
 	w := &WAL{
 		WAL:        wal,
 		stateStore: stateStore,
@@ -56,11 +56,17 @@ func newWAL(wal *walog.WAL, stateStore *meta.SyncStore) *WAL {
 		doneC:      make(chan struct{}),
 	}
 
-	go w.runCompact()
+	if startCompaction {
+		w.startCompaction()
+	}
 
 	return w
 }
 
+func (w *WAL) startCompaction() {
+	go w.runCompact()
+}
+
 func (w *WAL) Close() {
 	w.WAL.Close()
 	go func() {