Add missing fsync calls in the snapshot module

components/tikv_util/src/file.rs

@@ -46,6 +46,13 @@ pub fn create_dir_if_not_exist<P: AsRef<Path>>(dir: P) -> io::Result<bool> {
     }
 }
 
+/// Call fsync on directory by its path
+pub fn sync_dir<P: AsRef<Path>>(path: P) -> io::Result<()> {
+    // File::open will not error when opening a directory
+    // because it just call libc::open and do not do the file or dir check
+    fs::File::open(path)?.sync_all()
+}
+
 const DIGEST_BUFFER_SIZE: usize = 1024 * 1024;
 
 /// Calculates the given file's CRC32 checksum.
@@ -190,4 +197,12 @@ mod tests {
         assert!(!create_dir_if_not_exist(&subdir).unwrap());
         assert!(delete_dir_if_exist(&subdir).unwrap());
     }
+
+    #[test]
+    fn test_sync_dir() {
+        let tmp_dir = TempDir::new("").unwrap();
+        sync_dir(tmp_dir.path()).unwrap();
+        let non_existent_file = tmp_dir.path().join("non_existent_file");
+        sync_dir(non_existent_file).unwrap_err();
+    }
 }

src/raftstore/store/snap.rs

@@ -38,7 +38,7 @@ use crate::raftstore::Result as RaftStoreResult;
 use engine::rocks::util::io_limiter::{IOLimiter, LimitWriter};
 use tikv_util::codec::bytes::{BytesEncoder, CompactBytesFromFileDecoder};
 use tikv_util::collections::{HashMap, HashMapEntry as Entry};
-use tikv_util::file::{calc_crc32, delete_file_if_exist, file_exists, get_file_size};
+use tikv_util::file::{calc_crc32, delete_file_if_exist, file_exists, get_file_size, sync_dir};
 use tikv_util::time::duration_to_sec;
 use tikv_util::HandyRwLock;
 
@@ -306,6 +306,7 @@ struct MetaFile {
 pub struct Snap {
     key: SnapKey,
     display_path: String,
+    dir_path: PathBuf,
     cf_files: Vec<CfFile>,
     cf_index: usize,
     meta_file: MetaFile,
@@ -364,6 +365,7 @@ impl Snap {
         let mut s = Snap {
             key: key.clone(),
             display_path,
+            dir_path,
             cf_files,
             cf_index: 0,
             meta_file,
@@ -646,6 +648,7 @@ impl Snap {
             // to indicate if the sst file is empty.
             if cf_file.kv_count > 0 {
                 fs::rename(&cf_file.tmp_path, &cf_file.path)?;
+                sync_dir(&self.dir_path)?;
                 cf_file.size = size;
                 // add size
                 self.size_track.fetch_add(size, Ordering::SeqCst);
@@ -667,6 +670,7 @@ impl Snap {
             f.flush()?;
         }
         fs::rename(&self.meta_file.tmp_path, &self.meta_file.path)?;
+        sync_dir(&self.dir_path)?;
         self.hold_tmp_files = false;
         Ok(())
     }
@@ -957,6 +961,7 @@ impl Snapshot for Snap {
             fs::rename(&cf_file.tmp_path, &cf_file.path)?;
             self.size_track.fetch_add(cf_file.size, Ordering::SeqCst);
         }
+        sync_dir(&self.dir_path)?;
         // write meta file
         let mut v = vec![];
         self.meta_file.meta.write_to_vec(&mut v)?;
@@ -966,6 +971,7 @@ impl Snapshot for Snap {
             meta_file.sync_all()?;
         }
         fs::rename(&self.meta_file.tmp_path, &self.meta_file.path)?;
+        sync_dir(&self.dir_path)?;
         self.hold_tmp_files = false;
         Ok(())
     }
@@ -1054,6 +1060,7 @@ impl Write for Snap {
                 file.write_all(&next_buf[0..left])?;
                 digest.write(&next_buf[0..left]);
                 cf_file.written_size += left as u64;
+                cf_file.file.as_mut().unwrap().sync_all()?;
                 self.cf_index += 1;
                 next_buf = &next_buf[left..];
             } else {