lib/{storage,mergeset}: make sure the newly created data part is visible in the parent directory before storing it in parts.json

The newly created data part could become missing after unclean shutdown (such as hardware power off),
since the contents of the parent directory wasn't synced to disk before storing the newly created data part in the parts.json file.

Fix this by syncing the parent directory contents before storing the newly created part in the parts.json file.

This commit is based on https://github.com/VictoriaMetrics/VictoriaLogs/pull/507

lib/fs/fs.go

@@ -18,6 +18,15 @@ import (
 
 var tmpFileNum atomicutil.Uint64
 
+// MustSyncPathAndParentDir fsyncs the path and the parent dir.
+//
+// This guarantees that the path is visible and readable after unclean shutdown.
+func MustSyncPathAndParentDir(path string) {
+	MustSyncPath(path)
+	parentDirPath := filepath.Dir(path)
+	MustSyncPath(parentDirPath)
+}
+
 // MustSyncPath syncs contents of the given path.
 func MustSyncPath(path string) {
 	mustSyncPath(path)

lib/mergeset/inmemory_part.go

@@ -50,8 +50,7 @@ func (mp *inmemoryPart) MustStoreToDisk(path string) {
 
 	mp.ph.MustWriteMetadata(path)
 
-	fs.MustSyncPath(path)
-	// Do not sync parent directory - it must be synced by the caller.
+	fs.MustSyncPathAndParentDir(path)
 }
 
 // Init initializes mp from ib.

lib/mergeset/table.go

@@ -1246,7 +1246,7 @@ func (tb *Table) mergeParts(pws []*partWrapper, stopCh <-chan struct{}, isFinal
 		mpNew.ph = *ph
 	} else {
 		// Make sure the created part directory listing is synced.
-		fs.MustSyncPath(dstPartPath)
+		fs.MustSyncPathAndParentDir(dstPartPath)
 	}
 
 	// Atomically swap the source parts with the newly created part.

lib/storage/inmemory_part.go

@@ -52,8 +52,7 @@ func (mp *inmemoryPart) MustStoreToDisk(path string) {
 
 	mp.ph.MustWriteMetadata(path)
 
-	fs.MustSyncPath(path)
-	// Do not sync parent directory - it must be synced by the caller.
+	fs.MustSyncPathAndParentDir(path)
 }
 
 // InitFromRows initializes mp from the given rows.

lib/storage/partition.go

@@ -1451,7 +1451,7 @@ func (pt *partition) mergeParts(pws []*partWrapper, stopCh <-chan struct{}, isFi
 		mpNew.ph = *ph
 	} else {
 		// Make sure the created part directory listing is synced.
-		fs.MustSyncPath(dstPartPath)
+		fs.MustSyncPathAndParentDir(dstPartPath)
 	}
 
 	// Atomically swap the source parts with the newly created part.