[dbnode] Fix index flush recovery when previous flush attempts have failed

src/dbnode/persist/fs/files.go

@@ -115,12 +115,18 @@ func (f FileSetFile) IsZero() bool {
 	return len(f.AbsoluteFilepaths) == 0
 }
 
-// HasCheckpointFile returns a bool indicating whether the given set of
+// HasCompleteCheckpointFile returns a bool indicating whether the given set of
 // fileset files has a checkpoint file.
-func (f FileSetFile) HasCheckpointFile() bool {
+func (f FileSetFile) HasCompleteCheckpointFile() bool {
 	for _, fileName := range f.AbsoluteFilepaths {
 		if strings.Contains(fileName, checkpointFileSuffix) {
-			return true
+			exists, err := CompleteCheckpointFileExists(fileName)
+			if err != nil {
+				continue
+			}
+			if exists {
+				return true
+			}
 		}
 	}
 
@@ -161,7 +167,7 @@ func (f FileSetFilesSlice) LatestVolumeForBlock(blockStart time.Time) (FileSetFi
 					break
 				}
 
-				if curr.HasCheckpointFile() && curr.ID.VolumeIndex >= bestSoFar.ID.VolumeIndex {
+				if curr.HasCompleteCheckpointFile() && curr.ID.VolumeIndex >= bestSoFar.ID.VolumeIndex {
 					bestSoFar = curr
 					bestSoFarExists = true
 				}
@@ -812,7 +818,7 @@ func FileSetAt(filePathPrefix string, namespace ident.ID, shard uint32, blockSta
 				)
 			}
 
-			if !fileset.HasCheckpointFile() {
+			if !fileset.HasCompleteCheckpointFile() {
 				continue
 			}
 
@@ -841,7 +847,7 @@ func IndexFileSetsAt(filePathPrefix string, namespace ident.ID, blockStart time.
 	matches.sortByTimeAscending()
 	for _, fileset := range matches {
 		if fileset.ID.BlockStart.Equal(blockStart) {
-			if !fileset.HasCheckpointFile() {
+			if !fileset.HasCompleteCheckpointFile() {
 				continue
 			}
 			filesets = append(filesets, fileset)
@@ -1198,7 +1204,7 @@ func SnapshotFileSetExistsAt(prefix string, namespace ident.ID, shard uint32, bl
 		return false, nil
 	}
 
-	return latest.HasCheckpointFile(), nil
+	return latest.HasCompleteCheckpointFile(), nil
 }
 
 // NextSnapshotMetadataFileIndex returns the next snapshot metadata file index.

src/dbnode/persist/fs/files_test.go

@@ -737,7 +737,11 @@ func TestNextIndexFileSetVolumeIndex(t *testing.T) {
 		curr = index
 
 		p := filesetPathFromTimeAndIndex(dataDir, blockStart, index, checkpointFileSuffix)
-		err = ioutil.WriteFile(p, []byte("bar"), defaultNewFileMode)
+
+		digestBuf := digest.NewBuffer()
+		digestBuf.WriteDigest(digest.Checksum([]byte("bar")))
+
+		err = ioutil.WriteFile(p, digestBuf, defaultNewFileMode)
 		require.NoError(t, err)
 	}
 }
@@ -780,14 +784,35 @@ func TestMultipleForBlockStart(t *testing.T) {
 	require.Equal(t, numSnapshotsPerBlock-1, latestSnapshot.ID.VolumeIndex)
 }
 
-func TestSnapshotFileHasCheckPointFile(t *testing.T) {
+func TestSnapshotFileHasCompleteCheckpointFile(t *testing.T) {
+	dir := createTempDir(t)
+	defer os.RemoveAll(dir)
+
+	checkpointFilePath := path.Join(dir, "123-checkpoint-0.db")
+
+	// Test a valid complete checkpoint file
+	digestBuffer := digest.NewBuffer()
+	digestBuffer.WriteDigest(digest.Checksum([]byte{1, 2, 3}))
+	err := ioutil.WriteFile(checkpointFilePath, digestBuffer, defaultNewFileMode)
+	require.NoError(t, err)
+
+	// Check validates a valid checkpoint file
 	require.Equal(t, true, FileSetFile{
-		AbsoluteFilepaths: []string{"123-checkpoint-0.db"},
-	}.HasCheckpointFile())
+		AbsoluteFilepaths: []string{checkpointFilePath},
+	}.HasCompleteCheckpointFile())
 
+	// Check fails when checkpoint exists but not valid
+	err = ioutil.WriteFile(checkpointFilePath, []byte{42}, defaultNewFileMode)
+	require.NoError(t, err)
+	require.Equal(t, false, FileSetFile{
+		AbsoluteFilepaths: []string{checkpointFilePath},
+	}.HasCompleteCheckpointFile())
+
+	// Check ignores index file path
+	indexFilePath := path.Join(dir, "123-index-0.db")
 	require.Equal(t, false, FileSetFile{
-		AbsoluteFilepaths: []string{"123-index-0.db"},
-	}.HasCheckpointFile())
+		AbsoluteFilepaths: []string{indexFilePath},
+	}.HasCompleteCheckpointFile())
 }
 
 func TestSnapshotDirPath(t *testing.T) {
@@ -1107,7 +1132,16 @@ func createDataFiles(t *testing.T,
 		} else {
 			infoFilePath = filesetPathFromTime(shardDir, ts, fileSuffix)
 		}
-		createFile(t, infoFilePath, nil)
+		var contents []byte
+		if fileSuffix == checkpointFileSuffix {
+			// If writing a checkpoint file then write out a checksum of contents
+			// so that when code that validates the checkpoint file runs it returns
+			// successfully
+			digestBuf := digest.NewBuffer()
+			digestBuf.WriteDigest(digest.Checksum(contents))
+			contents = []byte(digestBuf)
+		}
+		createFile(t, infoFilePath, contents)
 	}
 	return dir
 }
@@ -1121,13 +1155,26 @@ type indexFileSetFileIdentifiers []indexFileSetFileIdentifier
 
 func (indexFilesets indexFileSetFileIdentifiers) create(t *testing.T, prefixDir string) {
 	for _, fileset := range indexFilesets {
-		fileSetFileIdentifiers{fileset.FileSetFileIdentifier}.create(t, prefixDir, persist.FileSetFlushType, fileset.Suffix)
+		idents := fileSetFileIdentifiers{fileset.FileSetFileIdentifier}
+		idents.create(t, prefixDir, persist.FileSetFlushType, fileset.Suffix)
 	}
 }
 
 type fileSetFileIdentifiers []FileSetFileIdentifier
 
 func (filesets fileSetFileIdentifiers) create(t *testing.T, prefixDir string, fileSetType persist.FileSetType, fileSuffixes ...string) {
+	writeFile := func(t *testing.T, path string, contents []byte) {
+		if strings.Contains(path, checkpointFileSuffix) {
+			// If writing a checkpoint file then write out a checksum of contents
+			// so that when code that validates the checkpoint file runs it returns
+			// successfully
+			digestBuf := digest.NewBuffer()
+			digestBuf.WriteDigest(digest.Checksum(contents))
+			contents = []byte(digestBuf)
+		}
+		createFile(t, path, contents)
+	}
+
 	for _, suffix := range fileSuffixes {
 		for _, fileset := range filesets {
 			switch fileset.FileSetContentType {
@@ -1141,10 +1188,10 @@ func (filesets fileSetFileIdentifiers) create(t *testing.T, prefixDir string, fi
 				switch fileSetType {
 				case persist.FileSetFlushType:
 					path = filesetPathFromTime(shardDir, blockStart, suffix)
-					createFile(t, path, nil)
+					writeFile(t, path, nil)
 				case persist.FileSetSnapshotType:
 					path = filesetPathFromTimeAndIndex(shardDir, blockStart, 0, fileSuffix)
-					createFile(t, path, nil)
+					writeFile(t, path, nil)
 				default:
 					panic("unknown FileSetType")
 				}
@@ -1158,7 +1205,7 @@ func (filesets fileSetFileIdentifiers) create(t *testing.T, prefixDir string, fi
 				switch fileSetType {
 				case persist.FileSetFlushType:
 					path = filesetPathFromTimeAndIndex(indexDir, blockStart, volumeIndex, suffix)
-					createFile(t, path, nil)
+					writeFile(t, path, nil)
 				case persist.FileSetSnapshotType:
 					fallthrough
 				default:

src/dbnode/persist/fs/index_write.go

@@ -125,9 +125,18 @@ func (w *indexWriter) Open(opts IndexWriterOpenOptions) error {
 	if err := os.MkdirAll(w.namespaceDir, w.newDirectoryMode); err != nil {
 		return err
 	}
-	w.checkpointFilePath = filesetPathFromTimeAndIndex(w.namespaceDir, blockStart, w.volumeIndex, checkpointFileSuffix)
 	w.infoFilePath = filesetPathFromTimeAndIndex(w.namespaceDir, blockStart, w.volumeIndex, infoFileSuffix)
 	w.digestFilePath = filesetPathFromTimeAndIndex(w.namespaceDir, blockStart, w.volumeIndex, digestFileSuffix)
+	w.checkpointFilePath = filesetPathFromTimeAndIndex(w.namespaceDir, blockStart, w.volumeIndex, checkpointFileSuffix)
+
+	exists, err := CompleteCheckpointFileExists(w.checkpointFilePath)
+	if err != nil {
+		return err
+	}
+	if exists {
+		return fmt.Errorf("checkpoint already exists for volume: %s",
+			w.checkpointFilePath)
+	}
 
 	return nil
 }
@@ -175,14 +184,6 @@ func (w *indexWriter) WriteSegmentFileSet(
 			err := fmt.Errorf("unknown fileset type: %s", w.fileSetType)
 			return w.markSegmentWriteError(segType, segFileType, err)
 		}
-		exists, err := FileExists(filePath)
-		if err != nil {
-			return err
-		}
-		if exists {
-			err := fmt.Errorf("segment file type already exists at %s", filePath)
-			return w.markSegmentWriteError(segType, segFileType, err)
-		}
 
 		fd, err := OpenWritable(filePath, w.newFileMode)
 		if err != nil {

src/dbnode/persist/fs/persist_manager_test.go

@@ -23,6 +23,7 @@ package fs
 import (
 	"errors"
 	"io"
+	"io/ioutil"
 	"os"
 	"testing"
 	"time"
@@ -33,9 +34,9 @@ import (
 	"github.com/m3db/m3/src/m3ninx/index/segment"
 	m3ninxfs "github.com/m3db/m3/src/m3ninx/index/segment/fst"
 	m3ninxpersist "github.com/m3db/m3/src/m3ninx/persist"
-	m3test "github.com/m3db/m3/src/x/test"
 	"github.com/m3db/m3/src/x/checked"
 	"github.com/m3db/m3/src/x/ident"
+	m3test "github.com/m3db/m3/src/x/test"
 	xtest "github.com/m3db/m3/src/x/test"
 
 	"github.com/golang/mock/gomock"
@@ -327,9 +328,12 @@ func TestPersistenceManagerPrepareIndexFileExists(t *testing.T) {
 	blockStart := time.Unix(1000, 0)
 	indexDir := createIndexDataDir(t, pm.filePathPrefix, testNs1ID)
 	checkpointFilePath := filesetPathFromTimeAndIndex(indexDir, blockStart, 0, checkpointFileSuffix)
-	f, err := os.Create(checkpointFilePath)
+
+	digestBuf := digest.NewBuffer()
+	digestBuf.WriteDigest(digest.Checksum([]byte("foo")))
+
+	err := ioutil.WriteFile(checkpointFilePath, digestBuf, defaultNewFileMode)
 	require.NoError(t, err)
-	f.Close()
 
 	flush, err := pm.StartIndexPersist()
 	require.NoError(t, err)