perf(ecstore): parallelize tmp xl.meta write and shard fdatasync on commit (#4487)

The non-inline rename_data commit path made the tmp xl.meta durable and then
fdatasynced the shard files in two sequential awaits, each its own blocking
round-trip. The two operations touch disjoint paths (the tmp xl.meta under the
tmp bucket vs the shard data dir) and have no ordering constraint between them:
both only need to be durable before the commit renames that follow.

Run them concurrently with tokio::join!, dropping a blocking round-trip from
the PUT commit critical path (backlog#922 step 2). The commit ordering is
unchanged — both futures complete before any rename, and a failure in either
aborts before the rename exactly as the sequential version did (tmp-meta error
is still surfaced first). Payload and metadata durability semantics are
identical under strict and relaxed tiers.

Validated by the rename_data crash-consistency harness (backlog#935): a crash
at any pre-commit point still reopens as old-or-new, never mixed. Existing
rename_data / durability-tier / disk::os tests are unchanged and pass.

Refs: rustfs/backlog#922 (HP-1 step 2), rustfs/backlog#936

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/disk/local.rs

@@ -4977,28 +4977,40 @@ impl DiskAPI for LocalDisk {
             } else {
                 SyncMode::None
             };
-            self.write_all_private(
+            // The tmp xl.meta write and the shard-file fdatasync are independent
-                src_volume,
+            // (disjoint paths) and both only need to be durable before the commit
-                &format!("{}/{}", &src_path, STORAGE_FORMAT_FILE),
+            // renames below, so run them concurrently to drop a blocking
-                new_dst_buf.into(),
+            // round-trip from the PUT commit critical path (rustfs/backlog#922
-                tmp_meta_sync,
+            // step 2). The "contents durable -> rename -> dst dir fsync" ordering
-                src_file_parent,
+            // is unchanged — both futures complete before any rename — which the
-            )
+            // rename_data crash-consistency harness (backlog#935) exercises.
-            .await?;
+            //
-
+            // Shard durability: once rename_data succeeds the write is
-            // Make shard files durable before the commit rename: once rename_data
+            // acknowledged, so data must not live only in the page cache.
-            // succeeds the write is acknowledged, so data must not live only in the
+            // Multipart parts were already synced during rename_part, so their
-            // page cache. Multipart parts were already synced during rename_part, so
+            // fdatasync here is a cheap no-op. A missing source dir is left for the
-            // their fdatasync here is a cheap no-op. A missing source dir is left for
+            // rename below to report through the existing rollback path. Payload
-            // the rename below to report through the existing rollback path.
+            // durability is kept by both strict and relaxed.
-            // Payload durability: kept by both strict and relaxed.
+            // Bound to a local so the borrow lives across the join! below.
-            if durability.syncs_data_shards()
+            let tmp_meta_rel_path = format!("{}/{}", &src_path, STORAGE_FORMAT_FILE);
-                && let Some((src_data_path, _)) = has_data_dir_path.as_ref()
+            let tmp_meta_write =
-                && let Err(err) = os::sync_dir_files(src_data_path).await
+                self.write_all_private(src_volume, &tmp_meta_rel_path, new_dst_buf.into(), tmp_meta_sync, src_file_parent);
-                && err.kind() != ErrorKind::NotFound
+            let shard_sync = async {
-            {
+                if durability.syncs_data_shards()
-                return Err(to_file_error(err).into());
+                    && let Some((src_data_path, _)) = has_data_dir_path.as_ref()
-            }
+                    && let Err(err) = os::sync_dir_files(src_data_path).await
+                    && err.kind() != ErrorKind::NotFound
+                {
+                    return Err::<(), DiskError>(to_file_error(err).into());
+                }
+                Ok(())
+            };
+            let (tmp_meta_res, shard_sync_res) = tokio::join!(tmp_meta_write, shard_sync);
+            // Surface a tmp-meta failure first (its prior serial position), then a
+            // shard-sync failure; either aborts before any rename, exactly as the
+            // sequential version did.
+            tmp_meta_res?;
+            shard_sync_res?;
 
             if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref()
                 && let Err(err) = rename_all(src_data_path, dst_data_path, &skip_parent).await