Merge pull request #1011 from nats-io/fix_filestore_empty

[FIXED] FileStore: Empty() should remove all message files (dat/idx)

stores/filestore.go

@@ -3831,24 +3831,23 @@ func (ms *FileMsgStore) Empty() error {
 	defer ms.Unlock()
 
 	var err error
-	// Remove/close all file slices
+	// Close all file slices
 	for sliceID, slice := range ms.files {
 		ms.fm.remove(slice.file)
 		ms.fm.remove(slice.idxFile)
 		if slice.file.handle != nil {
 			err = util.CloseFile(err, slice.file.handle)
 		}
-		if lerr := os.Remove(slice.file.name); lerr != nil && err == nil {
-			err = lerr
-		}
 		if slice.idxFile.handle != nil {
 			err = util.CloseFile(err, slice.idxFile.handle)
 		}
-		if lerr := os.Remove(slice.idxFile.name); lerr != nil && err == nil {
-			err = lerr
-		}
 		delete(ms.files, sliceID)
 	}
+	// Remove all message files (dat and idx) present
+	msgfiles, _ := filepath.Glob(filepath.Join(ms.fm.rootDir, ms.channelName, msgFilesPrefix+"*.*"))
+	for _, f := range msgfiles {
+		os.Remove(f)
+	}
 	// Reset generic counters
 	ms.empty()
 	// FileMsgStore specific

stores/filestore_msg_test.go

@@ -1745,6 +1745,40 @@ func TestFSRecoverEmptyIndexMsgFile(t *testing.T) {
 	}
 }
 
+func TestFSEmptyRemovesAllMsgsFiles(t *testing.T) {
+	cleanupFSDatastore(t)
+	defer cleanupFSDatastore(t)
+
+	s := createDefaultFileStore(t)
+	defer s.Close()
+
+	c := storeCreateChannel(t, s, "foo")
+	ms := c.Msgs
+	storeMsg(t, c, "foo", 1, []byte("msg"))
+	ms.(*FileMsgStore).RLock()
+	datname := ms.(*FileMsgStore).writeSlice.file.name
+	idxname := ms.(*FileMsgStore).writeSlice.idxFile.name
+	ms.(*FileMsgStore).RUnlock()
+	s.Close()
+
+	os.Remove(datname)
+
+	s, rs := openDefaultFileStore(t)
+	defer s.Close()
+
+	c = getRecoveredChannel(t, rs, "foo")
+	if n, b := msgStoreState(t, c.Msgs); n != 0 || b != 0 {
+		t.Fatalf("Unexpected state: %v - %v", n, b)
+	}
+	if err := c.Msgs.Empty(); err != nil {
+		t.Fatalf("Error on empty: %v", err)
+	}
+	// Make sure we have remove the index file too.
+	if fi, err := os.Stat(idxname); err == nil || fi != nil {
+		t.Fatalf("Expected index file to be gone, it was not")
+	}
+}
+
 func TestFSGetSeqFromTimestamp(t *testing.T) {
 	cleanupFSDatastore(t)
 	defer cleanupFSDatastore(t)