lib/storage: substitute remaining calls to fs.MustRemoveAll with fs.MustRemoveDirAtomic

Updates https://github.com/VictoriaMetrics/VictoriaMetrics/issues/3038

lib/fs/fs.go

@@ -204,22 +204,26 @@ func IsEmptyDir(path string) bool {
 //  2. Remove the "<dir>.must-remove.XYZ" in background.
 //
 // If the process crashes after the step 1, then the directory must be removed
-// on the next process start by calling MustRemoveTemporaryDirs.
+// on the next process start by calling MustRemoveTemporaryDirs on the parent directory.
 func MustRemoveDirAtomic(dir string) {
+	if !IsPathExist(dir) {
+		return
+	}
 	n := atomic.AddUint64(&atomicDirRemoveCounter, 1)
 	tmpDir := fmt.Sprintf("%s.must-remove.%d", dir, n)
 	if err := os.Rename(dir, tmpDir); err != nil {
 		logger.Panicf("FATAL: cannot move %s to %s: %s", dir, tmpDir, err)
 	}
-	MustSyncPath(dir)
 	MustRemoveAll(tmpDir)
+	parentDir := filepath.Dir(dir)
+	MustSyncPath(parentDir)
 }
 
 var atomicDirRemoveCounter = uint64(time.Now().UnixNano())
 
 // MustRemoveTemporaryDirs removes all the subdirectories with ".must-remove.<XYZ>" suffix.
 //
-// Such directories may be left on unclean shutdown during MustRemoveDirAtomic.
+// Such directories may be left on unclean shutdown during MustRemoveDirAtomic call.
 func MustRemoveTemporaryDirs(dir string) {
 	d, err := os.Open(dir)
 	if err != nil {

lib/storage/block_stream_writer.go

@@ -110,7 +110,7 @@ func (bsw *blockStreamWriter) InitFromFilePart(path string, nocache bool, compre
 	timestampsPath := path + "/timestamps.bin"
 	timestampsFile, err := filestream.Create(timestampsPath, nocache)
 	if err != nil {
-		fs.MustRemoveAll(path)
+		fs.MustRemoveDirAtomic(path)
 		return fmt.Errorf("cannot create timestamps file: %w", err)
 	}
 
@@ -118,7 +118,7 @@ func (bsw *blockStreamWriter) InitFromFilePart(path string, nocache bool, compre
 	valuesFile, err := filestream.Create(valuesPath, nocache)
 	if err != nil {
 		timestampsFile.MustClose()
-		fs.MustRemoveAll(path)
+		fs.MustRemoveDirAtomic(path)
 		return fmt.Errorf("cannot create values file: %w", err)
 	}
 
@@ -127,7 +127,7 @@ func (bsw *blockStreamWriter) InitFromFilePart(path string, nocache bool, compre
 	if err != nil {
 		timestampsFile.MustClose()
 		valuesFile.MustClose()
-		fs.MustRemoveAll(path)
+		fs.MustRemoveDirAtomic(path)
 		return fmt.Errorf("cannot create index file: %w", err)
 	}
 
@@ -139,7 +139,7 @@ func (bsw *blockStreamWriter) InitFromFilePart(path string, nocache bool, compre
 		timestampsFile.MustClose()
 		valuesFile.MustClose()
 		indexFile.MustClose()
-		fs.MustRemoveAll(path)
+		fs.MustRemoveDirAtomic(path)
 		return fmt.Errorf("cannot create metaindex file: %w", err)
 	}
 

lib/storage/index_db.go

@@ -308,7 +308,7 @@ func (db *indexDB) decRef() {
 	}
 
 	logger.Infof("dropping indexDB %q", tbPath)
-	fs.MustRemoveAll(tbPath)
+	fs.MustRemoveDirAtomic(tbPath)
 	logger.Infof("indexDB %q has been dropped", tbPath)
 }
 

lib/storage/index_db_test.go

@@ -2080,5 +2080,5 @@ func stopTestStorage(s *Storage) {
 	s.metricIDCache.Stop()
 	s.metricNameCache.Stop()
 	s.tsidCache.Stop()
-	fs.MustRemoveAll(s.cachePath)
+	fs.MustRemoveDirAtomic(s.cachePath)
 }