perf(pnpr): flush index markers with a plain fsync on Apple platforms (#15367)

Write the .present markers of a package index rebuild with a plain fsync
on Apple platforms instead of sync_all, which is F_FULLFSYNC there and
flushes the drive cache once per package (about 270 times per registry
start, most of the macOS Rust test time). The F_FULLFSYNC that publishes
.complete afterwards flushes the drive cache for the markers written
before it, so the index stays consistent across a power loss. The
markers keep the temporary-file rename of write_atomic, so a symlink
planted at a marker path is replaced rather than followed.

---------

Co-authored-by: Zoltan Kochan <z@kochan.io>

.changeset/pnpr-index-rebuild-plain-fsync.md

@@ -0,0 +1,5 @@
+---
+"@pnpm/pnpr": patch
+---
+
+On macOS, pnpr now builds its package index faster when it starts over a storage directory that has no index yet.

Cargo.lock

@@ -6768,6 +6768,7 @@ dependencies = [
  "base64 0.23.1",
  "futures-util",
  "getrandom 0.4.3",
+ "libc",
  "object_store",
  "pnpm-crypto-hash",
  "pnpm-network",

pnpr/crates/storage/Cargo.toml

@@ -31,6 +31,9 @@ tokio = { workspace = true }
 tempfile = { workspace = true }
 tracing = { workspace = true }
 
+[target.'cfg(target_vendor = "apple")'.dependencies]
+libc = { workspace = true }
+
 [dev-dependencies]
 reqwest = { workspace = true }
 tempfile = { workspace = true }

pnpr/crates/storage/src/local_store/hosted_backend.rs

@@ -2,7 +2,8 @@ use super::{
     Arc, AsyncReadExt, AsyncSeekExt, BlobFinalize, Body, BoxStream, CanonicalPackageName,
     DocumentWrite, ErrorKind, GetRange, HostedBackend, HostedBlobFile, HostedDocumentForUpdate,
     HostedDocumentVersion, HostedRevisionRefWrite, Path, PathBuf, RangedBlob, Result, SeekFrom,
-    Store, StreamExt, async_trait, fs, read_dir_if_present, stream, streaming, write_atomic,
+    Store, StreamExt, async_trait, create_tmp_file, fs, read_dir_if_present, stream, streaming,
+    write_atomic,
 };
 
 /// The single-node filesystem backend. It owns its directory tree
@@ -15,18 +16,14 @@ impl HostedBackend for Store {
         if fs::try_exists(&complete).await? {
             return Ok(());
         }
+        let index = self.root.join(".package-index");
         let mut files = self.list_blob_files();
         while let Some(file) = files.next().await {
             let file = file?;
             let Some(name) = file.path.strip_suffix("/package.json") else { continue };
-            write_atomic(
-                &self.root
-                    .join(".package-index")
-                    .join(name)
-                    .join(".present"),
-                b"",
-            )
-            .await?;
+            let marker_dir = index.join(name);
+            fs::create_dir_all(&marker_dir).await?;
+            write_index_marker(&marker_dir.join(".present")).await?;
         }
         write_atomic(&complete, b"").await
     }
@@ -252,3 +249,44 @@ async fn blob_file(root: &Path, entry: &fs::DirEntry) -> Result<HostedBlobFile>
         .replace('\\', "/");
     Ok(HostedBlobFile { path, modified: metadata.modified()?, size: metadata.len() })
 }
+
+/// Publishes an empty index marker by rename, as `write_atomic` does, so a
+/// symlink planted at the marker path is replaced rather than followed. It
+/// does not `sync_all` the marker: on Apple platforms that is `F_FULLFSYNC`,
+/// which also flushes the drive cache at about 4 ms alone and 19 ms under
+/// load, once per package, and the `F_FULLFSYNC` that publishes `.complete`
+/// afterwards flushes the drive cache for every marker written before it.
+async fn write_index_marker(path: &Path) -> Result<()> {
+    let (file, tmp) = create_tmp_file(path).await?;
+    let published = match flush_to_device(file).await {
+        Ok(()) => fs::rename(&tmp, path).await,
+        Err(err) => Err(err),
+    };
+    if let Err(err) = published {
+        let _ = fs::remove_file(&tmp).await;
+        return Err(err.into());
+    }
+    Ok(())
+}
+
+/// `fsync` without the drive-cache flush that `sync_all` adds on Apple
+/// platforms; `sync_all` everywhere else.
+async fn flush_to_device(file: fs::File) -> std::io::Result<()> {
+    #[cfg(target_vendor = "apple")]
+    {
+        let file = file.into_std().await;
+        tokio::task::spawn_blocking(move || {
+            use std::os::fd::AsRawFd as _;
+            // SAFETY: `file` keeps the descriptor open for the whole call.
+            if unsafe { libc::fsync(file.as_raw_fd()) } == 0 {
+                Ok(())
+            } else {
+                Err(std::io::Error::last_os_error())
+            }
+        })
+        .await
+        .map_err(std::io::Error::other)?
+    }
+    #[cfg(not(target_vendor = "apple"))]
+    file.sync_all().await
+}

pnpr/crates/storage/src/tests.rs

@@ -247,6 +247,25 @@ async fn package_index_migrates_nested_legacy_documents_and_ignores_removed_ones
     assert_eq!(storage.hosted_package_names().await.unwrap(), ["acme/app/tool"]);
 }
 
+#[cfg(unix)]
+#[tokio::test]
+async fn package_index_rebuild_replaces_a_symlink_planted_at_a_marker() {
+    let tmp = TempDir::new().unwrap();
+    let storage = storage_in(&tmp);
+    let root = tmp.path().join("storage");
+    fs::create_dir_all(root.join("acme/app")).await.unwrap();
+    fs::write(root.join("acme/app/package.json"), b"{}").await.unwrap();
+    let outside = tmp.path().join("outside");
+    fs::write(&outside, b"keep").await.unwrap();
+    let marker = root.join(".package-index/acme/app/.present");
+    fs::create_dir_all(marker.parent().unwrap()).await.unwrap();
+    fs::symlink(&outside, &marker).await.unwrap();
+    storage.rebuild_package_index().await.unwrap();
+    assert_eq!(fs::read(&outside).await.unwrap(), b"keep");
+    assert!(fs::symlink_metadata(&marker).await.unwrap().is_file());
+    assert_eq!(storage.hosted_package_names().await.unwrap(), ["acme/app"]);
+}
+
 #[tokio::test]
 async fn package_index_removes_deleted_and_failed_package_entries() {
     let tmp = TempDir::new().unwrap();