diff --git a/Cargo.lock b/Cargo.lock index 594744e8f0..c04b3e4d73 100644 --- a/Cargo.lock +++ b/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", diff --git a/Cargo.toml b/Cargo.toml index fa41f0f24c..156ca0b733 100644 --- a/Cargo.toml +++ b/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 " depends = "$auto" diff --git a/lib/collection/src/shards/shard_holder/mod.rs b/lib/collection/src/shards/shard_holder/mod.rs index 64981ada64..89d7db4996 100644 --- a/lib/collection/src/shards/shard_holder/mod.rs +++ b/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(()) }, diff --git a/lib/common/common/Cargo.toml b/lib/common/common/Cargo.toml index 658e446398..14ce2a276b 100644 --- a/lib/common/common/Cargo.toml +++ b/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 } diff --git a/lib/common/common/src/fs.rs b/lib/common/common/src/fs.rs new file mode 100644 index 0000000000..fd24a58c48 --- /dev/null +++ b/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(()) +} diff --git a/lib/common/common/src/lib.rs b/lib/common/common/src/lib.rs index cd93275e16..615bc5ba63 100644 --- a/lib/common/common/src/lib.rs +++ b/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; diff --git a/lib/segment/src/common/validate_snapshot_archive.rs b/lib/segment/src/common/validate_snapshot_archive.rs index f983fa6cd9..4303c204d0 100644 --- a/lib/segment/src/common/validate_snapshot_archive.rs +++ b/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) } diff --git a/lib/storage/src/content_manager/snapshots/recover.rs b/lib/storage/src/content_manager/snapshots/recover.rs index 3fadf037b6..772ee5431c 100644 --- a/lib/storage/src/content_manager/snapshots/recover.rs +++ b/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??;