Use syncfs (#7883)

* Remove call to Archive::set_sync

Also, this method was the last remaining part of our `tar-rs` fork,
so we can switch to the upstream version now.

* Do syncfs

Cargo.lock

@@ -1281,6 +1281,7 @@ dependencies = [
  "log",
  "memmap2",
  "memory",
+ "nix 0.30.1",
  "num-traits",
  "num_cpus",
  "ordered-float 5.1.0",
@@ -7045,8 +7046,9 @@ checksum = "55937e1799185b12863d447f42597ed69d9928686b8d88a1df17376a097d8369"
 
 [[package]]
 name = "tar"
-version = "0.4.41"
-source = "git+https://github.com/qdrant/tar-rs?branch=main#856dbd090eede1736604f23cfe99a104b5639734"
+version = "0.4.44"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d863878d212c87a19c1a610eb53bb01fe12951c0501cf5a0d65f724914a667a"
 dependencies = [
  "filetime",
  "libc",

Cargo.toml

@@ -377,10 +377,6 @@ opt-level = 3
 [profile.ci.package.sha2]
 opt-level = 3
 
-[patch.crates-io]
-# Temporary patch until our PRs are merged.
-tar = { git = "https://github.com/qdrant/tar-rs", branch = "main" }
-
 [package.metadata.deb]
 maintainer = "Qdrant Team <team@qdrant.com>"
 depends = "$auto"

lib/collection/src/shards/shard_holder/mod.rs

@@ -1267,6 +1267,7 @@ impl ShardHolder {
                         this_peer_id,
                         is_distributed,
                     )?;
+                    common::fs::bulk_sync_dir(&snapshot_temp_dir)?;
 
                     Ok(())
                 },

lib/common/common/Cargo.toml

@@ -20,6 +20,7 @@ bitvec = { workspace = true }
 bytemuck = { workspace = true }
 chrono = { workspace = true }
 fs-err = { workspace = true }
+nix = { workspace = true }
 num-traits = { workspace = true }
 num_cpus = "1.17"
 ordered-float = { workspace = true }

lib/common/common/src/fs.rs

@@ -0,0 +1,40 @@
+use std::io;
+use std::path::Path;
+
+use fs_err::File;
+
+/// Commits filesystem caches for the given directory.
+///
+/// On Linux, it commits the entire filesystem containing the directory.
+pub fn bulk_sync_dir(dir: &Path) -> io::Result<()> {
+    // Matches all platforms that have `nix::unistd::syncfs` function.
+    // https://github.com/nix-rust/nix/blob/v0.30.1/src/unistd.rs#L1679
+    #[cfg(any(target_os = "linux", target_os = "android", target_os = "hurd"))]
+    // If the directory contains a lot of small files, calling `syncfs` once
+    // could be faster than calling `fsync` on each file individually.
+    // See https://man7.org/linux/man-pages/man2/syncfs.2.html
+    match nix::unistd::syncfs(File::open(dir)?) {
+        Ok(()) => return Ok(()),
+        // Don't return an error as it could be caused by issues outside our
+        // control. Just log a warning.
+        Err(e) => log::warn!("syncfs failed for {}: {e}", dir.display()),
+    }
+
+    // Fallback
+    sync_dir_with_fsync(dir)
+}
+
+/// Calls `fsync` recursively.
+fn sync_dir_with_fsync(dir: &Path) -> io::Result<()> {
+    for entry in fs_err::read_dir(dir)? {
+        let entry = entry?;
+        if entry.file_type()?.is_dir() {
+            sync_dir_with_fsync(&entry.path())?;
+        } else {
+            File::open(entry.path())?.sync_all()?;
+        }
+    }
+    #[cfg(unix)]
+    File::open(dir)?.sync_all()?;
+    Ok(())
+}

lib/common/common/src/lib.rs

@@ -13,6 +13,7 @@ pub mod either_variant;
 pub mod ext;
 pub mod fixed_length_priority_queue;
 pub mod flags;
+pub mod fs;
 pub mod is_alive_lock;
 pub mod iterator_ext;
 pub mod math;

lib/segment/src/common/validate_snapshot_archive.rs

@@ -59,7 +59,6 @@ pub fn open_snapshot_archive(
 
     let mut ar = tar::Archive::new(io::BufReader::new(file));
     ar.set_overwrite(false);
-    ar.set_sync(true);
 
     Ok(ar)
 }

lib/storage/src/content_manager/snapshots/recover.rs

@@ -157,7 +157,9 @@ async fn _do_recover_from_snapshot(
             &tmp_collection_dir_clone,
             this_peer_id,
             is_distributed,
-        )
+        )?;
+        common::fs::bulk_sync_dir(&tmp_collection_dir_clone)?;
+        Ok::<(), StorageError>(())
     });
     restoring.await??;