Fix inconsistent state between WAL and saved Snapshot

orderer/consensus/etcdraft/storage.go

@@ -75,22 +75,9 @@ func CreateStorage(
 		return nil, err
 	}
 
-	snapshot, err := sn.Load()
+	snapshot, w, st, ents, err := loadNewestAvailableSnapshot(lg, walDir, snapDir)
 	if err != nil {
-		if err == snap.ErrNoSnapshot {
-			lg.Debugf("No snapshot found at %s", snapDir)
-		} else {
-			return nil, errors.Errorf("failed to load snapshot: %s", err)
-		}
-	} else {
-		// snapshot found
-		lg.Debugf("Loaded snapshot at Term %d and Index %d, Nodes: %+v",
-			snapshot.Metadata.Term, snapshot.Metadata.Index, snapshot.Metadata.ConfState.Nodes)
-	}
-
-	w, st, ents, err := createOrReadWAL(lg, walDir, snapshot)
-	if err != nil {
-		return nil, errors.Errorf("failed to create or read WAL: %s", err)
+		return nil, errors.Errorf("Failed to load snapshot and WAL: %s", err)
 	}
 
 	if snapshot != nil {
@@ -120,26 +107,11 @@ func CreateStorage(
 // ListSnapshots returns a list of RaftIndex of snapshots stored on disk.
 // If a file is corrupted, rename the file.
 func ListSnapshots(logger *flogging.FabricLogger, snapDir string) []uint64 {
-	dir, err := os.Open(snapDir)
+	snapfiles, err := listSnapshotFiles(logger, snapDir)
 	if err != nil {
-		logger.Errorf("Failed to open snapshot directory %s: %s", snapDir, err)
+		logger.Errorf("Failed to list snapshot files from %s: %s", snapDir, err)
 		return nil
 	}
-	defer dir.Close()
-
-	filenames, err := dir.Readdirnames(-1)
-	if err != nil {
-		logger.Errorf("Failed to read snapshot files: %s", err)
-		return nil
-	}
-
-	snapfiles := []string{}
-	for i := range filenames {
-		if strings.HasSuffix(filenames[i], ".snap") {
-			snapfiles = append(snapfiles, filenames[i])
-		}
-	}
-	sort.Strings(snapfiles)
 
 	var snapshots []uint64
 	for _, snapfile := range snapfiles {
@@ -242,15 +214,17 @@ func (rs *RaftStorage) Snapshot() raftpb.Snapshot {
 
 // Store persists etcd/raft data
 func (rs *RaftStorage) Store(entries []raftpb.Entry, hardstate raftpb.HardState, snapshot raftpb.Snapshot) error {
-	if err := rs.wal.Save(hardstate, entries); err != nil {
-		return err
-	}
-
 	if !raft.IsEmptySnap(snapshot) {
 		if err := rs.saveSnap(snapshot); err != nil {
 			return err
 		}
+	}
+
+	if err := rs.wal.Save(hardstate, entries); err != nil {
+		return err
+	}
 
+	if !raft.IsEmptySnap(snapshot) {
 		if err := rs.ram.ApplySnapshot(snapshot); err != nil {
 			if err == raft.ErrSnapOutOfDate {
 				rs.lg.Warnf("Attempted to apply out-of-date snapshot at Term %d and Index %d",
@@ -447,3 +421,54 @@ func (rs *RaftStorage) Close() error {
 
 	return nil
 }
+
+func loadNewestAvailableSnapshot(lg *flogging.FabricLogger, walDir, snapDir string) (*raftpb.Snapshot, *wal.WAL, raftpb.HardState, []raftpb.Entry, error) {
+	snapfiles, err := listSnapshotFiles(lg, snapDir)
+	if err != nil {
+		lg.Errorf("Failed to list snapshot files from %s: %s", snapDir, err)
+	}
+	for i := len(snapfiles) - 1; i >= 0; i-- {
+		snapshot, err := snap.Read(lg.Zap(), filepath.Join(snapDir, snapfiles[i]))
+		if err != nil {
+			lg.Warnf("Can not read snapshot from %s: %s", snapfiles[i], err)
+			continue
+		}
+		w, st, ents, err := createOrReadWAL(lg, walDir, snapshot)
+		if err != nil {
+			lg.Warnf("Create or read wal error: %s", err)
+			continue
+		}
+		if snapshot.Metadata.Index <= st.Commit {
+			return snapshot, w, st, ents, nil
+		}
+		if err := w.Close(); err != nil {
+			return nil, nil, raftpb.HardState{}, nil, err
+		}
+	}
+	lg.Warnf("Not available snapshot found in %s", snapDir)
+	w, st, ents, err := createOrReadWAL(lg, walDir, nil)
+	return nil, w, st, ents, err
+}
+
+func listSnapshotFiles(logging *flogging.FabricLogger, snapDir string) ([]string, error) {
+	dir, err := os.Open(snapDir)
+	if err != nil {
+		return nil, err
+	}
+	defer dir.Close()
+
+	filenames, err := dir.Readdirnames(-1)
+	if err != nil {
+		return nil, err
+	}
+
+	snapfiles := []string{}
+	for i := range filenames {
+		if strings.HasSuffix(filenames[i], ".snap") {
+			snapfiles = append(snapfiles, filenames[i])
+		}
+	}
+	sort.Strings(snapfiles)
+
+	return snapfiles, nil
+}

orderer/consensus/etcdraft/storage_test.go

@@ -80,7 +80,7 @@ func TestOpenWAL(t *testing.T) {
 		for i := 0; i < 10; i++ {
 			store.Store(
 				[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 10)}},
-				raftpb.HardState{},
+				raftpb.HardState{Commit: uint64(i)},
 				raftpb.Snapshot{},
 			)
 		}
@@ -155,7 +155,7 @@ func TestTakeSnapshot(t *testing.T) {
 			for i := 0; i < 10; i++ {
 				store.Store(
 					[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 100)}},
-					raftpb.HardState{},
+					raftpb.HardState{Commit: uint64(i)},
 					raftpb.Snapshot{},
 				)
 			}
@@ -216,7 +216,7 @@ func TestTakeSnapshot(t *testing.T) {
 			for i := 0; i < 10; i++ {
 				store.Store(
 					[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 100)}},
-					raftpb.HardState{},
+					raftpb.HardState{Commit: uint64(i)},
 					raftpb.Snapshot{},
 				)
 			}
@@ -282,7 +282,7 @@ func TestTakeSnapshot(t *testing.T) {
 			for i := 0; i < 10; i++ {
 				store.Store(
 					[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 100)}},
-					raftpb.HardState{},
+					raftpb.HardState{Commit: uint64(i)},
 					raftpb.Snapshot{},
 				)
 			}
@@ -369,7 +369,7 @@ func TestApplyOutOfDateSnapshot(t *testing.T) {
 		for i := 0; i < 10; i++ {
 			store.Store(
 				[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 100)}},
-				raftpb.HardState{},
+				raftpb.HardState{Commit: uint64(i)},
 				raftpb.Snapshot{},
 			)
 		}
@@ -395,3 +395,82 @@ func TestApplyOutOfDateSnapshot(t *testing.T) {
 		assertFileCount(t, 12, 1)
 	})
 }
+
+func TestAbortWhenWritingSnapshot(t *testing.T) {
+	t.Run("Abort when writing snapshot", func(t *testing.T) {
+		setup(t)
+		defer clean(t)
+
+		// set SegmentSizeBytes to a small value so that
+		// every entry persisted to wal would result in
+		// a new wal being created.
+		oldSegmentSizeBytes := wal.SegmentSizeBytes
+		wal.SegmentSizeBytes = 10
+		defer func() {
+			wal.SegmentSizeBytes = oldSegmentSizeBytes
+		}()
+
+		// create 5 new entry
+		for i := 0; i < 5; i++ {
+			store.Store(
+				[]raftpb.Entry{{Index: uint64(i), Data: make([]byte, 100)}},
+				raftpb.HardState{Commit: uint64(i)},
+				raftpb.Snapshot{},
+			)
+		}
+		assertFileCount(t, 6, 0)
+
+		// Assume an orderer missed some records due to exceptions and receives a new snapshot from other orderers.
+		commit := 10
+		store.Store(
+			[]raftpb.Entry{},
+			raftpb.HardState{Commit: uint64(commit)},
+			raftpb.Snapshot{
+				Metadata: raftpb.SnapshotMetadata{
+					Index: uint64(commit),
+				},
+				Data: make([]byte, 100),
+			},
+		)
+		err = store.Close()
+		assert.NoError(t, err)
+
+		// In old logic, it will use rs.wal.Save(hardstate, entries) to save the state firstly, so we remove the snapshot files.
+		// sd, err := os.Open(snapDir)
+		// assert.NoError(t, err)
+		// defer sd.Close()
+		// names, err := sd.Readdirnames(-1)
+		// assert.NoError(t, err)
+		// sort.Sort(sort.Reverse(sort.StringSlice(names)))
+		// os.Remove(filepath.Join(snapDir, names[0]))
+		// wd, err := os.Open(walDir)
+		// assert.NoError(t, err)
+		// defer wd.Close()
+		// names, err = wd.Readdirnames(-1)
+		// assert.NoError(t, err)
+		// sort.Sort(sort.Reverse(sort.StringSlice(names)))
+		// os.Remove(filepath.Join(walDir, names[0]))
+
+		// But in the new logic, it will use rs.saveSnap(snapshot) to save the snapshot firstly, so we remove the WAL files.
+		wd, err := os.Open(walDir)
+		assert.NoError(t, err)
+		defer wd.Close()
+		names, err := wd.Readdirnames(-1)
+		assert.NoError(t, err)
+		sort.Sort(sort.Reverse(sort.StringSlice(names)))
+		os.Remove(filepath.Join(walDir, names[0]))
+
+		// Then restart the orderer.
+		ram := raft.NewMemoryStorage()
+		store, err = CreateStorage(logger, walDir, snapDir, ram)
+		assert.NoError(t, err)
+
+		// Check the state from go.etcd.io/etcd/raft/raft.go
+		// func (r *raft) loadState(state pb.HardState)
+		hd, _, err := store.ram.InitialState()
+		assert.NoError(t, err)
+		lastIndex, err := store.ram.LastIndex()
+		assert.NoError(t, err)
+		assert.False(t, hd.Commit > lastIndex)
+	})
+}