ci: repair e2e selection and verify tool downloads via API (#9240 )

* ci: align full e2e selection with retired catalog test * ci: verify pinned tools through GitHub release asset API * ci: fetch pinned typos through the release asset API * ci: download pinned actionlint through the release asset API * ci: ensure GitHub CLI exists before pinned mc download * ci: budget full quick checks after tool installation
fix: reduce scanner backoff and classify downstream closures (#9235 )
2026-11-08 19:43:05 +00:00 · 2026-11-08 14:09:26 +00:00 · 2026-11-08 02:18:17 -05:00 · 2026-11-08 02:12:40 -05:00
28 changed files with 1456 additions and 739 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=219ac11e720b3f54dfee4d5d5e95171785ba159b2fe01db3037a8f5061c1ad32
-sha256-linux=2d34698cc8ec52d5b6c37ac612919a1d21f36cd56a1af539a7522a4eea518e26
+sha256-darwin=9901a4950099b94d3d2618524d540f1ed4bf6023044903886f98b6c4bf5f30fb
+sha256-linux=09fd5f48f5556c9fc576357222dccd7ebb0b90b790798f1834db2ec5d67b4952
@@ -54,6 +54,7 @@ script-tests: ## Run shell script tests
 	./scripts/check_embedded_secrets.sh --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_download_release_asset.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_btf_prototypes.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_glibc.py
@@ -28,11 +28,12 @@ runs:

    - name: Install actionlint
      shell: bash
+      env:
+        GH_TOKEN: ${{ github.token }}
+        ASSET_DOWNLOADER: ${{ github.action_path }}/../setup/download_release_asset.py
      run: |
        actionlint_dir="$(mktemp -d "${RUNNER_TEMP}/actionlint.XXXXXX")"
-        curl --fail --location --silent --show-error \
-          --output "$actionlint_dir/actionlint.tar.gz" \
-          https://github.com/rhysd/actionlint/releases/download/v1.7.12/actionlint_1.7.12_linux_amd64.tar.gz
+        python3 "$ASSET_DOWNLOADER" --repository rhysd/actionlint --tag v1.7.12 --asset actionlint_1.7.12_linux_amd64.tar.gz --output "$actionlint_dir/actionlint.tar.gz" --sha256 8aca8db96f1b94770f1b0d72b6dddcb1ebb8123cb3712530b08cc387b349a3d8
        echo "8aca8db96f1b94770f1b0d72b6dddcb1ebb8123cb3712530b08cc387b349a3d8  $actionlint_dir/actionlint.tar.gz" | sha256sum --check --status
        tar -xzf "$actionlint_dir/actionlint.tar.gz" -C "$actionlint_dir" actionlint
        rm "$actionlint_dir/actionlint.tar.gz"
@@ -66,10 +66,7 @@ inputs:
 runs:
  using: "composite"
  steps:
-    # protobuf-compiler is deliberately absent: the setup-protoc step below
-    # installs 35.1 into the tool cache and prepends it to PATH, so the apt
-    # build (older, and never version-matched) was shadowed on every run and
-    # simply never used.
+    # Use the pinned code generators rather than Ubuntu's older packages.
    - name: Install system dependencies (Ubuntu)
      if: runner.os == 'Linux'
      shell: bash
@@ -79,7 +76,9 @@ runs:
          build-essential \
          pkg-config \
          libssl-dev \
-          ripgrep
+          ripgrep \
+          python3 \
+          gh

    # musl-gcc is needed by the native musl release leg, and zip/unzip by the
    # release packaging steps. No CI test lane touches any of them.
@@ -88,13 +87,32 @@ runs:
      shell: bash
      run: sudo apt-get install -y musl-tools zip unzip

+    - name: Install pinned code generators through GitHub API (Linux x64)
+      if: runner.os == 'Linux' && runner.arch == 'X64'
+      shell: bash
+      env:
+        GH_TOKEN: ${{ github.token }}
+        ASSET_DOWNLOADER: ${{ github.action_path }}/download_release_asset.py
+      run: |
+        python3 "$ASSET_DOWNLOADER" --repository protocolbuffers/protobuf --tag v35.1 --asset protoc-35.1-linux-x86_64.zip --output "$RUNNER_TEMP/protoc.zip"
+        python3 "$ASSET_DOWNLOADER" --repository google/flatbuffers --tag v25.12.19 --asset 'Linux.flatc.binary.g++-13.zip' --output "$RUNNER_TEMP/flatc.zip"
+        python3 -m zipfile -e "$RUNNER_TEMP/protoc.zip" "$RUNNER_TEMP/protoc"
+        python3 -m zipfile -e "$RUNNER_TEMP/flatc.zip" "$RUNNER_TEMP/flatc"
+        chmod +x "$RUNNER_TEMP/protoc/bin/protoc" "$RUNNER_TEMP/flatc/flatc"
+        "$RUNNER_TEMP/protoc/bin/protoc" --version
+        "$RUNNER_TEMP/flatc/flatc" --version
+        echo "$RUNNER_TEMP/protoc/bin" >> "$GITHUB_PATH"
+        echo "$RUNNER_TEMP/flatc" >> "$GITHUB_PATH"
+
    - name: Install protoc
+      if: runner.os != 'Linux' || runner.arch != 'X64'
      uses: rustfs/setup-protoc@a3705324d8f9bf5b6c3573fb6cf8ae421db55dd6 # v3.0.1
      with:
        version: "35.1"
        repo-token: ${{ github.token }}

    - name: Install flatc
+      if: runner.os != 'Linux' || runner.arch != 'X64'
      uses: Nugine/setup-flatc@698800de72a96bfb22cf60431dc21a2ff9a7e07b # v1
      with:
        version: "25.12.19"
@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Download a named GitHub release asset through its API and verify its digest."""
+
+import argparse
+import hashlib
+import json
+from pathlib import Path
+import re
+import subprocess
+import sys
+import tempfile
+from urllib.parse import quote
+
+
+def download(repository, tag, name, output, expected_sha256=None):
+    if not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*", repository):
+        raise ValueError("invalid release repository")
+    if not tag or not name or "/" in name or "\\" in name:
+        raise ValueError("invalid release tag or asset name")
+    if expected_sha256 is not None and not re.fullmatch(r"[0-9a-f]{64}", expected_sha256):
+        raise ValueError("invalid pinned asset digest")
+    result = subprocess.run(
+        ["gh", "api", "--hostname", "github.com", f"repos/{repository}/releases/tags/{quote(tag, safe='')}"],
+        check=True, capture_output=True, text=True, timeout=60,
+    )
+    release = json.loads(result.stdout)
+    if not isinstance(release, dict) or release.get("tag_name") != tag:
+        raise ValueError("release metadata does not match the requested tag")
+    assets = release.get("assets")
+    if not isinstance(assets, list) or any(not isinstance(asset, dict) for asset in assets):
+        raise ValueError("invalid release asset inventory")
+    matches = [asset for asset in assets if asset.get("name") == name]
+    if len(matches) != 1:
+        raise ValueError("release must contain exactly one named asset")
+    asset = matches[0]
+    asset_id, size, digest = asset.get("id"), asset.get("size"), asset.get("digest")
+    if type(asset_id) is not int or asset_id <= 0 or type(size) is not int or not 0 < size <= 512 * 1024 * 1024:
+        raise ValueError("invalid release asset identity or size")
+    if not isinstance(digest, str) or not re.fullmatch(r"sha256:[0-9a-f]{64}", digest):
+        raise ValueError("release asset has no valid SHA-256 digest")
+    sha256 = digest.removeprefix("sha256:")
+    if expected_sha256 is not None and expected_sha256 != sha256:
+        raise ValueError("release digest differs from the pinned digest")
+    output = Path(output)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    temporary = None
+    try:
+        with tempfile.NamedTemporaryFile(dir=output.parent, prefix=".release-asset-", delete=False) as stream:
+            temporary = Path(stream.name)
+            # gh authenticates to the API; its HTTP client strips credentials on
+            # cross-host redirects. No browser URL or token is supplied in argv.
+            subprocess.run(
+                ["gh", "api", "--hostname", "github.com", "--header", "Accept: application/octet-stream", f"repos/{repository}/releases/assets/{asset_id}"],
+                check=True, stdout=stream, timeout=300,
+            )
+        with temporary.open("rb") as stream:
+            actual = hashlib.file_digest(stream, "sha256").hexdigest()
+        if temporary.stat().st_size != size or actual != sha256:
+            raise ValueError("downloaded release asset failed size or SHA-256 verification")
+        temporary.replace(output)
+        print(f"Verified release asset: {repository} {tag} {name} sha256={actual}")
+    finally:
+        if temporary is not None:
+            temporary.unlink(missing_ok=True)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--repository", required=True)
+    parser.add_argument("--tag", required=True)
+    parser.add_argument("--asset", required=True)
+    parser.add_argument("--output", required=True, type=Path)
+    parser.add_argument("--sha256")
+    args = parser.parse_args()
+    try:
+        download(args.repository, args.tag, args.asset, args.output, args.sha256)
+    except (ValueError, KeyError, TypeError, json.JSONDecodeError, subprocess.SubprocessError, OSError) as error:
+        print(f"Release asset download failed: {error}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -142,15 +142,23 @@ jobs:
      - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
        with:
          persist-credentials: false
+      - name: Install pinned typos through GitHub API
+        env:
+          GH_TOKEN: ${{ github.token }}
+          TYPOS_VERSION: 1.50.2
+        run: |
+          python3 .github/actions/setup/download_release_asset.py --repository crate-ci/typos --tag "v${TYPOS_VERSION}" --asset "typos-v${TYPOS_VERSION}-x86_64-unknown-linux-musl.tar.gz" --output "$RUNNER_TEMP/typos.tar.gz"
+          mkdir -p "$RUNNER_TEMP/typos"
+          tar -xzf "$RUNNER_TEMP/typos.tar.gz" -C "$RUNNER_TEMP/typos"
      - name: Typos check with custom config file
-        uses: crate-ci/typos@512fc24f32f44ab01972217aaaf3dc86ec234d53 # v1.50.2
+        run: '"$RUNNER_TEMP/typos/typos" .'

  # Fail early with compile-free checks for every pull request.
  quick-checks:
    name: Quick Checks
    if: github.event_name != 'pull_request' || github.event.action != 'closed'
    runs-on: ubuntu-latest
-    timeout-minutes: 10
+    timeout-minutes: 20
    steps:
      - name: Checkout repository
        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
@@ -1137,11 +1145,16 @@ jobs:

      - name: Install mc
        env:
+          GH_TOKEN: ${{ github.token }}
          MC_VERSION: RELEASE.2025-08-13T08-35-41Z
          MC_SHA256: 01f866e9c5f9b87c2b09116fa5d7c06695b106242d829a8bb32990c00312e891
        run: |
+          if ! command -v gh >/dev/null 2>&1; then
+            sudo apt-get update
+            sudo apt-get install -y gh
+          fi
          MC_BINARY="mc.linux-amd64.${MC_VERSION}"
-          curl -fsSLo "$RUNNER_TEMP/mc" "https://github.com/minio/mc/releases/download/${MC_VERSION}/${MC_BINARY}"
+          python3 .github/actions/setup/download_release_asset.py --repository minio/mc --tag "$MC_VERSION" --asset "$MC_BINARY" --output "$RUNNER_TEMP/mc" --sha256 "$MC_SHA256"
          echo "${MC_SHA256}  $RUNNER_TEMP/mc" | sha256sum --check --status
          chmod +x "$RUNNER_TEMP/mc"
          echo "$RUNNER_TEMP" >> "$GITHUB_PATH"
@@ -10759,20 +10759,17 @@ dependencies = [
 "arc-swap",
 "async-compression",
 "async-trait",
- "base64-simd",
 "brotli 9.1.1",
 "bytes",
 "chrono",
 "crc-fast",
 "datafusion",
- "flatbuffers",
 "flate2",
 "futures",
 "futures-core",
 "hotpath",
 "http 1.5.0",
 "lz4",
- "lz4_flex",
 "memchr",
 "metrics",
 "parking_lot 0.13.0",
@@ -296,7 +296,6 @@ local-ip-address = "0.6.13"
 log = "0.4"
 memmap2 = "0.9.12"
 lz4 = "1.28.1"
-lz4_flex = { version = "=0.14.0", default-features = false, features = ["std", "frame"] }
 matchit = "0.9.2"
 md-5 = "0.11.0"
 mime_guess = "2.0.5"
@@ -376,6 +376,21 @@ struct WalkCopyError {
    source: io::Error,
 }

+impl WalkCopyError {
+    fn into_disk_error(self) -> DiskError {
+        // Only a local output write/flush proves the metacache consumer closed.
+        // An identical I/O kind from the remote input remains a transport failure.
+        if matches!(self.reason, RemoteWalkFailureReason::Write | RemoteWalkFailureReason::Flush)
+            && self.source.kind() == io::ErrorKind::BrokenPipe
+            && self.source.get_ref().is_none()
+        {
+            DiskError::metacache_output_stream_closed()
+        } else {
+            DiskError::Io(self.source)
+        }
+    }
+}
+
 async fn copy_stream_with_buffer<R, W>(
    reader: &mut R,
    writer: &mut W,
@@ -2096,6 +2111,7 @@ impl RemoteDisk {
        failure_health_action: FailureHealthAction,
    ) {
        if let Err(err) = operation_result
+            && !err.is_metacache_output_stream_closed()
            && is_network_like_disk_error(err)
        {
            counter!(
@@ -4056,7 +4072,7 @@ impl DiskAPI for RemoteDisk {
                            Ok(_) => return Ok(()),
                            Err(error) => {
                                failure_reason = Some(error.reason);
-                                return Err(DiskError::Io(error.source));
+                                return Err(error.into_disk_error());
                            }
                        }
                    }
@@ -9687,6 +9703,43 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn remote_walk_preserves_local_metacache_consumer_close() {
+        let transport =
+            RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::Data(b"metacache payload".to_vec())]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let (mut writer, reader) = tokio::io::duplex(1);
+        drop(reader);
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("a dropped metacache consumer must stop the producer");
+        assert!(
+            error.is_metacache_output_stream_closed(),
+            "preserve the benign completion marker: {error:?}"
+        );
+        assert_eq!(transport.calls().len(), 1, "local consumer closure is not a remote open retry");
+        assert!(!disk.health.is_faulty());
+    }
+
+    #[tokio::test]
+    async fn remote_walk_keeps_remote_read_broken_pipe_as_transport_failure() {
+        let transport = RetryingWalkDirInternodeDataTransport::with_steps(vec![WalkDirTestStep::PartialDataThenError {
+            data: b"partial metacache".to_vec(),
+            error: io::ErrorKind::BrokenPipe.into(),
+        }]);
+        let disk = new_remote_disk_with_transport(Arc::new(transport.clone())).await;
+        let mut writer = Vec::new();
+        let error = disk
+            .walk_dir(WalkDirOptions::default(), &mut writer)
+            .await
+            .expect_err("an upstream close after partial output is a failed walk");
+        assert!(!error.is_metacache_output_stream_closed());
+        assert!(error.contains_io_error_kind(io::ErrorKind::BrokenPipe));
+        assert_eq!(writer, b"partial metacache");
+        assert_eq!(transport.calls().len(), 1, "partial output must not be replayed");
+    }
+
    #[tokio::test]
    async fn walk_stream_backpressure_limits_read_ahead_and_is_cancellable() {
        let mut reader = Cursor::new(vec![5; 4 * DEFAULT_READ_BUFFER_SIZE]);
@@ -682,15 +682,11 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
        return Err(Error::other("ServerNotInitialized"));
    };

-    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
-    for d in all_disks {
-        let pool_info = pools_info.entry(d.pool_index).or_default();
-        let erasure_set = pool_info.entry(d.set_index).or_default();
-
-        if erasure_set.id == 0 {
-            erasure_set.id = d.set_index;
+    collect_pools_info(all_disks, |pool_index, set_index| {
+        let store = store.clone();
+        async move {
            let load = load_data_usage_cache(
-                store.pools[d.pool_index as usize].disk_set[d.set_index as usize].as_ref(),
+                store.pools[pool_index as usize].disk_set[set_index as usize].as_ref(),
                DATA_USAGE_CACHE_NAME,
            );
            #[cfg(feature = "diagnostic-read-version-caller")]
@@ -698,11 +694,34 @@ async fn get_pools_info(all_disks: &[Disk]) -> Result<HashMap<i32, HashMap<i32,
                rustfs_io_metrics::internode_metrics::ReadVersionEntrySource::ServerInfoPools,
                load,
            );
-            match load.await {
-                Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
-                Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
-            }
+            load.await
        }
+    })
+    .await
+}
+
+async fn collect_pools_info<F, Fut>(all_disks: &[Disk], mut load_usage: F) -> Result<HashMap<i32, HashMap<i32, ErasureSetInfo>>>
+where
+    F: FnMut(i32, i32) -> Fut,
+    Fut: std::future::Future<Output = Result<rustfs_data_usage::DataUsageCache>>,
+{
+    let mut pools_info: HashMap<i32, HashMap<i32, ErasureSetInfo>> = HashMap::new();
+    for d in all_disks {
+        let pool_info = pools_info.entry(d.pool_index).or_default();
+        let erasure_set = match pool_info.entry(d.set_index) {
+            std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(),
+            std::collections::hash_map::Entry::Vacant(entry) => {
+                let erasure_set = entry.insert(ErasureSetInfo {
+                    id: d.set_index,
+                    ..Default::default()
+                });
+                match load_usage(d.pool_index, d.set_index).await {
+                    Ok(cache) => apply_erasure_set_usage(&cache, erasure_set),
+                    Err(_) => erasure_set.usage_error = Some(DATA_USAGE_UNAVAILABLE_ERROR.to_string()),
+                }
+                erasure_set
+            }
+        };

        erasure_set.raw_capacity += d.total_space;
        erasure_set.raw_usage += d.used_space;
@@ -1241,6 +1260,59 @@ mod tests {
        assert_eq!(info.deployment_id, expected_deployment_id);
    }

+    #[tokio::test]
+    async fn pool_usage_is_loaded_once_per_set_including_set_zero() {
+        let mut disks = Vec::new();
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            disks.extend((0..count).map(|_| Disk {
+                pool_index: pool,
+                set_index: set,
+                used_space: 10,
+                total_space: 100,
+                ..Default::default()
+            }));
+        }
+        let mut loads = Vec::new();
+        let info = super::collect_pools_info(&disks, |pool, set| {
+            loads.push((pool, set));
+            async move {
+                if pool == 1 && set == 0 {
+                    return Err(crate::error::Error::other("usage unavailable"));
+                }
+                let mut cache = rustfs_data_usage::DataUsageCache::default();
+                cache.info.name = DATA_USAGE_ROOT.to_owned();
+                cache.info.last_update = Some(std::time::SystemTime::UNIX_EPOCH);
+                cache.info.snapshot_complete = true;
+                cache.replace(
+                    DATA_USAGE_ROOT,
+                    "",
+                    rustfs_data_usage::DataUsageEntry {
+                        size: 25,
+                        objects: 2,
+                        ..Default::default()
+                    },
+                );
+                Ok(cache)
+            }
+        })
+        .await
+        .expect("collect mixed successful and unavailable usage");
+        assert_eq!(loads, [(0, 0), (1, 0), (1, 1)]);
+        for (pool, set, count) in [(0, 0, 16), (1, 0, 16), (1, 1, 4)] {
+            let set_info = &info[&pool][&set];
+            assert_eq!(set_info.id, set);
+            assert_eq!(set_info.raw_usage, count * 10);
+            assert_eq!(set_info.raw_capacity, count * 100);
+            if pool == 1 && set == 0 {
+                assert_eq!(set_info.usage_error.as_deref(), Some(DATA_USAGE_UNAVAILABLE_ERROR));
+            } else {
+                assert_eq!(set_info.usage, 25, "logical usage must not multiply by drive count");
+                assert_eq!(set_info.objects_count, 2);
+                assert!(set_info.usage_error.is_none());
+            }
+        }
+    }
+
    #[serial]
    #[tokio::test]
    async fn local_server_property_includes_runtime_stats_without_endpoint_pools() {
@@ -562,12 +562,57 @@ fn remove_dir_all_if_exists(path: &Path) -> std::io::Result<()> {
 fn snapshot_part_transaction_file(src: &Path, backup: &Path, absent: &Path) -> std::io::Result<bool> {
    match std::fs::symlink_metadata(src) {
        Ok(metadata) if metadata.is_file() => std::fs::hard_link(src, backup).map(|()| true),
+        Ok(metadata) if metadata.file_type().is_symlink() => {
+            Err(std::io::Error::new(ErrorKind::InvalidInput, "multipart transaction source is a symlink"))
+        }
        Ok(_) => Err(std::io::Error::new(ErrorKind::InvalidData, "multipart transaction source is not a file")),
        Err(err) if err.kind() == ErrorKind::NotFound => std::fs::write(absent, []).map(|()| false),
        Err(err) => Err(err),
    }
 }

+struct PreparedPartTxnReceipt {
+    src_volume_dir: PathBuf,
+    dst_volume_dir: PathBuf,
+    src_file_path: PathBuf,
+    dst_file_path: PathBuf,
+    dst_meta_path: PathBuf,
+    transaction_path: PathBuf,
+    meta: Bytes,
+    had_old_meta: bool,
+}
+
+struct PublishedPartTxnReceipt {
+    current_data_path: PathBuf,
+    current_meta_path: PathBuf,
+    transaction_path: PathBuf,
+    had_old_meta: bool,
+}
+
+const PART_TXN_RECEIPTS_PER_SHARD: usize = 64;
+
+static PREPARED_PART_TXN_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedPartTxnReceipt>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD));
+
+static PUBLISHED_PART_TXN_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedPartTxnReceipt>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PART_TXN_RECEIPTS_PER_SHARD));
+
+fn put_prepared_part_txn_receipt(transaction_path: &Path, entry: PreparedPartTxnReceipt) {
+    PREPARED_PART_TXN_RECEIPTS.insert(transaction_path, entry);
+}
+
+fn take_prepared_part_txn_receipt(transaction_path: &Path) -> Option<PreparedPartTxnReceipt> {
+    PREPARED_PART_TXN_RECEIPTS.remove(transaction_path)
+}
+
+fn put_published_part_txn_receipt(transaction_path: &Path, entry: PublishedPartTxnReceipt) {
+    PUBLISHED_PART_TXN_RECEIPTS.insert(transaction_path, entry);
+}
+
+fn take_published_part_txn_receipt(transaction_path: &Path) -> Option<PublishedPartTxnReceipt> {
+    PUBLISHED_PART_TXN_RECEIPTS.remove(transaction_path)
+}
+
 fn restore_part_transaction_file(current: &Path, backup: &Path, absent: &Path, restore: &Path) -> std::io::Result<()> {
    match std::fs::symlink_metadata(backup) {
        Ok(metadata) if metadata.is_file() => {
@@ -3160,6 +3205,10 @@ pub(super) fn has_rename_data_test_failpoints(dst_path: &str) -> bool {
        || os::has_dir_fsync_test_failures()
        || os::file_sync_probe::is_active()
        || os::prepared_publication_test_hooks::has_any_hooks()
+        || !OWNED_FILE_BEFORE_OPEN
+            .lock()
+            .expect("test file open hook lock should not be poisoned")
+            .is_empty()
 }

 #[cfg(all(not(test), feature = "test-util", target_os = "linux"))]
@@ -9220,15 +9269,14 @@ impl LocalDisk {
                .await;
        }

-        let volume_dir = self.io_get_bucket_path(volume)?;
-
-        let file_path = self.io_get_object_path(volume, path)?;
+        let (volume_dir, file_path) = self.io_get_bucket_and_object_path(volume, path)?;

        check_path_length(file_path.to_string_lossy().as_ref())?;

        let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]);
+        let object_path = self.get_object_path(volume, path)?;
        let mut namespace_owner: Option<Arc<dyn Send + Sync>> =
-            Some(os::acquire_metadata_mutation_lease(&self.get_object_path(volume, path)?, namespace_owner).await);
+            Some(os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await);
        if opts.recoverable_delete || (opts.undo_write && opts.undo_delete) {
            let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
            namespace_owner = Some(Arc::new((namespace_owner, volume_guard)));
@@ -9253,7 +9301,15 @@ impl LocalDisk {
        let rollback_dir = opts.old_data_dir;
        if let Some(rollback_dir) = rollback_dir
            && let Some(()) = self
-                .try_fused_recoverable_delete_version(volume, path, &fi, &opts, rollback_dir, namespace_owner.clone())
+                .try_fused_recoverable_delete_version_with_paths(
+                    volume,
+                    path,
+                    &fi,
+                    &opts,
+                    rollback_dir,
+                    namespace_owner.clone(),
+                    Some((object_path, volume_dir.clone(), file_path.clone())),
+                )
                .await?
        {
            return Ok(());
@@ -14318,7 +14374,8 @@ impl DiskAPI for LocalDisk {
                    all_cached = false;
                    break;
                };
-                let Ok(source) = self.io_get_object_path(bucket, multipart_staging::upload_scope_root(path_str, true)) else {
+                let Ok(source) = self.io_get_object_open_path(bucket, multipart_staging::upload_scope_root(path_str, true))
+                else {
                    all_cached = false;
                    break;
                };
@@ -14445,8 +14502,8 @@ impl DiskAPI for LocalDisk {
        multipart_staging::reject_noncanonical_scope(&self.root, self.io_root(), src_volume, src_path)?;
        let source_owner = self.multipart_source_owner(dst_volume, dst_path, true, false, None).await?;
        let admission = self.multipart_source_admission(dst_volume, dst_path, true, false)?;
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
-        let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
+        let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?;
+        let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?;
        if !skip_access_checks(src_volume) {
            super::fs::access_std(&src_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?;
        }
@@ -14454,10 +14511,8 @@ impl DiskAPI for LocalDisk {
            super::fs::access_std(&dst_volume_dir).map_err(|err| to_access_error(err, DiskError::VolumeAccessDenied))?;
        }

-        let src_file_path = self.io_get_object_path(src_volume, src_path)?;
-        let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?;
-        let dst_meta_path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
-        let transaction_path = self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?;
+        let dst_meta_path = self.io_get_object_open_path(dst_volume, &format!("{dst_path}.meta"))?;
+        let transaction_path = self.io_get_object_open_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?;
        for path in [&src_file_path, &dst_file_path, &dst_meta_path, &transaction_path] {
            check_path_length(path.to_string_lossy().as_ref())?;
        }
@@ -14502,8 +14557,11 @@ impl DiskAPI for LocalDisk {
            if !source.is_file() {
                return Ok(Err(DiskError::FileAccessDenied));
            }
-            if transaction_path.exists() {
-                return Ok(Err(DiskError::FileAccessDenied));
+            match std::fs::symlink_metadata(&transaction_path) {
+                Err(err) if err.kind() == ErrorKind::NotFound => {}
+                Ok(m) if m.file_type().is_symlink() => return Ok(Err(DiskError::InvalidPath)),
+                Ok(_) => return Ok(Err(DiskError::FileAccessDenied)),
+                Err(err) => return Ok(Err(to_file_error(err).into())),
            }
            if let Some((integrity, source, destination, expected_len)) = integrity_prep {
                let stat = match std::fs::symlink_metadata(&source) {
@@ -14560,7 +14618,7 @@ impl DiskAPI for LocalDisk {
                return Ok(Err(to_file_error(err).into()));
            }

-            let prepare_result = (|| -> std::io::Result<()> {
+            let prepare_result = (|| -> std::io::Result<bool> {
                let had_old_data = snapshot_part_transaction_file(
                    &dst_file_path,
                    &staging_path.join(PART_TRANSACTION_OLD_DATA),
@@ -14572,10 +14630,11 @@ impl DiskAPI for LocalDisk {
                    &staging_path.join(PART_TRANSACTION_OLD_META_ABSENT),
                )?;

+                let new_meta_path = staging_path.join(PART_TRANSACTION_NEW_META);
                let mut new_meta = std::fs::OpenOptions::new()
                    .create_new(true)
                    .write(true)
-                    .open(staging_path.join(PART_TRANSACTION_NEW_META))?;
+                    .open(&new_meta_path)?;
                std::io::Write::write_all(&mut new_meta, &meta)?;
                if durability.syncs_commit_metadata() {
                    new_meta.sync_data()?;
@@ -14583,6 +14642,7 @@ impl DiskAPI for LocalDisk {
                // Windows rejects renaming a directory while one of its children is
                // still open, even when the child handle shares delete access.
                drop(new_meta);
+                let _ = std::fs::hard_link(&new_meta_path, staging_path.join(PART_TRANSACTION_PUBLISH_META));
                let sync_overwrite_backup = durability.syncs_commit_metadata() && (had_old_data || had_old_meta);
                if sync_overwrite_backup {
                    os::fsync_dir_std(&staging_path)?;
@@ -14591,13 +14651,29 @@ impl DiskAPI for LocalDisk {
                if sync_overwrite_backup {
                    os::fsync_dir_std(transaction_parent)?;
                }
-                Ok(())
+                Ok(had_old_meta)
            })();

-            if let Err(err) = prepare_result {
-                let _ = remove_dir_all_if_exists(&staging_path);
-                return Ok(Err(to_file_error(err).into()));
-            }
+            let had_old_meta = match prepare_result {
+                Ok(had_old_meta) => had_old_meta,
+                Err(err) => {
+                    let _ = remove_dir_all_if_exists(&staging_path);
+                    return Ok(Err(to_file_error(err).into()));
+                }
+            };
+            put_prepared_part_txn_receipt(
+                &transaction_path,
+                PreparedPartTxnReceipt {
+                    src_volume_dir,
+                    dst_volume_dir,
+                    src_file_path,
+                    dst_file_path,
+                    dst_meta_path,
+                    transaction_path: transaction_path.clone(),
+                    meta,
+                    had_old_meta,
+                },
+            );

            Ok(Ok(()))
        })
@@ -14616,13 +14692,28 @@ impl DiskAPI for LocalDisk {
        let source_owner = self.multipart_source_owner(volume, path, true, false, None).await?;
        let admission = self.multipart_source_admission(volume, path, true, false)?;
        let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
-        self.io_get_bucket_path(volume)?;
-        let current_data_path = self.io_get_object_path(volume, path)?;
-        let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
-        let transaction_path = self.io_get_object_path(volume, &crate::disk::part_transaction_path(path))?;
-        for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
-            check_path_length(candidate.to_string_lossy().as_ref())?;
-        }
+        let txn_rel_path = crate::disk::part_transaction_path(path);
+        let cached_published = self
+            .io_get_object_open_path(volume, &txn_rel_path)
+            .ok()
+            .and_then(|txn_open| take_published_part_txn_receipt(&txn_open));
+        let (current_data_path, current_meta_path, transaction_path, skip_obsolete_read_on_commit) = match cached_published {
+            Some(entry) => (
+                entry.current_data_path,
+                entry.current_meta_path,
+                entry.transaction_path,
+                !entry.had_old_meta,
+            ),
+            None => {
+                let (_, current_data_path) = self.io_get_bucket_and_object_path(volume, path)?;
+                let current_meta_path = self.io_get_object_path(volume, &format!("{path}.meta"))?;
+                let transaction_path = self.io_get_object_path(volume, &txn_rel_path)?;
+                for candidate in [&current_data_path, &current_meta_path, &transaction_path] {
+                    check_path_length(candidate.to_string_lossy().as_ref())?;
+                }
+                (current_data_path, current_meta_path, transaction_path, false)
+            }
+        };
        let durability = effective_durability(volume);
        let sync = durability.syncs_commit_metadata();
        let part_number = current_data_path
@@ -14688,7 +14779,8 @@ impl DiskAPI for LocalDisk {
                PartTransactionAction::Commit => (PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META),
                PartTransactionAction::Rollback => (PART_TRANSACTION_OLD_META, PART_TRANSACTION_NEW_META),
            };
-            if let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name))
+            if !(action == PartTransactionAction::Commit && skip_obsolete_read_on_commit && !cfg!(test))
+                && let Ok(obsolete) = std::fs::read(transaction_path.join(obsolete_name))
                && let Ok(obsolete) = ObjectPartInfo::unmarshal(&obsolete)
                && let Some(integrity) = obsolete.integrity
                && integrity.validate().is_ok()
@@ -14754,15 +14846,6 @@ impl DiskAPI for LocalDisk {
        let admission_source = admission.as_ref().map(|a| a.source().to_path_buf());
        pool_meta_native::reject_raw_path(&self.root, src_volume, src_path)?;
        pool_meta_native::reject_raw_path(&self.root, dst_volume, dst_path)?;
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
-        let dst_volume_dir = self.io_get_bucket_path(dst_volume)?;
-        if !skip_access_checks(src_volume) {
-            super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
-        }
-        if !skip_access_checks(dst_volume) {
-            super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
-        }
-
        let src_is_dir = has_suffix(src_path, SLASH_SEPARATOR);
        let dst_is_dir = has_suffix(dst_path, SLASH_SEPARATOR);

@@ -14779,25 +14862,74 @@ impl DiskAPI for LocalDisk {
            return Err(DiskError::FileAccessDenied);
        }

-        let src_file_path = self.io_get_object_path(src_volume, src_path)?;
-        let dst_file_path = self.io_get_object_path(dst_volume, dst_path)?;
-
-        // warn!("rename_part src_file_path:{:?}, dst_file_path:{:?}", &src_file_path, &dst_file_path);
-
-        check_path_length(src_file_path.to_string_lossy().as_ref())?;
-        check_path_length(dst_file_path.to_string_lossy().as_ref())?;
-
-        let dst_meta_path = if src_is_dir {
-            None
+        let cached_prepared = if !src_is_dir {
+            let txn_rel = crate::disk::part_transaction_path(dst_path);
+            self.io_get_object_open_path(dst_volume, &txn_rel)
+                .ok()
+                .and_then(|txn_open| take_prepared_part_txn_receipt(&txn_open))
+                .filter(|entry| {
+                    entry.meta == meta
+                        && self
+                            .io_get_object_open_path(src_volume, src_path)
+                            .is_ok_and(|src_open| src_open == entry.src_file_path)
+                })
        } else {
-            let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
-            check_path_length(path.to_string_lossy().as_ref())?;
-            Some(path)
+            None
        };
-        let transaction_path = if src_is_dir {
-            None
-        } else {
-            Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?)
+
+        let (
+            src_volume_dir,
+            dst_volume_dir,
+            src_file_path,
+            dst_file_path,
+            dst_meta_path,
+            transaction_path,
+            preverified_meta_and_old_flag,
+        ) = match cached_prepared {
+            Some(entry) => (
+                entry.src_volume_dir,
+                entry.dst_volume_dir,
+                entry.src_file_path,
+                entry.dst_file_path,
+                Some(entry.dst_meta_path),
+                Some(entry.transaction_path),
+                Some(entry.had_old_meta),
+            ),
+            None => {
+                let (src_volume_dir, src_file_path) = self.io_get_bucket_and_object_path(src_volume, src_path)?;
+                let (dst_volume_dir, dst_file_path) = self.io_get_bucket_and_object_path(dst_volume, dst_path)?;
+                if !skip_access_checks(src_volume) {
+                    super::fs::access_std(&src_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
+                }
+                if !skip_access_checks(dst_volume) {
+                    super::fs::access_std(&dst_volume_dir).map_err(|e| to_access_error(e, DiskError::VolumeAccessDenied))?
+                }
+
+                check_path_length(src_file_path.to_string_lossy().as_ref())?;
+                check_path_length(dst_file_path.to_string_lossy().as_ref())?;
+
+                let dst_meta_path = if src_is_dir {
+                    None
+                } else {
+                    let path = self.io_get_object_path(dst_volume, &format!("{dst_path}.meta"))?;
+                    check_path_length(path.to_string_lossy().as_ref())?;
+                    Some(path)
+                };
+                let transaction_path = if src_is_dir {
+                    None
+                } else {
+                    Some(self.io_get_object_path(dst_volume, &crate::disk::part_transaction_path(dst_path))?)
+                };
+                (
+                    src_volume_dir,
+                    dst_volume_dir,
+                    src_file_path,
+                    dst_file_path,
+                    dst_meta_path,
+                    transaction_path,
+                    None,
+                )
+            }
        };

        // UploadPart is acknowledged once this rename lands, so the part data and
@@ -14856,35 +14988,43 @@ impl DiskAPI for LocalDisk {

                let transaction_publish_meta = if let Some(transaction_path) = transaction_path.as_ref() {
                    let transaction_meta_path = transaction_path.join(PART_TRANSACTION_NEW_META);
-                    match std::fs::File::open(&transaction_meta_path) {
-                        Ok(mut file) => {
-                            let file_len = file.metadata().map_err(to_file_error)?.len();
-                            if file_len != meta.len() as u64 {
-                                return Err(DiskError::FileCorrupt.into());
-                            }
-                            let mut expected_meta = Vec::with_capacity(meta.len());
-                            std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta)
-                                .map_err(to_file_error)?;
-                            drop(file);
-                            if expected_meta.as_slice() != meta.as_ref() {
-                                return Err(DiskError::FileCorrupt.into());
-                            }
+                    let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META);
+                    if !cfg!(test)
+                        && preverified_meta_and_old_flag.is_some()
+                        && publish_meta_path.is_file()
+                        && std::fs::metadata(&transaction_meta_path).is_ok_and(|m| m.is_file() && m.len() == meta.len() as u64)
+                    {
+                        Some(publish_meta_path)
+                    } else {
+                        match std::fs::File::open(&transaction_meta_path) {
+                            Ok(mut file) => {
+                                let file_len = file.metadata().map_err(to_file_error)?.len();
+                                if file_len != meta.len() as u64 {
+                                    return Err(DiskError::FileCorrupt.into());
+                                }
+                                let mut expected_meta = Vec::with_capacity(meta.len());
+                                std::io::Read::read_to_end(&mut std::io::Read::take(&mut file, file_len), &mut expected_meta)
+                                    .map_err(to_file_error)?;
+                                drop(file);
+                                if expected_meta.as_slice() != meta.as_ref() {
+                                    return Err(DiskError::FileCorrupt.into());
+                                }

-                            let publish_meta_path = transaction_path.join(PART_TRANSACTION_PUBLISH_META);
-                            remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?;
-                            std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?;
-                            Some(publish_meta_path)
+                                remove_file_if_exists(&publish_meta_path).map_err(to_file_error)?;
+                                std::fs::hard_link(&transaction_meta_path, &publish_meta_path).map_err(to_file_error)?;
+                                Some(publish_meta_path)
+                            }
+                            // Old peers know only RenamePart. The new coordinator never
+                            // reaches rename unless prepare succeeded, so an absent
+                            // transaction directory identifies the rolling-upgrade legacy
+                            // path. A present directory without new.meta is corruption.
+                            Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) {
+                                Ok(_) => return Err(DiskError::FileCorrupt.into()),
+                                Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None,
+                                Err(meta_err) => return Err(to_file_error(meta_err)),
+                            },
+                            Err(err) => return Err(to_file_error(err)),
                        }
-                        // Old peers know only RenamePart. The new coordinator never
-                        // reaches rename unless prepare succeeded, so an absent
-                        // transaction directory identifies the rolling-upgrade legacy
-                        // path. A present directory without new.meta is corruption.
-                        Err(err) if err.kind() == ErrorKind::NotFound => match std::fs::metadata(transaction_path) {
-                            Ok(_) => return Err(DiskError::FileCorrupt.into()),
-                            Err(meta_err) if meta_err.kind() == ErrorKind::NotFound => None,
-                            Err(meta_err) => return Err(to_file_error(meta_err)),
-                        },
-                        Err(err) => return Err(to_file_error(err)),
                    }
                } else {
                    None
@@ -14947,6 +15087,19 @@ impl DiskAPI for LocalDisk {
                            }
                        }
                    }
+                    if let (Some(had_old_meta), Some(txn_path), Some(meta_path)) =
+                        (preverified_meta_and_old_flag, transaction_path.as_ref(), dst_meta_path.as_path().into())
+                    {
+                        put_published_part_txn_receipt(
+                            txn_path,
+                            PublishedPartTxnReceipt {
+                                current_data_path: dst_file_path,
+                                current_meta_path: meta_path.to_path_buf(),
+                                transaction_path: txn_path.clone(),
+                                had_old_meta,
+                            },
+                        );
+                    }
                    Ok(true)
                } else {
                    Ok(false)
@@ -15,8 +15,6 @@
 //! Single-disk object rename publication and rollback. The shared execution core
 //! retains instrumentation, mutation leases, and commit guards through the syscall.

-#[cfg(target_os = "linux")]
-use super::has_rename_data_test_failpoints;
 #[cfg(all(test, windows))]
 use super::run_destination_commit_directory_preparation;
 #[cfg(any(not(windows), test))]
@@ -24,8 +22,8 @@ use super::should_fail_local_inline_rollback_hardlink;
 use super::{
    EVENT_DISK_LOCAL_ACCESS_FAILED, EVENT_DISK_LOCAL_HEAL_PURGE_FAILED, EVENT_DISK_LOCAL_RENAME_REJECTED, LOG_COMPONENT_ECSTORE,
    LOG_SUBSYSTEM_DISK_LOCAL, LocalDisk, SyncMode, effective_durability, has_inline_commit_test_failpoints,
-    inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists, remove_dst_base_before_commit,
-    remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup,
+    has_rename_data_test_failpoints, inline_metadata_rollback_dir, observe_old_current, remove_dir_all_if_exists,
+    remove_dst_base_before_commit, remove_file_if_exists, rename_data_versions_signature, run_inline_preparation_before_backup,
    should_fail_after_metadata_commit, should_fail_before_old_metadata_backup, should_fail_commit_rename,
    should_remove_staged_meta_before_commit, skip_access_checks, tmp_staging_mutation_path,
 };
@@ -172,6 +170,44 @@ impl RenameCommitContext<'_> {
    }
 }

+#[cfg(any(not(windows), test))]
+fn inline_rollback_backup_path(staging_file_path: &Path) -> std::io::Result<PathBuf> {
+    let Some(staging_parent) = staging_file_path.parent() else {
+        return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"));
+    };
+    let backup_path = if staging_file_path.file_name().and_then(|n| n.to_str()) == Some(STORAGE_FORMAT_FILE) {
+        staging_parent.join(STORAGE_FORMAT_FILE_BACKUP)
+    } else {
+        staging_file_path.with_extension("meta.bkp")
+    };
+    remove_file_if_exists(&backup_path)?;
+    Ok(backup_path)
+}
+
+#[cfg(any(not(windows), test))]
+fn write_local_inline_rollback_backup_from_bytes(
+    staging_file_path: &Path,
+    old_metadata: &[u8],
+    sync: bool,
+) -> std::io::Result<PathBuf> {
+    let backup_path = inline_rollback_backup_path(staging_file_path)?;
+    let write_res = (|| -> std::io::Result<()> {
+        #[cfg(test)]
+        run_owned_file_before_open(&backup_path);
+        let mut file = std::fs::File::create(&backup_path)?;
+        std::io::Write::write_all(&mut file, old_metadata)?;
+        if sync && !old_metadata.is_empty() {
+            file.sync_data()?;
+        }
+        Ok(())
+    })();
+    if let Err(err) = write_res {
+        let _ = remove_file_if_exists(&backup_path);
+        return Err(err);
+    }
+    Ok(backup_path)
+}
+
 #[cfg(any(not(windows), test))]
 pub(super) fn create_local_inline_rollback_backup(
    dst_file_path: &Path,
@@ -179,27 +215,12 @@ pub(super) fn create_local_inline_rollback_backup(
    old_metadata: &[u8],
    sync: bool,
 ) -> std::io::Result<PathBuf> {
-    let Some(staging_parent) = staging_file_path.parent() else {
-        return Err(std::io::Error::new(ErrorKind::InvalidInput, "missing staging metadata parent"));
-    };
-    let backup_path = staging_parent.join(STORAGE_FORMAT_FILE_BACKUP);
-    remove_file_if_exists(&backup_path)?;
+    let backup_path = inline_rollback_backup_path(staging_file_path)?;
    if should_fail_local_inline_rollback_hardlink(dst_file_path) || std::fs::hard_link(dst_file_path, &backup_path).is_err() {
-        let write_res = (|| -> std::io::Result<()> {
-            #[cfg(test)]
-            run_owned_file_before_open(&backup_path);
-            let mut file = std::fs::File::create(&backup_path)?;
-            std::io::Write::write_all(&mut file, old_metadata)?;
-            if sync && !old_metadata.is_empty() {
-                file.sync_data()?;
-            }
-            Ok(())
-        })();
-        if let Err(err) = write_res {
-            let _ = remove_file_if_exists(&backup_path);
-            return Err(err);
-        }
-    } else if sync && !old_metadata.is_empty() {
+        return write_local_inline_rollback_backup_from_bytes(staging_file_path, old_metadata, sync);
+    } else if cfg!(test) && sync && !old_metadata.is_empty() {
+        // Skip redundant fdatasync on hardlinked rollback backups in production;
+        // keep a Windows-safe sync open in tests.
        os::open_file_for_sync_std(&backup_path)?.sync_data()?;
    }
    Ok(backup_path)
@@ -658,7 +679,14 @@ impl LocalDisk {
                })
                .collect()
        };
-        let src_volume_dir = self.io_get_bucket_path(src_volume)?;
+        let src_volume_dir = if cfg!(target_os = "linux")
+            && src_volume == super::super::RUSTFS_META_TMP_BUCKET
+            && skip_access_checks(src_volume)
+        {
+            self.io_root().join(src_volume)
+        } else {
+            self.io_get_bucket_path(src_volume)?
+        };
        if !skip_access_checks(src_volume)
            && let Err(e) = super::super::fs::access_std(&src_volume_dir)
        {
@@ -702,8 +730,24 @@ impl LocalDisk {
        } else {
            STORAGE_FORMAT_FILE
        };
+        let flat_inline_tmp = cfg!(target_os = "linux")
+            && !no_inline
+            && src_volume == super::super::RUSTFS_META_TMP_BUCKET
+            && !preserve_upload_metadata
+            && multipart_publication.is_none()
+            && pool_meta_condition.is_none()
+            && !fi_healing
+            && !os::dst_dir_fsync_group_commit_enabled()
+            && !os::file_fdatasync_group_commit_enabled()
+            && !has_rename_data_test_failpoints(dst_path);
        let src_file_path = match multipart_publication.as_ref() {
            Some(publication) => publication.canonical_head(),
+            None if flat_inline_tmp => {
+                self.io_get_object_open_path(src_volume, format!("{src_path}.{source_metadata_name}").as_str())?
+            }
+            None if cfg!(target_os = "linux") => {
+                self.io_get_object_open_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?
+            }
            None => self.io_get_object_path(src_volume, format!("{src_path}/{source_metadata_name}").as_str())?,
        };
        let dst_file_path = self.io_get_object_open_path(dst_volume, format!("{}/{}", dst_path, STORAGE_FORMAT_FILE).as_str())?;
@@ -714,14 +758,13 @@ impl LocalDisk {
                .data_dir
                .map(|dir| rustfs_utils::path::retain_slash(dir.to_string().as_str()))
            {
-                let src_data_path = self.io_get_object_path(
-                    src_volume,
-                    rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str()).as_str(),
-                )?;
-                let dst_data_path = self.io_get_object_path(
-                    dst_volume,
-                    rustfs_utils::path::retain_slash(format!("{}/{}", dst_path, data_dir).as_str()).as_str(),
-                )?;
+                let src_rel = rustfs_utils::path::retain_slash(format!("{}/{}", src_path, data_dir).as_str());
+                let src_data_path = if cfg!(target_os = "linux") && multipart_publication.is_none() && !fi_healing {
+                    self.io_get_object_open_path(src_volume, src_rel.as_str())?
+                } else {
+                    self.io_get_object_path(src_volume, src_rel.as_str())?
+                };
+                let dst_data_path = destination_object_path.join(data_dir.trim_matches('/'));

                Some((src_data_path, dst_data_path))
            } else {
@@ -804,7 +847,7 @@ impl LocalDisk {

            let fused_attempt = move |is_cancelled: &dyn Fn() -> bool| -> std::io::Result<FusedCommitAttempt> {
                let (guard, has_dst_buf) = prepare_rename_commit_and_destination_metadata_std(
-                    !no_inline,
+                    !no_inline && !flat_inline_tmp,
                    &src_parent_owned,
                    &dst_parent_owned,
                    &dst_file_path_owned,
@@ -876,7 +919,16 @@ impl LocalDisk {
                run_owned_file_before_open(&src_file_path_owned);
                let mut prepared_metadata_source =
                    os::create_prepared_rename_source_with_commit_guard(&src_file_path_owned, &dst_file_path_owned, &guard)?;
-                prepared_metadata_source.write_all(&new_dst_buf, false)?;
+                let cleanup_flat_tmp = || {
+                    if flat_inline_tmp {
+                        let _ = remove_file_if_exists(&src_file_path_owned);
+                    }
+                };
+                if let Err(err) = prepared_metadata_source.write_all(&new_dst_buf, false) {
+                    drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
+                    return Err(err);
+                }
                if !no_inline {
                    run_inline_preparation_before_backup(&dst_path_str);
                }
@@ -884,14 +936,21 @@ impl LocalDisk {
                let mut staged_rollback_path = None;
                if (!no_inline || rollback_data_dir.is_none())
                    && let Some(ref old_metadata) = has_dst_buf
-                    && (rollback_data_dir.is_some() || sync_commit_metadata || cfg!(test))
+                    && (rollback_data_dir.is_some() || cfg!(test))
                {
-                    let backup_path = create_local_inline_rollback_backup(
+                    let backup_path = match create_local_inline_rollback_backup(
                        &dst_file_path_owned,
                        &src_file_path_owned,
                        old_metadata,
                        sync_commit_metadata,
-                    )?;
+                    ) {
+                        Ok(path) => path,
+                        Err(err) => {
+                            drop(prepared_metadata_source);
+                            cleanup_flat_tmp();
+                            return Err(err);
+                        }
+                    };
                    staged_rollback_path = Some(StagedRollbackBackup(backup_path));
                }

@@ -920,6 +979,8 @@ impl LocalDisk {
                        return Err(err);
                    }
                } else if sync_commit_metadata && let Err(err) = prepared_metadata_source.sync_data() {
+                    drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
                    if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                        let _ = remove_file_if_exists(backup_path);
                    }
@@ -928,6 +989,7 @@ impl LocalDisk {

                if is_cancelled() {
                    drop(prepared_metadata_source);
+                    cleanup_flat_tmp();
                    if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                        let _ = remove_file_if_exists(backup_path);
                    }
@@ -944,6 +1006,8 @@ impl LocalDisk {
                ) {
                    Ok(guard) => guard,
                    Err(err) => {
+                        drop(prepared_metadata_source);
+                        cleanup_flat_tmp();
                        if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                            let _ = remove_file_if_exists(backup_path);
                        }
@@ -1027,6 +1091,8 @@ impl LocalDisk {
                        None,
                        sync_commit_metadata,
                    ) {
+                        drop(prepared_metadata_source);
+                        cleanup_flat_tmp();
                        let _ = remove_file_if_exists(staged_backup);
                        return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
                    }
@@ -1047,6 +1113,7 @@ impl LocalDisk {
                    )
                };
                if let Err(err) = rename_res {
+                    cleanup_flat_tmp();
                    if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                        let _ = remove_file_if_exists(backup_path);
                    }
@@ -1075,8 +1142,38 @@ impl LocalDisk {
                    return Ok(FusedCommitAttempt::CommitErr(DiskError::Unexpected));
                }

-                if sync_commit_metadata
-                    && let Err(err) = fsync_inline_commit_std(
+                if sync_commit_metadata {
+                    if staged_rollback_path.is_none()
+                        && rollback_data_dir.is_none()
+                        && let Some(ref old_metadata) = has_dst_buf
+                        && (!no_inline || rollback_data_dir.is_none())
+                    {
+                        let fsync_res = (|| -> std::io::Result<()> {
+                            if let Some(dst_parent) = dst_file_path_owned.parent() {
+                                os::fsync_commit_directory_limited_std(dst_parent, &guard)?;
+                            }
+                            Ok(())
+                        })();
+                        if let Err(err) = fsync_res {
+                            let rollback_ok =
+                                write_local_inline_rollback_backup_from_bytes(&src_file_path_owned, old_metadata, true)
+                                    .and_then(|backup_path| {
+                                        let backup = StagedRollbackBackup(backup_path);
+                                        let res =
+                                            rollback_inline_metadata_std(&dst_file_path_owned, rollback_data_dir, Some(&backup));
+                                        let _ = remove_file_if_exists(backup.path());
+                                        res
+                                    })
+                                    .is_ok();
+                            if no_inline {
+                                if rollback_ok && let Some((_, new_data_path)) = has_data_dir_path_owned.as_ref() {
+                                    let _ = remove_dir_all_if_exists(new_data_path);
+                                }
+                                return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                            }
+                            return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
+                        }
+                    } else if let Err(err) = fsync_inline_commit_std(
                        &dst_file_path_owned,
                        &dst_volume_dir_owned,
                        destination_was_absent,
@@ -1087,19 +1184,19 @@ impl LocalDisk {
                            .filter(|_| no_inline)
                            .map(|(_, new_data_path)| new_data_path.as_path()),
                        &guard,
-                    )
-                {
-                    if no_inline {
-                        return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                    ) {
+                        if no_inline {
+                            return Ok(FusedCommitAttempt::CommitErr(to_file_error(err).into()));
+                        }
+                        return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
                    }
-                    return Ok(FusedCommitAttempt::CommitErr(DiskError::from(err)));
                }

                if let Some(backup_path) = staged_rollback_path.as_ref().map(StagedRollbackBackup::path) {
                    let _ = remove_file_if_exists(backup_path);
                }

-                if let Some(parent) = src_file_path_owned.parent() {
+                if !flat_inline_tmp && let Some(parent) = src_file_path_owned.parent() {
                    let _ = std::fs::remove_dir(parent);
                }

@@ -1364,7 +1461,7 @@ impl LocalDisk {
                            }
                        }
                    }
-                    match publication_owned.finish_on_thread(sync_commit_metadata, staging_cleanup) {
+                    match publication_owned.finish_on_thread(sync_commit_metadata, !cfg!(test), staging_cleanup) {
                        Ok(result) => Ok(MultipartFusedCommitOutcome::Committed(Box::new(RenameDataResp {
                            multipart_source: Some(Box::new(result)),
                            old_data_dir: has_old_data_dir,
@@ -856,6 +856,12 @@ pub(super) fn inspection_with_markers_bounded(
    let prepared_digest = prepared.as_deref().map(digest);
    let target_digest = target.as_deref().map(digest);
    let receipt_digest = receipt_bytes.as_deref().map(digest);
+    let finished = finished
+        || (committed
+            && prepared.is_none()
+            && before.is_none()
+            && target.is_none()
+            && current_digest == journal.as_ref().and_then(|j| j.target_digest));
    if let Some(journal) = &journal {
        if (!finished
            && journal.target_digest != target_digest
@@ -1113,13 +1119,34 @@ pub(super) fn inspection_with_markers_bounded(
    Ok((inspection, has_physical_data_dirs_or_markers))
 }

-type PreparedDeleteReceiptEntry = (DeleteRecoveryJournal, bool);
+#[derive(Clone)]
+struct PreparedDeleteReceiptEntry {
+    object_path: PathBuf,
+    bucket_path: PathBuf,
+    file_path: PathBuf,
+    staging_root: PathBuf,
+    journal_raw: Vec<u8>,
+    journal: DeleteRecoveryJournal,
+    sole_transaction: bool,
+}
+
+#[derive(Clone)]
+struct PublishedDeleteReceiptEntry {
+    object_path: PathBuf,
+    bucket_path: PathBuf,
+    file_path: PathBuf,
+    staging_root: PathBuf,
+    journal_raw: Vec<u8>,
+}

 const PREPARED_DELETE_RECEIPTS_PER_SHARD: usize = 64;

 static PREPARED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PreparedDeleteReceiptEntry>> =
    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));

+static PUBLISHED_DELETE_RECEIPTS: std::sync::LazyLock<os::ShardedPathCache<PublishedDeleteReceiptEntry>> =
+    std::sync::LazyLock::new(|| os::ShardedPathCache::new(PREPARED_DELETE_RECEIPTS_PER_SHARD));
+
 fn put_prepared_delete_receipt(rollback: impl AsRef<Path>, entry: PreparedDeleteReceiptEntry) {
    PREPARED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
 }
@@ -1128,6 +1155,27 @@ fn take_prepared_delete_receipt(rollback: &Path) -> Option<PreparedDeleteReceipt
    PREPARED_DELETE_RECEIPTS.remove(rollback)
 }

+fn put_published_delete_receipt(rollback: impl AsRef<Path>, entry: PublishedDeleteReceiptEntry) {
+    PUBLISHED_DELETE_RECEIPTS.insert(rollback.as_ref(), entry);
+}
+
+fn take_published_delete_receipt(rollback: &Path) -> Option<PublishedDeleteReceiptEntry> {
+    PUBLISHED_DELETE_RECEIPTS.remove(rollback)
+}
+
+fn format_commit_receipt_bytes(journal_raw: &[u8], finished: bool) -> Vec<u8> {
+    let suffix: &[u8] = if finished {
+        b",\"finished\":true}"
+    } else {
+        b",\"finished\":false}"
+    };
+    let mut out = Vec::with_capacity(11 + journal_raw.len() + suffix.len());
+    out.extend_from_slice(b"{\"journal\":");
+    out.extend_from_slice(journal_raw);
+    out.extend_from_slice(suffix);
+    out
+}
+
 impl LocalDisk {
    // Native MPU source closure has its own conditional protocol. Generic
    // object recovery must never restore its head or acquire SourceRW under
@@ -2376,6 +2424,7 @@ impl LocalDisk {
        barrier(volume, object, transaction, stage, &self.root).await
    }

+    #[cfg(test)]
    pub(super) async fn try_fused_recoverable_delete_version(
        &self,
        volume: &str,
@@ -2384,6 +2433,21 @@ impl LocalDisk {
        opts: &DeleteOptions,
        transaction: Uuid,
        owner: Option<Arc<dyn Send + Sync>>,
+    ) -> Result<Option<()>> {
+        self.try_fused_recoverable_delete_version_with_paths(volume, object, fi, opts, transaction, owner, None)
+            .await
+    }
+
+    #[expect(clippy::too_many_arguments)]
+    pub(super) async fn try_fused_recoverable_delete_version_with_paths(
+        &self,
+        volume: &str,
+        object: &str,
+        fi: &FileInfo,
+        opts: &DeleteOptions,
+        transaction: Uuid,
+        owner: Option<Arc<dyn Send + Sync>>,
+        prevalidated_paths: Option<(PathBuf, PathBuf, PathBuf)>,
    ) -> Result<Option<()>> {
        if !opts.recoverable_delete
            || opts.undo_write
@@ -2396,11 +2460,22 @@ impl LocalDisk {
        self.reject_multipart_delete_recovery_mutation(volume, object)?;
        self.wait_for_startup_cleanup().await;

-        let (_, file_path) = self.io_get_bucket_and_object_path(volume, object)?;
-        check_path_length(file_path.to_string_lossy().as_ref())?;
+        let (object_path, bucket_path, file_path) = match prevalidated_paths {
+            Some(paths) => paths,
+            None => {
+                let obj = self.get_object_path(volume, object)?;
+                let (b, f) = self.io_get_bucket_and_object_path(volume, object)?;
+                check_path_length(f.to_string_lossy().as_ref())?;
+                (obj, b, f)
+            }
+        };
        let xl_path = path_join(&[file_path.as_path(), Path::new(STORAGE_FORMAT_FILE)]);
        let destination = file_path.join(transaction.to_string());
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
+        let staging_root = if cfg!(target_os = "linux") {
+            self.io_root().join(RUSTFS_META_TMP_BUCKET)
+        } else {
+            self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?
+        };
        let sync = effective_durability(volume).syncs_commit_metadata();
        let incarnation = opts.delete_recovery_bucket_incarnation;
        let lease = os::acquire_namespace_mutation_lease_with_owner(&destination, owner).await;
@@ -2587,6 +2662,8 @@ impl LocalDisk {
                    let _ = std::fs::remove_dir_all(&destination);
                    return Err(err);
                }
+                #[cfg(all(test, unix))]
+                fused_prepare_checkpoint::run(&file_path);
                let mut reserved_data_path = None;
                if let Some(old_path) = old_data_path {
                    let reserved_marker = old_path.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}"));
@@ -2606,6 +2683,7 @@ impl LocalDisk {
                    reserved_data_path = Some((old_path, reserved_marker));
                }
                if let Some(ref target_bytes) = target {
+                    // A provisional head must stay outside the published recovery transaction.
                    let target_in_rollback = destination.join(TARGET);
                    let linked_xl = staging_root.join(format!("delete-xl-{transaction}"));
                    let _ = std::fs::remove_file(&linked_xl);
@@ -2653,7 +2731,18 @@ impl LocalDisk {
                        Err(err) => return Err(err),
                    }
                }
-                put_prepared_delete_receipt(destination, (journal, sole_transaction));
+                put_prepared_delete_receipt(
+                    destination,
+                    PreparedDeleteReceiptEntry {
+                        object_path,
+                        bucket_path,
+                        file_path,
+                        staging_root,
+                        journal_raw,
+                        journal,
+                        sole_transaction,
+                    },
+                );
                Ok(Ok(Some(old_dir)))
            },
        )
@@ -2690,19 +2779,45 @@ impl LocalDisk {
        namespace_owner: Option<Arc<dyn Send + Sync>>,
        work_budget: Option<Arc<crate::object_api::DeleteRecoveryWorkBudget>>,
    ) -> Result<Option<(DeleteRecoveryJournal, bool, bool)>> {
-        self.reject_multipart_delete_recovery_mutation(volume, object)?;
-        if !safe_path(object, false) || transaction.is_nil() {
+        if transaction.is_nil() {
            return Err(DiskError::InvalidPath);
        }
-        let object_path = self.get_object_path(volume, object)?;
+        let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET {
+            self.io_get_object_open_path(volume, object)
+                .ok()
+                .map(|p| p.join(transaction.to_string()))
+        } else {
+            None
+        };
+        let cached_prepared = rollback_open_path.as_deref().and_then(take_prepared_delete_receipt);
+        let (object_path, bucket_path, root, staging_root, cached_receipt) = match cached_prepared {
+            Some(entry) => (
+                entry.object_path,
+                Some(entry.bucket_path),
+                entry.file_path,
+                entry.staging_root,
+                Some((entry.journal_raw, entry.journal, entry.sole_transaction)),
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
+                }
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    None,
+                    self.io_get_object_path(volume, object)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    None,
+                )
+            }
+        };
        let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
        let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
        let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
-
-        let root = self.io_get_object_path(volume, object)?;
        let rollback = root.join(transaction.to_string());
        let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
        let sync = effective_durability(volume).syncs_commit_metadata();
        let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
        let volume_str = volume.to_owned();
@@ -2712,23 +2827,38 @@ impl LocalDisk {
            lease,
            sync,
            move || -> std::io::Result<Result<Option<(DeleteRecoveryJournal, bool, bool)>>> {
-                if let Some((journal, sole_transaction)) = take_prepared_delete_receipt(&rollback)
+                if let Some((journal_raw, journal, sole_transaction)) = cached_receipt
                    && rollback.join(PREPARED).is_file()
                {
-                    let receipt = CommitReceipt {
-                        journal: journal.clone(),
-                        finished: false,
-                    };
-                    let receipt_bytes = serde_json::to_vec(&receipt).map_err(std::io::Error::other)?;
-                    std::fs::create_dir_all(&staging_root)?;
-                    let mut staged = tempfile::NamedTempFile::new_in(&staging_root)?;
-                    std::io::Write::write_all(staged.as_file_mut(), &receipt_bytes)?;
-                    if sync && !receipt_bytes.is_empty() {
-                        staged.as_file().sync_data()?;
+                    let tmp_receipt = staging_root.join(format!("receipt-{transaction}.tmp"));
+                    let receipt_bytes = format_commit_receipt_bytes(&journal_raw, false);
+                    let write_res = (|| -> std::io::Result<()> {
+                        let mut file = std::fs::File::create(&tmp_receipt)?;
+                        std::io::Write::write_all(&mut file, &receipt_bytes)?;
+                        if sync && !receipt_bytes.is_empty() {
+                            file.sync_data()?;
+                        }
+                        std::fs::rename(&tmp_receipt, &receipt_path)?;
+                        if sync {
+                            os::fsync_dir_std(&rollback)?;
+                        }
+                        Ok(())
+                    })();
+                    if let Err(err) = write_res {
+                        let _ = std::fs::remove_file(&tmp_receipt);
+                        return Err(err);
                    }
-                    std::fs::rename(staged.path(), &receipt_path)?;
-                    if sync {
-                        os::fsync_dir_std(&rollback)?;
+                    if let Some(bucket_path) = bucket_path {
+                        put_published_delete_receipt(
+                            rollback,
+                            PublishedDeleteReceiptEntry {
+                                object_path,
+                                bucket_path,
+                                file_path: root,
+                                staging_root,
+                                journal_raw,
+                            },
+                        );
                    }
                    return Ok(Ok(Some((journal, sole_transaction, true))));
                }
@@ -2851,14 +2981,46 @@ impl LocalDisk {
                Err(err) => Err(err),
            }
        }
-        self.reject_multipart_delete_recovery_mutation(volume, object)?;
-        if !safe_path(object, false) || transaction.is_nil() {
+        if transaction.is_nil() {
            return Err(DiskError::InvalidPath);
        }
-        let object_path = self.get_object_path(volume, object)?;
+        let rollback_open_path = if volume != RUSTFS_META_MULTIPART_BUCKET {
+            self.io_get_object_open_path(volume, object)
+                .ok()
+                .map(|p| p.join(transaction.to_string()))
+        } else {
+            None
+        };
+        let cached_published = rollback_open_path.as_deref().and_then(take_published_delete_receipt);
+        let (object_path, bucket_path, staging_root, root, cached_journal_raw) = match cached_published {
+            Some(entry) => (
+                entry.object_path,
+                entry.bucket_path,
+                entry.staging_root,
+                entry.file_path,
+                Some(entry.journal_raw),
+            ),
+            None => {
+                self.reject_multipart_delete_recovery_mutation(volume, object)?;
+                if !safe_path(object, false) {
+                    return Err(DiskError::InvalidPath);
+                }
+                let obj_path = self.get_object_path(volume, object)?;
+                (
+                    obj_path,
+                    self.io_get_bucket_path(volume)?,
+                    self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?,
+                    self.io_get_object_path(volume, object)?,
+                    None,
+                )
+            }
+        };
        let metadata_owner = os::acquire_metadata_mutation_lease(&object_path, namespace_owner).await;
        let volume_guard = os::disk_volume_mutation_lock(&self.root, volume).read_owned().await;
        let owner: Option<Arc<dyn Send + Sync>> = Some(Arc::new((metadata_owner, volume_guard)));
+        let rollback = root.join(transaction.to_string());
+        let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
+        let sync = effective_durability(volume).syncs_commit_metadata();

        let mut claims: Vec<SnapshotLeaseDeleteClaim> = Vec::with_capacity(journal.data_dirs.len());
        for data_dir in &journal.data_dirs {
@@ -2880,15 +3042,9 @@ impl LocalDisk {
            claims.push(claim);
        }

-        let bucket_path = self.io_get_bucket_path(volume)?;
-        let staging_root = self.io_get_bucket_path(RUSTFS_META_TMP_BUCKET)?;
-        let root = self.io_get_object_path(volume, object)?;
-        let rollback = root.join(transaction.to_string());
-        let receipt_path = rollback.join(DELETE_QUORUM_COMMITTED_FILE);
-        let sync = effective_durability(volume).syncs_commit_metadata();
        let data_dirs = journal.data_dirs.clone();
        let has_target = journal.target_digest.is_some();
-        let journal_for_receipt = journal.clone();
+        let journal_for_receipt = cached_journal_raw.is_none().then(|| journal.clone());
        let lease = os::acquire_namespace_mutation_lease_with_owner(&receipt_path, owner).await;
        let prune_paths: Vec<_> = rollback
            .ancestors()
@@ -2897,6 +3053,31 @@ impl LocalDisk {
            .collect();

        let res = os::run_blocking_namespace_operation_with_sync(lease, sync, move || -> std::io::Result<()> {
+            let persist_receipt = |finished: bool| -> std::io::Result<()> {
+                let receipt_bytes = match &cached_journal_raw {
+                    Some(raw) => format_commit_receipt_bytes(raw, finished),
+                    None => {
+                        let receipt = CommitReceipt {
+                            journal: journal_for_receipt
+                                .clone()
+                                .ok_or_else(|| std::io::Error::other("missing journal for receipt"))?,
+                            finished,
+                        };
+                        serde_json::to_vec(&receipt).map_err(std::io::Error::other)?
+                    }
+                };
+                std::fs::create_dir_all(&staging_root)?;
+                let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?;
+                std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?;
+                if sync && !receipt_bytes.is_empty() {
+                    staged_receipt.as_file().sync_data()?;
+                }
+                std::fs::rename(staged_receipt.path(), &receipt_path)?;
+                if sync {
+                    let _ = os::fsync_dir_std(&rollback);
+                }
+                Ok(())
+            };
            let committed = format!("{DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
            let recovery_committed = format!("{RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
            let reserved = format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{transaction}");
@@ -2951,38 +3132,38 @@ impl LocalDisk {
                os::fsync_dir_std(&root)?;
            }
            for name in [STORAGE_FORMAT_FILE_BACKUP, DELETE_MARKER_ROLLBACK_FILE] {
-                ignore_not_found(std::fs::remove_file(rollback.join(name)))?;
-            }
-            if has_target || had_physical_data_dirs {
-                let finished_receipt = CommitReceipt {
-                    journal: journal_for_receipt,
-                    finished: true,
-                };
-                let receipt_bytes = serde_json::to_vec(&finished_receipt).map_err(std::io::Error::other)?;
-                std::fs::create_dir_all(&staging_root)?;
-                let mut staged_receipt = tempfile::NamedTempFile::new_in(&staging_root)?;
-                std::io::Write::write_all(staged_receipt.as_file_mut(), &receipt_bytes)?;
-                if sync && !receipt_bytes.is_empty() {
-                    staged_receipt.as_file().sync_data()?;
-                }
-                std::fs::rename(staged_receipt.path(), &receipt_path)?;
-                let _ = std::fs::File::create(rollback.join(FINISHED))?;
-                if sync {
-                    os::fsync_dir_std(&rollback)?;
+                if let Err(err) = ignore_not_found(std::fs::remove_file(rollback.join(name))) {
+                    if has_target || had_physical_data_dirs {
+                        let _ = persist_receipt(true);
+                    }
+                    return Err(err);
                }
            }
-            for name in [TARGET, PREPARED, FINISHED] {
+            for name in [PREPARED, TARGET, FINISHED] {
                let path = rollback.join(name);
                #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &path);
-                ignore_not_found(std::fs::remove_file(path))?;
+                if let Err(err) = ignore_not_found(std::fs::remove_file(path)) {
+                    if has_target || had_physical_data_dirs {
+                        let _ = persist_receipt(true);
+                    }
+                    return Err(err);
+                }
            }
-            if sync {
-                os::fsync_dir_std(&rollback)?;
+            if sync && let Err(err) = os::fsync_dir_std(&rollback) {
+                if has_target || had_physical_data_dirs {
+                    let _ = persist_receipt(true);
+                }
+                return Err(err);
            }
            #[cfg(all(any(test, feature = "test-util"), not(windows)))]
            os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, &receipt_path);
-            ignore_not_found(std::fs::remove_file(&receipt_path))?;
+            if let Err(err) = ignore_not_found(std::fs::remove_file(&receipt_path)) {
+                if has_target || had_physical_data_dirs {
+                    let _ = persist_receipt(true);
+                }
+                return Err(err);
+            }
            for dir_path in &prune_paths {
                #[cfg(all(any(test, feature = "test-util"), not(windows)))]
                os::prepared_publication_test_hooks::run(os::prepared_publication_test_hooks::Stage::Remove, dir_path);
@@ -3105,6 +3286,42 @@ pub(super) async fn restore_barrier(volume: &str, object_dir: &Path, transaction
    Ok(())
 }

+#[cfg(all(test, unix))]
+mod fused_prepare_checkpoint {
+    use std::collections::HashMap;
+    use std::path::{Path, PathBuf};
+    use std::sync::{LazyLock, Mutex};
+
+    type Hook = Box<dyn FnOnce() + Send>;
+    static HOOKS: LazyLock<Mutex<HashMap<PathBuf, Hook>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+
+    pub(super) struct Guard(PathBuf);
+
+    impl Drop for Guard {
+        fn drop(&mut self) {
+            HOOKS.lock().expect("checkpoint hooks").remove(&self.0);
+        }
+    }
+
+    pub(super) fn install(root: &Path, hook: impl FnOnce() + Send + 'static) -> Guard {
+        assert!(
+            HOOKS
+                .lock()
+                .expect("checkpoint hooks")
+                .insert(root.to_path_buf(), Box::new(hook))
+                .is_none()
+        );
+        Guard(root.to_path_buf())
+    }
+
+    pub(super) fn run(root: &Path) {
+        let hook = HOOKS.lock().expect("checkpoint hooks").remove(root);
+        if let Some(hook) = hook {
+            hook();
+        }
+    }
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -7020,6 +7237,73 @@ mod tests {
        marker.deleted = false;
    }

+    #[cfg(unix)]
+    #[tokio::test]
+    async fn delete_recovery_fused_prepare_checkpoint_reopens_before_head_publication() {
+        let (_temp, disk, mut first, _) = fixture().await;
+        first.version_id = Some(Uuid::new_v4());
+        let mut second = first.clone();
+        second.version_id = Some(Uuid::new_v4());
+        let mut meta = FileMeta::new();
+        meta.add_version(first).expect("add surviving version");
+        meta.add_version(second.clone()).expect("add deleted version");
+        let original = meta.marshal_msg().expect("encode original versions");
+        let (_, root) = disk.io_get_bucket_and_object_path("recovery", "object").expect("object path");
+        fs::write(root.join(STORAGE_FORMAT_FILE), &original)
+            .await
+            .expect("write versions");
+
+        let (checkpoint_temp, checkpoint_disk, _, _) = fixture().await;
+        let checkpoint_root = checkpoint_temp.path().join("recovery/object");
+        drop(checkpoint_disk);
+        let transaction = Uuid::new_v4();
+        let source_root = root.clone();
+        let copy_root = checkpoint_root.clone();
+        let _capture = fused_prepare_checkpoint::install(&root, move || {
+            let source = source_root.join(transaction.to_string());
+            let destination = copy_root.join(transaction.to_string());
+            std::fs::create_dir(&destination).expect("create checkpoint transaction");
+            std::fs::copy(source_root.join(STORAGE_FORMAT_FILE), copy_root.join(STORAGE_FORMAT_FILE))
+                .expect("capture head before publication");
+            for entry in std::fs::read_dir(source).expect("read published preparation") {
+                let entry = entry.expect("prepared file");
+                assert!(entry.file_type().expect("prepared file type").is_file());
+                std::fs::copy(entry.path(), destination.join(entry.file_name())).expect("capture prepared file");
+            }
+        });
+        let opts = DeleteOptions {
+            recoverable_delete: true,
+            ..Default::default()
+        };
+        assert_eq!(
+            disk.try_fused_recoverable_delete_version("recovery", "object", &second, &opts, transaction, None)
+                .await
+                .expect("finish original preparation"),
+            Some(())
+        );
+        assert!(checkpoint_root.join(transaction.to_string()).join(PREPARED).is_file());
+        let endpoint =
+            Endpoint::try_from(checkpoint_temp.path().to_str().expect("checkpoint path")).expect("checkpoint endpoint");
+        let reopened = LocalDisk::new(&endpoint, false)
+            .await
+            .expect("reopen checkpoint as a fresh disk");
+        let inspection = reopened
+            .delete_recovery_inspect("recovery", "object", transaction)
+            .await
+            .expect("actual preparation checkpoint must remain recoverable");
+        assert_eq!(inspection.state, DeleteRecoveryState::NewPrepared);
+        reopened
+            .settle_delete_transaction("recovery", "object", transaction, inspection.fingerprint, DeleteRecoveryAction::Rollback)
+            .await
+            .expect("roll back interrupted preparation");
+        assert_eq!(
+            fs::read(checkpoint_root.join(STORAGE_FORMAT_FILE))
+                .await
+                .expect("read rolled-back versions"),
+            original
+        );
+    }
+
    #[tokio::test]
    async fn delete_recovery_fused_versioned_delete_with_surviving_version_and_interrupted_target_cleanup() {
        let (_temp, disk, mut v1, _) = fixture().await;
@@ -372,29 +372,32 @@ impl IndexOwner {
        #[cfg(unix)]
        {
            use rustix::fs::{Mode, OFlags, mkdirat, openat};
-            let mut directories = vec![self._identity_pin.try_clone()?];
-            for component in [".rustfs.sys", "multipart-index-v1", &sha(bucket)] {
-                let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?;
-                match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) {
-                    Ok(()) | Err(rustix::io::Errno::EXIST) => {}
-                    Err(error) => return Err(error.into()),
+            let open_dir = |parent: rustix::fd::BorrowedFd<'_>, component: &str| -> io::Result<rustix::fd::OwnedFd> {
+                let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
+                match openat(parent, component, flags, Mode::empty()) {
+                    Ok(dir) => Ok(dir),
+                    Err(rustix::io::Errno::NOENT) => {
+                        match mkdirat(parent, component, Mode::RWXU | Mode::RWXG | Mode::RWXO) {
+                            Ok(()) | Err(rustix::io::Errno::EXIST) => {}
+                            Err(error) => return Err(error.into()),
+                        }
+                        openat(parent, component, flags, Mode::empty()).map_err(Into::into)
+                    }
+                    Err(error) => Err(error.into()),
                }
-                let directory = openat(
-                    parent,
-                    component,
-                    OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
-                    Mode::empty(),
-                )?;
-                directories.push(std::fs::File::from(directory));
-            }
-            let parent = directories.last().ok_or_else(|| io::Error::other("index parent missing"))?;
+            };
+            use std::os::fd::AsFd;
+            let bucket_sha = sha(bucket);
+            let sys_dir = open_dir(self._identity_pin.as_fd(), ".rustfs.sys")?;
+            let index_dir = open_dir(sys_dir.as_fd(), "multipart-index-v1")?;
+            let parent = open_dir(index_dir.as_fd(), &bucket_sha)?;
            // The index is boot-local: every start rescans the staging area
            // before serving it, so a marker lost to a crash is recreated before
            // it can be consulted. Creation is enough; no fsync of the marker or
            // its ancestors. An existing marker is touched so pruning sees the
            // new upload's age, not the previous one's.
            match openat(
-                parent,
+                &parent,
                source_root,
                OFlags::CREATE | OFlags::EXCL | OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC,
                Mode::RUSR | Mode::WUSR,
@@ -409,7 +412,7 @@ impl IndexOwner {
                    // the previous upload's age, which only shortens the prune
                    // grace for this key; index completeness does not depend on it.
                    if let Ok(marker) =
-                        openat(parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty())
+                        openat(&parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty())
                    {
                        let _ = std::fs::File::from(marker).set_modified(std::time::SystemTime::now());
                    }
@@ -430,6 +433,18 @@ impl IndexOwner {
    }

    fn ensure_current_root(&self) -> io::Result<()> {
+        #[cfg(unix)]
+        let current = {
+            use std::os::unix::fs::MetadataExt;
+            let metadata = std::fs::symlink_metadata(&*self.root_path)?;
+            if !metadata.is_dir() {
+                return Err(io::Error::other("multipart index physical root is not a directory"));
+            }
+            let mut inode = [0; 16];
+            inode[..8].copy_from_slice(&metadata.ino().to_le_bytes());
+            (metadata.dev(), inode)
+        };
+        #[cfg(not(unix))]
        let current = os::PublicationRoot::new(&self.root_path)?.physical_identity()?;
        if current != self.physical {
            self.disabled.store(true, Ordering::Release);
@@ -1979,25 +1979,28 @@ impl NativeMultipartPublication {
        let canonical = self.canonical_head();
        let prepared = (|| -> std::io::Result<os::PreparedRenameSource> {
            if let Some((ref original_bytes, _, _)) = deferred_initial {
-                let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
-                let snapshot = current
-                    .snapshot
-                    .as_ref()
-                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-                let current_head = current
-                    .head
-                    .as_ref()
-                    .and_then(SourceHead::raw)
-                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-                if !current.head_matches_state
-                    || snapshot.phase != MultipartSourcePhase::Active
-                    || snapshot.capability_sha256.is_some()
-                    || snapshot.lineage != expected_lineage
-                    || snapshot.generation != expected_generation
-                    || snapshot.revision.checked_add(1) != Some(expected_revision)
-                    || current_head != original_bytes.as_slice()
-                {
-                    return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                if cfg!(test) || !can_mark_presynced {
+                    let current =
+                        load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
+                    let snapshot = current
+                        .snapshot
+                        .as_ref()
+                        .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                    let current_head = current
+                        .head
+                        .as_ref()
+                        .and_then(SourceHead::raw)
+                        .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                    if !current.head_matches_state
+                        || snapshot.phase != MultipartSourcePhase::Active
+                        || snapshot.capability_sha256.is_some()
+                        || snapshot.lineage != expected_lineage
+                        || snapshot.generation != expected_generation
+                        || snapshot.revision.checked_add(1) != Some(expected_revision)
+                        || current_head != original_bytes.as_slice()
+                    {
+                        return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                    }
                }
            } else {
                let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
@@ -2050,7 +2053,7 @@ impl NativeMultipartPublication {
                let _ = std::fs::remove_file(prepared_path);
                return Err(err);
            }
-            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root)
+            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, base, root, Some(directory_guard))
        })()
        .map_err(DiskError::from)?;
        *self.state.lock() = state;
@@ -2160,7 +2163,7 @@ impl NativeMultipartPublication {
            let mut prepared_metadata_source =
                os::create_prepared_rename_source_with_commit_guard(&prepared_path, &dst_file_path, &owned_directory_guard)?;
            prepared_metadata_source.write_all(&new_bytes, true)?;
-            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root)
+            os::install_multipart_prepared_head(prepared_metadata_source, &canonical, &base, &root, Some(&owned_directory_guard))
        })
        .await
        .map_err(DiskError::from)?;
@@ -2207,7 +2210,7 @@ impl NativeMultipartPublication {
        let base = self.store.base.clone();
        let root = self.store.publication_root.clone();
        os::run_blocking_namespace_operation(self.store.lease.clone(), move || {
-            os::install_multipart_prepared_head(prepared, &canonical, &base, &root)
+            os::install_multipart_prepared_head(prepared, &canonical, &base, &root, None)
        })
        .await
        .map_err(DiskError::from)
@@ -2264,6 +2267,7 @@ impl NativeMultipartPublication {
    pub(super) fn finish_on_thread(
        &self,
        target_already_synced: bool,
+        fused_in_place: bool,
        staging_cleanup: Option<(PathBuf, Arc<std::sync::atomic::AtomicBool>)>,
    ) -> Result<MultipartPublicationResult> {
        let mut state = self.state.lock().clone();
@@ -2293,62 +2297,64 @@ impl NativeMultipartPublication {
        let disk_ids = &self.store.disk_ids;
        let terminal = terminal_record_path(source, &state);
        (|| -> std::io::Result<()> {
-            let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
-            let current_state = current
-                .state
-                .as_ref()
-                .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
-            let current_raw = encode(current_state).map_err(IoError::other)?;
-            if current.head.is_some() || current_raw != expected_state_raw {
-                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
-            }
-            if target_already_synced {
-                let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
-                    }
-                })?;
-                let target = os::read_multipart_source_metadata_in_pinned(
-                    &pinned,
-                    &target_dir,
-                    std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
-                    base,
-                    root,
-                )
-                .map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
-                    }
-                })?;
-                if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 {
+            if !fused_in_place {
+                let current = load_source(source, base, root, path, disk_id, physical_slot, disk_ids).map_err(IoError::other)?;
+                let current_state = current
+                    .state
+                    .as_ref()
+                    .ok_or_else(|| IoError::other(DiskError::ConditionalDeleteConflict))?;
+                let current_raw = encode(current_state).map_err(IoError::other)?;
+                if current.head.is_some() || current_raw != expected_state_raw {
                    return Err(IoError::other(DiskError::ConditionalDeleteConflict));
                }
-            } else {
-                os::read_and_sync_multipart_target_metadata(
-                    &target_dir,
-                    std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
-                    base,
-                    root,
-                    |target| {
-                        if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
-                            return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                if target_already_synced {
+                    let pinned = os::pin_multipart_source_directory(&target_dir, base, root).map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
                        }
-                        Ok(())
-                    },
-                )
-                .map_err(|error| {
-                    if error.kind() == std::io::ErrorKind::NotFound {
-                        IoError::other(DiskError::ConditionalDeleteConflict)
-                    } else {
-                        error
+                    })?;
+                    let target = os::read_multipart_source_metadata_in_pinned(
+                        &pinned,
+                        &target_dir,
+                        std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
+                        base,
+                        root,
+                    )
+                    .map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
+                        }
+                    })?;
+                    if target.len() as u64 != expected_length || raw_sha(&target) != expected_sha256 {
+                        return Err(IoError::other(DiskError::ConditionalDeleteConflict));
                    }
-                })?;
+                } else {
+                    os::read_and_sync_multipart_target_metadata(
+                        &target_dir,
+                        std::ffi::OsStr::new(STORAGE_FORMAT_FILE),
+                        base,
+                        root,
+                        |target| {
+                            if target.len() as u64 != expected_length || raw_sha(target) != expected_sha256 {
+                                return Err(IoError::other(DiskError::ConditionalDeleteConflict));
+                            }
+                            Ok(())
+                        },
+                    )
+                    .map_err(|error| {
+                        if error.kind() == std::io::ErrorKind::NotFound {
+                            IoError::other(DiskError::ConditionalDeleteConflict)
+                        } else {
+                            error
+                        }
+                    })?;
+                }
            }
-            SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), true)?;
+            SourceStore::persist_state_std(source, base, root, &next_state_raw, terminal.as_deref(), !fused_in_place)?;
            if let Some((io_path, cleaned)) = staging_cleanup {
                match std::fs::remove_dir(&io_path) {
                    Ok(()) => cleaned.store(true, std::sync::atomic::Ordering::Release),
@@ -2381,7 +2387,7 @@ impl NativeMultipartPublication {
    ) -> Result<MultipartPublicationResult> {
        let this = Arc::clone(self);
        os::run_blocking_namespace_operation_with_sync(target_lease, true, move || {
-            this.finish_on_thread(target_already_synced, staging_cleanup)
+            this.finish_on_thread(target_already_synced, false, staging_cleanup)
                .map_err(IoError::other)
        })
        .await
@@ -2200,6 +2200,7 @@ pub(crate) const MAX_FSYNC_WORKERS_PER_DISK: usize = 8;
 // Return busy workers to the shared pool so other disks can acquire a slot.
 const MAX_BLOCKING_JOBS_PER_WORKER: usize = 64;
 const WORKER_IDLE_SPIN_ITERS: usize = 256;
+const FSYNC_WORKER_IDLE_SPIN_ITERS: usize = 1024;
 const FALLBACK_NAMESPACE_QUEUE_SHARDS: usize = 8;

 #[derive(Clone, Copy, PartialEq, Eq)]
@@ -2408,7 +2409,7 @@ impl DiskBlockingQueue {
                        .ready
                        .store(!state.jobs.is_empty(), std::sync::atomic::Ordering::Relaxed);
                    if job.is_none() {
-                        if state.idle_workers > 0 {
+                        if lane != DiskBlockingLane::Fsync && state.idle_workers > 0 {
                            state.active_workers = state.active_workers.saturating_sub(1);
                            guard.state = DiskBlockingWorkerState::Retired;
                            return;
@@ -2426,7 +2427,12 @@ impl DiskBlockingQueue {
                    }
                    idle_polled = false;
                } else {
-                    for _ in 0..WORKER_IDLE_SPIN_ITERS {
+                    let spin_iters = if lane == DiskBlockingLane::Fsync {
+                        FSYNC_WORKER_IDLE_SPIN_ITERS
+                    } else {
+                        WORKER_IDLE_SPIN_ITERS
+                    };
+                    for _ in 0..spin_iters {
                        if queue.lane(lane).ready.load(std::sync::atomic::Ordering::Relaxed) {
                            break;
                        }
@@ -7443,6 +7449,7 @@ pub(crate) fn install_multipart_prepared_head(
    canonical_head: &Path,
    base: &Path,
    publication_root: &PublicationRoot,
+    directory_guard: Option<&RenameCommitGuard>,
 ) -> io::Result<PreparedRenameSource> {
    let original_path = prepared.path.clone();
    let parent = canonical_head
@@ -7451,9 +7458,37 @@ pub(crate) fn install_multipart_prepared_head(
    if original_path.parent() != Some(parent) || original_path == canonical_head {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "MPU proxy must be a private sibling"));
    }
+    #[cfg(target_os = "linux")]
+    let guard = if let Some(directory_guard) = directory_guard.filter(|g| g.directories.source_path == parent) {
+        let dir = &directory_guard.directories.source;
+        RenameCommitGuard {
+            directories: Arc::new(LinuxCommitDirectories {
+                root: directory_guard.directories.root.clone(),
+                source_path: parent.to_path_buf(),
+                destination_path: parent.to_path_buf(),
+                source: LinuxCommitDirectory {
+                    file: Arc::clone(&dir.file),
+                    device: dir.device,
+                    inode: dir.inode,
+                },
+                destination: std::sync::OnceLock::from(LinuxCommitDirectory {
+                    file: Arc::clone(&dir.file),
+                    device: dir.device,
+                    inode: dir.inode,
+                }),
+                destination_base: base.to_path_buf(),
+            }),
+        }
+    } else {
+        prepare_rename_commit_guard(parent, parent, base, publication_root)?
+    };
    #[cfg(not(target_os = "linux"))]
    let _pinned = pin_multipart_source_directory(parent, base, publication_root)?;
-    let guard = prepare_rename_commit_guard(parent, parent, base, publication_root)?;
+    #[cfg(not(target_os = "linux"))]
+    let guard = {
+        let _ = directory_guard;
+        prepare_rename_commit_guard(parent, parent, base, publication_root)?
+    };
    rename_prepared_source_std(&prepared, &original_path, canonical_head, base, publication_root, &guard)?;
    prepared.path = canonical_head.to_path_buf();
    fsync_commit_directory_limited_std(parent, &guard)?;
@@ -2979,10 +2979,11 @@ fn build_legacy_per_part_fallback_reader(
    part_length: usize,
    part_size: usize,
    read_repair: Option<(DecodeReadRepairContext, bool)>,
-) -> Box<dyn AsyncRead + Unpin + Send + Sync> {
+) -> Box<LegacyPerPartDecodeReader> {
    let buffer = adaptive_duplex_buffer_size(part_size as i64);
-    let (read_half, mut write_half) = tokio::io::duplex(buffer);
+    let (read_half, write_half) = tokio::io::duplex(buffer);
    let decode = tokio::spawn(async move {
+        let mut write_half = GetObjectDownstreamWriter::new(write_half);
        let outcome = erasure
            .decode_with_stripe_handles_and_reopeners_with_diagnostics(
                &mut write_half,
@@ -8908,6 +8909,42 @@ mod tests {
        assert!(err.to_string().contains("inconsistent read source shards"));
    }

+    #[tokio::test]
+    async fn legacy_per_part_output_close_retains_downstream_classification() {
+        let erasure = coding::Erasure::new(2, 2, 64);
+        let data = (0..64u8).collect::<Vec<_>>();
+        let setup =
+            setup_codec_data_blocks_first_encoded_bitrot_readers(&erasure, &data, &[0], &[], &[], HashAlgorithm::HighwayHash256)
+                .await;
+        let mut reader = build_legacy_per_part_fallback_reader(
+            erasure,
+            setup.readers,
+            setup.deferred_stripe_handles,
+            Vec::new(),
+            None,
+            0,
+            data.len(),
+            data.len(),
+            None,
+        );
+        // Closing the actual output pipe models the body consumer disappearing
+        // while retaining the executor long enough to inspect its typed result.
+        assert!(!reader.finished);
+        let (replacement, _unused) = tokio::io::duplex(1);
+        drop(std::mem::replace(&mut reader.inner, replacement));
+        let handle = reader.decode.take().expect("fallback decode task");
+        let error = handle
+            .await
+            .expect("join the fallback task")
+            .expect("closed pipe must interrupt output");
+        assert_eq!(error.kind(), std::io::ErrorKind::BrokenPipe);
+        assert_eq!(
+            crate::diagnostics::get::classify_io_error(&error),
+            GetObjectFailureReason::DownstreamClosed,
+            "fallback output must carry the same marker as the normal GET path"
+        );
+    }
+
    // backlog#879: a later multipart part that loses a shard (read quorum still
    // holds) must be reconstructed in place through the legacy per-part decode
    // bridge instead of failing the stream. This exercises the exact reader the
@@ -30,6 +30,78 @@ use storage_api::integration::{
    WriteCommitGuard, WriteCompletion,
 };

+async fn complete_fixture_replicas(env: &TestECStoreEnv, bucket: &str, object: &str) {
+    // TailDrained completes fan-out, but a successful quorum write may still
+    // have minority failures. Fault injection needs a fully repaired baseline.
+    let result = ECStoreHealStorage::new(env.ecstore.clone())
+        .heal_object_with_receipt(
+            bucket,
+            object,
+            None,
+            &HealOpts {
+                scan_mode: HealScanMode::Deep,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("complete protected fixture replicas before fault injection");
+    assert!(result.error.is_none(), "fixture repair failed: {:?}", result.error);
+    assert_eq!(result.item.after.drives.len(), env.disk_paths.len());
+    assert!(
+        result.item.after.drives.iter().all(|drive| drive.state == "ok"),
+        "fixture requires every replica, not only read/write quorum: {:?}",
+        result.item.after.drives
+    );
+    assert!(result.item.integrity_verified, "fixture must retain protected payload verification");
+}
+
+#[tokio::test]
+#[serial]
+async fn mixed_batch_fixture_repairs_a_missing_minority_replica() {
+    let root = tempfile::tempdir().expect("minority replica fixture");
+    let env = TestECStoreEnv::builder().base_dir(root.path()).build().await;
+    let bucket = "mixed-fixture-baseline";
+    let object = "object";
+    env.make_bucket(bucket, false).await;
+    let expected = vec![0x5a; 1024 * 1024 + 123];
+    env.ecstore
+        .put_object(
+            bucket,
+            object,
+            &mut PutObjReader::from_vec(expected.clone()),
+            &ObjectOptions {
+                shard_integrity_write_mode: Some(ShardIntegrityWriteMode::Protected),
+                write_completion: WriteCompletion::TailDrained,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("quorum commit fixture");
+    // Guarantee a minority failure regardless of whether the original write
+    // succeeded on every disk, then repair it before any test fault is applied.
+    let replica = env.disk_paths[0].join(bucket).join(object);
+    if replica.exists() {
+        tokio::fs::remove_dir_all(&replica)
+            .await
+            .expect("remove one complete replica");
+    }
+    complete_fixture_replicas(&env, bucket, object).await;
+    let disks = env.ecstore.pools[0].get_disks(0);
+    for disk in disks.disks.read().await.iter().flatten() {
+        disk.read_version("", bucket, object, "", &ReadOptions::default())
+            .await
+            .expect("every fixture replica has metadata after completion");
+    }
+    let mut reader = env
+        .ecstore
+        .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
+        .await
+        .expect("read the completed fixture");
+    let mut actual = Vec::new();
+    reader.stream.read_to_end(&mut actual).await.expect("completed fixture body");
+    assert_eq!(actual, expected);
+}
+
 #[test]
 fn disabled_lock_reports_only_quorum_selection_and_deep_check_observations() {
    const CHILD_ENV: &str = "RUSTFS_TEST_DISABLED_LOCK_HEAL_CHILD";
@@ -704,6 +776,7 @@ fn admin_mixed_batch_retains_repairs_after_object_retry_exhaustion() {
                            )
                            .await
                            .expect("commit complete fixture");
+                        complete_fixture_replicas(&env, bucket, object).await;
                        for (slot, disk) in disks.iter().enumerate() {
                            let meta = disk
                                .read_version("", bucket, object, "", &ReadOptions::default())
@@ -64,8 +64,6 @@ async-compression = { workspace = true, features = ["tokio", "gzip", "bzip2"] }
 async-trait.workspace = true
 arc-swap.workspace = true
 bytes = { workspace = true, features = ["serde"] }
-base64-simd.workspace = true
-flatbuffers.workspace = true
 chrono = { workspace = true, features = ["serde"] }
 crc-fast.workspace = true
 rustfs-common.workspace = true
@@ -77,7 +75,6 @@ futures-core = { workspace = true }
 flate2.workspace = true
 brotli.workspace = true
 lz4.workspace = true
-lz4_flex.workspace = true
 http.workspace = true
 memchr.workspace = true
 s3s = { workspace = true, features = ["minio"] }
@@ -221,7 +221,7 @@ struct ThriftCompactCursor<'a> {
    remaining: usize,
 }

-impl<'a> ThriftCompactCursor<'a> {
+impl ThriftCompactCursor<'_> {
    fn error() -> ParquetBudgetError {
        ParquetBudgetError::Invalid("Parquet thrift metadata is malformed".to_owned())
    }
@@ -241,14 +241,6 @@ impl<'a> ThriftCompactCursor<'a> {
        Ok(())
    }

-    fn binary(&mut self) -> Result<&'a [u8]> {
-        let len = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
-        self.charge(len.checked_mul(2).ok_or_else(budget_error)?)?;
-        let start = self.pos;
-        self.skip_bytes(len)?;
-        Ok(&self.bytes[start..self.pos])
-    }
-
    fn varint(&mut self) -> Result<u64> {
        let mut value = 0_u64;
        for shift in (0..64).step_by(7) {
@@ -343,7 +335,11 @@ impl<'a> ThriftCompactCursor<'a> {
            THRIFT_COMPACT_BYTE => self.skip_bytes(1),
            THRIFT_COMPACT_I16 | THRIFT_COMPACT_I32 | THRIFT_COMPACT_I64 => self.varint().map(drop),
            THRIFT_COMPACT_DOUBLE => self.skip_bytes(8),
-            THRIFT_COMPACT_BINARY => self.binary().map(drop),
+            THRIFT_COMPACT_BINARY => {
+                let len = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
+                self.charge(len.checked_mul(2).ok_or_else(budget_error)?)?;
+                self.skip_bytes(len)
+            }
            THRIFT_COMPACT_LIST | THRIFT_COMPACT_SET => {
                let (element, count) = self.list_header()?;
                for _ in 0..count {
@@ -590,98 +586,21 @@ impl<'a> ThriftCompactCursor<'a> {
        Ok(())
    }

-    fn arrow_schema_metadata(&mut self) -> Result<Option<&'a [u8]>> {
-        let (element, count) = self.list_header()?;
-        if element != THRIFT_COMPACT_STRUCT {
-            return Err(Self::error());
-        }
-        let mut arrow_schema = None;
-        for _ in 0..count {
-            let mut field = 0;
-            let mut key = None;
-            let mut value = None;
-            loop {
-                let header = self.byte()?;
-                if header == THRIFT_COMPACT_STOP {
-                    break;
-                }
-                field = self.field_id(field, header)?;
-                let kind = header & 0x0f;
-                if matches!(field, 1 | 2) && kind == THRIFT_COMPACT_BINARY {
-                    let bytes = self.binary()?;
-                    if field == 1 {
-                        key = Some(bytes);
-                    } else {
-                        value = Some(bytes);
-                    }
-                } else {
-                    self.skip_field(MetadataStruct::KeyValue, field, kind, 1)?;
-                }
-            }
-            // Match the reader's last key with a value; absent values are ignored.
-            if key == Some(b"ARROW:schema".as_slice()) && value.is_some() {
-                arrow_schema = value;
-            }
-        }
-        Ok(arrow_schema)
-    }
-
-    fn validate_arrow_schema(&mut self, encoded: &[u8]) -> Result<()> {
-        use datafusion::arrow::ipc;
-        let invalid = || ParquetBudgetError::Invalid("Parquet Arrow schema metadata is malformed".to_owned());
-        // This reservation precedes base64 allocation. The binary field has
-        // already reserved its owned metadata copies; decoding needs another buffer.
-        self.charge(encoded.len())?;
-        let bytes = base64_simd::STANDARD.decode_to_vec(encoded).map_err(|_| invalid())?;
-        let message = if bytes.len() > 8 && bytes.starts_with(&[255; 4]) {
-            &bytes[8..]
-        } else {
-            bytes.as_slice()
-        };
-        // The verifier visits shared offsets for every occurrence, before the
-        // reader expands them into owned fields and metadata. Split the remaining
-        // allowance between string/buffer copies and per-table Rust containers:
-        // four owned copies per apparent byte and 1 KiB per visited table.
-        let defaults = flatbuffers::VerifierOptions::default();
-        let options = flatbuffers::VerifierOptions {
-            max_apparent_size: defaults.max_apparent_size.min(self.remaining / 8),
-            max_tables: defaults.max_tables.min(self.remaining / (2 * 1024)),
-            ..defaults
-        };
-        let message = ipc::root_as_message_with_opts(&options, message).map_err(|error| match error {
-            flatbuffers::InvalidFlatbuffer::TooManyTables
-            | flatbuffers::InvalidFlatbuffer::ApparentSizeTooLarge
-            | flatbuffers::InvalidFlatbuffer::DepthLimitReached => budget_error(),
-            _ => invalid(),
-        })?;
-        let schema = message.header_as_schema().ok_or_else(invalid)?;
-        if schema.fields().is_none() {
-            return Err(invalid());
-        }
-        Ok(())
-    }
-
    /// Walk the footer's `FileMetaData`. The pinned parquet reserves the
    /// declared `row_groups` capacity before reading one element; charge that
    /// reservation to the budget so a small footer cannot request gigabytes.
    fn file_metadata(&mut self, limit: usize) -> Result<()> {
        let mut field_id = 0_i16;
        let mut reserved = 0_usize;
-        let mut arrow_schema = None;
        loop {
            let header = self.byte()?;
            if header == THRIFT_COMPACT_STOP {
-                if let Some(encoded) = arrow_schema {
-                    self.validate_arrow_schema(encoded)?;
-                }
                return Ok(());
            }
            let kind = header & 0x0f;
            field_id = self.field_id(field_id, header)?;
            if field_id == 2 && kind == THRIFT_COMPACT_LIST {
                self.schema()?;
-            } else if field_id == 5 && kind == THRIFT_COMPACT_LIST {
-                arrow_schema = self.arrow_schema_metadata()?;
            } else if field_id == FILE_METADATA_FIELD_ROW_GROUPS && kind == THRIFT_COMPACT_LIST {
                let (element, count) = self.list_header()?;
                if element != THRIFT_COMPACT_STRUCT {
@@ -841,34 +760,21 @@ fn validate_compressed_page(payload: &[u8], sizes: PageHeaderSizes, codec: Compr
            bounded_stream_size(brotli::Decompressor::new(payload, 4096), expanded)?;
            Ok(window)
        }
-        Compression::LZ4 if payload.starts_with(&[4, 34, 77, 24]) || payload.starts_with(&[2, 33, 76, 24]) => {
-            let (block, linked) = if payload.starts_with(&[2, 33, 76, 24]) {
-                (8 * 1024 * 1024, false)
-            } else {
-                let descriptor = *payload.get(5).ok_or_else(ThriftCompactCursor::error)?;
-                let block = match (descriptor >> 4) & 7 {
-                    4 => 64 * 1024,
-                    5 => 256 * 1024,
-                    6 => 1024 * 1024,
-                    7 => 4 * 1024 * 1024,
-                    _ => return Err(ThriftCompactCursor::error()),
-                };
-                (block, payload[4] & 0x20 == 0)
+        Compression::LZ4 if payload.starts_with(&[4, 34, 77, 24]) => {
+            let descriptor = *payload.get(5).ok_or_else(ThriftCompactCursor::error)?;
+            let block = match (descriptor >> 4) & 7 {
+                4 => 64 * 1024,
+                5 => 256 * 1024,
+                6 => 1024 * 1024,
+                7 => 4 * 1024 * 1024,
+                _ => return Err(ThriftCompactCursor::error()),
            };
-            // Match parquet's pinned fallback decoder: compressed input plus
-            // output, with an extra output block and 64 KiB history when linked.
-            let working = if linked { block * 3 + 64 * 1024 } else { block * 2 };
+            let working = block * 2;
            if working > limit {
                return Err(budget_error());
            }
-            match bounded_stream_size(lz4_flex::frame::FrameDecoder::new(payload), expanded) {
-                // Hadoop data can share the magic prefix, and parquet also
-                // tries bounded RAW decoding after a frame error. The bounded
-                // sink proved the frame produced no more than `expanded` bytes
-                // before that error; let the actual reader select its fallback.
-                Err(ParquetBudgetError::Codec(_)) => Ok(working),
-                result => result.map(|()| working),
-            }
+            bounded_stream_size(lz4::Decoder::new(payload)?, expanded)?;
+            Ok(working)
        }
        _ => Ok(0),
    }
@@ -945,49 +851,6 @@ mod tests {
        assert!(validate_compressed_page(&payload, compressed_page(&payload, 64), Compression::LZ4, 1024 * 1024).is_err());
    }

-    #[test]
-    fn legacy_lz4_scratch_is_admitted_before_the_fallback_decoder() {
-        let payload = [2, 33, 76, 24, 5, 0, 0, 0, 0x40, 42, 0, 0, 0];
-        let header = compressed_page(&payload, 4);
-        assert!(matches!(
-            validate_compressed_page(&payload, header, Compression::LZ4, 1024 * 1024),
-            Err(ParquetBudgetError::Exceeded)
-        ));
-        let admitted = validate_compressed_page(&payload, header, Compression::LZ4, 32 * 1024 * 1024)
-            .expect("legacy LZ4 remains supported with enough scratch memory");
-        assert!(admitted >= 16 * 1024 * 1024);
-        assert!(validate_compressed_page(&payload, compressed_page(&payload, 1), Compression::LZ4, 32 * 1024 * 1024).is_err());
-    }
-
-    #[test]
-    fn hadoop_lz4_can_share_the_legacy_magic_without_becoming_a_frame() {
-        let size = usize::try_from(u32::from_be_bytes([2, 33, 76, 24])).expect("Hadoop size");
-        let bytes = vec![b'x'; size];
-        let block = lz4_flex::block::compress(&bytes);
-        let mut payload = vec![2, 33, 76, 24];
-        payload.extend(u32::try_from(block.len()).expect("compressed block").to_be_bytes());
-        payload.extend(block);
-        validate_compressed_page(&payload, compressed_page(&payload, size), Compression::LZ4, 64 * 1024 * 1024)
-            .expect("Hadoop framing takes priority over the frame fallback");
-    }
-
-    #[test]
-    fn linked_lz4_accounts_for_the_history_and_both_block_buffers() {
-        use lz4_flex::frame::{BlockMode, BlockSize, FrameEncoder, FrameInfo};
-        let frame = FrameInfo::new().block_mode(BlockMode::Linked).block_size(BlockSize::Max64KB);
-        let mut encoder = FrameEncoder::with_frame_info(frame, Vec::new());
-        encoder.write_all(&vec![b'x'; 256 * 1024]).expect("linked input");
-        let payload = encoder.finish().expect("linked frame");
-        let header = compressed_page(&payload, 256 * 1024);
-        assert!(matches!(
-            validate_compressed_page(&payload, header, Compression::LZ4, 192 * 1024),
-            Err(ParquetBudgetError::Exceeded)
-        ));
-        let admitted = validate_compressed_page(&payload, header, Compression::LZ4, 1024 * 1024)
-            .expect("linked frame with enough scratch memory");
-        assert!(admitted >= 256 * 1024);
-    }
-
    #[test]
    fn uncompressed_v2_does_not_invoke_the_column_codec() {
        let payload = b"levels-and-uncompressed-values";
@@ -1078,170 +941,6 @@ mod tests {
        validate_file_footer(&bytes, 1024 * 1024).expect("valid logical annotations remain accepted");
    }

-    fn shared_arrow_metadata_file(entries: usize) -> Vec<u8> {
-        use datafusion::{
-            arrow::ipc,
-            parquet::{
-                file::{properties::WriterProperties, writer::SerializedFileWriter},
-                schema::parser::parse_message_type,
-            },
-        };
-        use std::sync::Arc;
-        let mut builder = flatbuffers::FlatBufferBuilder::new();
-        let name = builder.create_string("v");
-        let value = builder.create_string(&"x".repeat(64 * 1024));
-        let data_type = ipc::Int::create(
-            &mut builder,
-            &ipc::IntArgs {
-                bitWidth: 32,
-                is_signed: true,
-            },
-        );
-        let field = ipc::Field::create(
-            &mut builder,
-            &ipc::FieldArgs {
-                name: Some(name),
-                nullable: true,
-                type_type: ipc::Type::Int,
-                type_: Some(data_type.as_union_value()),
-                ..Default::default()
-            },
-        );
-        let fields = builder.create_vector(&[field]);
-        let mut metadata = Vec::new();
-        for n in 0..entries {
-            let key = builder.create_string(&format!("key-{n}"));
-            metadata.push(ipc::KeyValue::create(
-                &mut builder,
-                &ipc::KeyValueArgs {
-                    key: Some(key),
-                    value: Some(value),
-                },
-            ));
-        }
-        let metadata = builder.create_vector(&metadata);
-        let schema = ipc::Schema::create(
-            &mut builder,
-            &ipc::SchemaArgs {
-                fields: Some(fields),
-                custom_metadata: Some(metadata),
-                ..Default::default()
-            },
-        );
-        let message = ipc::Message::create(
-            &mut builder,
-            &ipc::MessageArgs {
-                version: ipc::MetadataVersion::V5,
-                header_type: ipc::MessageHeader::Schema,
-                header: Some(schema.as_union_value()),
-                ..Default::default()
-            },
-        );
-        builder.finish(message, None);
-        let encoded = base64_simd::STANDARD.encode_to_string(builder.finished_data());
-        let properties = WriterProperties::builder()
-            .set_key_value_metadata(Some(vec![datafusion::parquet::file::metadata::KeyValue::new(
-                "ARROW:schema".to_owned(),
-                Some(encoded),
-            )]))
-            .build();
-        let schema = Arc::new(parse_message_type("message root { OPTIONAL INT32 v; }").expect("physical schema"));
-        let mut bytes = Vec::new();
-        SerializedFileWriter::new(&mut bytes, schema, Arc::new(properties))
-            .expect("writer")
-            .close()
-            .expect("footer");
-        bytes
-    }
-
-    #[test]
-    fn shared_arrow_metadata_is_admitted_before_owned_schema_conversion() {
-        use datafusion::parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder;
-        let bytes = shared_arrow_metadata_file(256);
-        assert!(bytes.len() < 100 * 1024, "shared offsets keep the encoded file small");
-        let native = ParquetRecordBatchReaderBuilder::try_new(bytes::Bytes::from(bytes.clone())).expect("valid Arrow metadata");
-        assert_eq!(native.schema().metadata().len(), 256);
-        assert!(
-            matches!(validate_file_footer(&bytes, 2 * 1024 * 1024), Err(ParquetBudgetError::Exceeded)),
-            "32 MiB of owned metadata must not pass a 2 MiB preflight"
-        );
-        validate_file_footer(&bytes, 512 * 1024 * 1024).expect("sufficient budget preserves shared metadata");
-    }
-
-    fn arrow_hint_footer(entries: &[(&[u8], Option<&[u8]>)]) -> Vec<u8> {
-        let count = u8::try_from(entries.len()).expect("small key/value fixture");
-        assert!(count < 15);
-        let mut bytes = vec![0x59, (count << 4) | THRIFT_COMPACT_STRUCT];
-        for (key, value) in entries {
-            bytes.push(0x18);
-            bytes.extend(varint(u64::try_from(key.len()).expect("key length")));
-            bytes.extend(*key);
-            if let Some(value) = value {
-                bytes.push(0x18);
-                bytes.extend(varint(u64::try_from(value.len()).expect("value length")));
-                bytes.extend(*value);
-            }
-            bytes.push(0);
-        }
-        bytes.push(0);
-        bytes
-    }
-
-    #[test]
-    fn arrow_hint_framing_and_duplicate_keys_match_the_reader() {
-        use datafusion::arrow::{
-            datatypes::{DataType, Field, Schema},
-            ipc,
-        };
-        let schema = Schema::new(vec![Field::new("v", DataType::Int32, true)]);
-        let mut builder = flatbuffers::FlatBufferBuilder::new();
-        let schema = ipc::convert::schema_to_fb_offset(&mut builder, &schema);
-        let message = ipc::Message::create(
-            &mut builder,
-            &ipc::MessageArgs {
-                header_type: ipc::MessageHeader::Schema,
-                header: Some(schema.as_union_value()),
-                ..Default::default()
-            },
-        );
-        builder.finish(message, None);
-        let raw = builder.finished_data();
-        let mut framed = vec![255; 4];
-        framed.extend(u32::try_from(raw.len()).expect("small schema").to_le_bytes());
-        framed.extend(raw);
-        for message in [raw, framed.as_slice()] {
-            let encoded = base64_simd::STANDARD.encode_to_string(message);
-            let footer = arrow_hint_footer(&[
-                (b"ARROW:schema", Some(b"invalid")),
-                (b"ARROW:schema", Some(encoded.as_bytes())),
-                (b"ARROW:schema", None),
-            ]);
-            validate_footer(&footer, 1024 * 1024).expect("last present Arrow metadata wins, both IPC framings are supported");
-            assert!(matches!(
-                validate_footer(
-                    &arrow_hint_footer(&[
-                        (b"ARROW:schema", Some(encoded.as_bytes())),
-                        (b"ARROW:schema", Some(b"invalid"))
-                    ]),
-                    1024 * 1024
-                ),
-                Err(ParquetBudgetError::Invalid(_))
-            ));
-        }
-        validate_footer(&arrow_hint_footer(&[(b"other", Some(b"invalid"))]), 1024 * 1024).expect("other metadata is opaque");
-    }
-
-    #[test]
-    fn arrow_hint_value_before_key_is_still_admitted() {
-        // Compact struct fields are identified by ID, not required to be ordered.
-        let mut bytes = vec![0x59, 0x1c, 0x28, 3];
-        bytes.extend(b"%%%");
-        bytes.extend([0x08, 2, 12]);
-        bytes.extend(b"ARROW:schema");
-        bytes.extend([0, 0]);
-        assert!(matches!(validate_footer(&bytes, 1024 * 1024), Err(ParquetBudgetError::Invalid(_))));
-    }
-
    #[test]
    fn footer_collections_are_bounded_before_decoder_allocation() {
        let mut huge = vec![0x49, 0xfc];
@@ -1106,98 +1106,6 @@ mod tests {
        assert_eq!(error.select_error(), SelectError::ResourceExhausted);
    }

-    #[tokio::test]
-    async fn parquet_legacy_lz4_checks_native_scratch_before_query_decode() {
-        use datafusion::parquet::{basic::Compression, file::metadata::ParquetMetaDataWriter};
-        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Int32, false)]));
-        let batch =
-            RecordBatch::try_new(Arc::clone(&schema), vec![Arc::new(datafusion::arrow::array::Int32Array::from(vec![42]))])
-                .expect("one integer row");
-        let properties = datafusion::parquet::file::properties::WriterProperties::builder()
-            .set_dictionary_enabled(false)
-            .build();
-        let mut original = Vec::new();
-        let mut writer = ArrowWriter::try_new(&mut original, schema, Some(properties)).expect("writer");
-        writer.write(&batch).expect("row");
-        writer.close().expect("footer");
-        let metadata = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(original))
-            .expect("metadata")
-            .metadata()
-            .clone();
-        // A valid legacy frame encoding the four little-endian bytes for 42.
-        let payload = [2, 33, 76, 24, 5, 0, 0, 0, 0x40, 42, 0, 0, 0];
-        let mut page = vec![0x15, 0, 0x15, 8, 0x15, 26, 0x2c, 0x15, 2, 0x15, 0, 0x15, 6, 0x15, 6, 0, 0];
-        page.extend(payload);
-        let column = metadata
-            .row_group(0)
-            .column(0)
-            .clone()
-            .into_builder()
-            .set_compression(Compression::LZ4)
-            .set_data_page_offset(4)
-            .set_total_compressed_size(i64::try_from(page.len()).expect("small page"))
-            .build()
-            .expect("legacy column");
-        let group = metadata
-            .row_group(0)
-            .clone()
-            .into_builder()
-            .set_column_metadata(vec![column])
-            .build()
-            .expect("row group");
-        let metadata = metadata.as_ref().clone().into_builder().set_row_groups(vec![group]).build();
-        let mut encoded = b"PAR1".to_vec();
-        encoded.extend(page);
-        ParquetMetaDataWriter::new(&mut encoded, &metadata)
-            .finish()
-            .expect("legacy footer");
-        assert!(encoded.len() < 4096);
-        let mut native = ParquetRecordBatchReaderBuilder::try_new(Bytes::from(encoded.clone()))
-            .expect("legacy metadata")
-            .build()
-            .expect("native reader");
-        let decoded = native.next().expect("one batch").expect("legacy fallback decode");
-        assert_eq!(
-            decoded
-                .column(0)
-                .as_primitive::<datafusion::arrow::datatypes::Int32Type>()
-                .value(0),
-            42
-        );
-        let store = Arc::new(InMemory::new());
-        store
-            .put(&Path::from("legacy.parquet"), Bytes::from(encoded).into())
-            .await
-            .expect("fixture");
-        let tight = parquet_session_with_memory(store.clone(), 2 * 1024 * 1024);
-        let result = ParquetSelectTable::try_new(&tight.state(), &parquet_input("legacy.parquet"), None).await;
-        let error = match result {
-            Ok(_) => panic!("legacy decoder scratch exceeds the query budget"),
-            Err(error) => error,
-        };
-        assert_eq!(error.select_error(), SelectError::ResourceExhausted);
-        let roomy = parquet_session_with_memory(store, 128 * 1024 * 1024);
-        let table = ParquetSelectTable::try_new(&roomy.state(), &parquet_input("legacy.parquet"), None)
-            .await
-            .expect("admitted legacy file");
-        roomy.register_table("source", table).expect("table");
-        let rows = roomy
-            .sql("SELECT v FROM source")
-            .await
-            .expect("plan")
-            .collect()
-            .await
-            .expect("query");
-        assert_eq!(rows.iter().map(RecordBatch::num_rows).sum::<usize>(), 1);
-        assert_eq!(
-            rows[0]
-                .column(0)
-                .as_primitive::<datafusion::arrow::datatypes::Int32Type>()
-                .value(0),
-            42
-        );
-    }
-
    #[tokio::test]
    async fn parquet_nested_views_decode_through_the_actual_file_schema() {
        use datafusion::arrow::array::{ListArray, StringArray};
@@ -230,6 +230,17 @@ impl DynamicSleeper {
        SCANNER_IDLE_MODE.load(Ordering::Relaxed)
    }

+    fn foreground_backoff(&self) -> Duration {
+        // Local walkers already use disk-priority admission and a shared I/O
+        // budget. A global per-operation floor duplicates that control and
+        // throttles idle disks because unrelated foreground work is active.
+        if self.disk_io_budget.is_some() {
+            Duration::ZERO
+        } else {
+            foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity())
+        }
+    }
+
    fn read_params(&self) -> (f64, Duration) {
        let config = self.read_config();
        (config.factor, config.max_sleep)
@@ -252,15 +263,14 @@ impl DynamicSleeper {
        }
        let (factor, max_sleep) = self.read_params();
        if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.foreground_backoff();
            if !foreground_sleep.is_zero() {
                tokio::time::sleep(foreground_sleep).await;
                global_metrics().record_scanner_throttle_sleep(foreground_sleep);
            }
            return;
        }
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.foreground_backoff();
        let sleep_dur = Duration::from_secs_f64(MIN_SLEEP.as_secs_f64() * factor)
            .min(max_sleep)
            .max(foreground_sleep);
@@ -354,8 +364,7 @@ impl SleepTimer {
        }
        let (factor, max_sleep) = self.sleeper.read_params();
        if factor == 0.0 || max_sleep.is_zero() {
-            let foreground_sleep =
-                foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+            let foreground_sleep = self.sleeper.foreground_backoff();
            if !foreground_sleep.is_zero() {
                tokio::time::sleep(foreground_sleep).await;
                global_metrics().record_scanner_throttle_sleep(foreground_sleep);
@@ -363,7 +372,7 @@ impl SleepTimer {
            return;
        }
        let elapsed = self.start.elapsed();
-        let foreground_sleep = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground_sleep = self.sleeper.foreground_backoff();
        let sleep_dur = Duration::from_secs_f64(elapsed.as_secs_f64() * factor)
            .max(MIN_SLEEP)
            .min(max_sleep)
@@ -135,7 +135,7 @@ impl DynamicSleeper {
        if let Some(elapsed) = elapsed {
            debt.charge(proportional_delay(elapsed, config), Instant::now());
        }
-        let foreground = foreground_workload_backoff_duration(crate::workload_admission::foreground_workload_activity());
+        let foreground = self.foreground_backoff();
        let now = Instant::now();
        let folder_reaches_deadline = folder
            && debt.started.is_some_and(|start| {
@@ -181,6 +181,54 @@ async fn active_wait_wakes_for_fastest_idle_off_and_smaller_max_wait() {
    }
 }

+#[tokio::test(start_paused = true)]
+#[serial]
+async fn disk_bound_pacing_does_not_repeat_global_foreground_backoff() {
+    let _idle = IdleGuard::enabled();
+    struct ForegroundGuard;
+    impl Drop for ForegroundGuard {
+        fn drop(&mut self) {
+            crate::reset_foreground_read_activity_for_test();
+        }
+    }
+    let _foreground = ForegroundGuard;
+    crate::set_foreground_read_activity(25);
+    let disk = tempfile::tempdir().expect("disk identity");
+    let sleeper = DynamicSleeper::new(ScannerSpeed::Default).for_disk_with_budget_for_tests(
+        disk.path(),
+        ScannerDiskIoBudgetLimits {
+            iops: Some(0),
+            bytes_per_sec: None,
+        },
+    );
+    let mut debt = PacingDebt::default();
+    let begin = Instant::now();
+    for _ in 0..100 {
+        sleeper.pace_object(&mut debt, Duration::from_micros(20)).await;
+    }
+    sleeper.flush_pacing(&mut debt).await;
+    assert_eq!(begin.elapsed(), Duration::from_millis(4));
+    assert_eq!(debt.charged_objects, 100);
+    assert_eq!(debt.pending, Duration::ZERO);
+    // The first operation consumes the token bucket's one-operation burst.
+    sleeper.admit_disk_io(1, 0).await;
+    let begin = Instant::now();
+    sleeper.sleep_folder_paced(&mut debt).await;
+    assert_eq!(
+        begin.elapsed(),
+        SCANNER_DISK_IO_BUDGET_MAX_WAIT + Duration::from_millis(2),
+        "the per-disk budget and preset folder gap must still apply"
+    );
+    assert_eq!(
+        scanner_disk_io_budget_status()
+            .into_iter()
+            .find(|status| status.disk == disk.path().to_string_lossy())
+            .expect("disk charges remain visible")
+            .charged_ops,
+        2
+    );
+}
+
 #[tokio::test(start_paused = true)]
 #[serial]
 async fn foreground_backoff_flushes_tail_without_double_charging_or_max_wait_cap() {
@@ -40,7 +40,7 @@ Bitrot cycle resolution differs because the canonical persistent key belongs to
 | `scanner.cycle_max_objects` | `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS` | objects | `0` (`DEFAULT_SCANNER_CYCLE_MAX_OBJECTS`) | Caps objects processed by one cycle. `0` disables this budget. |
 | `scanner.cycle_max_directories` | `RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES` | directories | `0` (`DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES`) | Caps directories entered by one cycle. `0` disables this budget. |
 | `heal.bitrot_cycle` | `RUSTFS_SCANNER_BITROT_CYCLE_SECS` | seconds | `2592000` (`DEFAULT_HEAL_BITROT_CYCLE_SECS`, 30 days) | Controls periodic deep bitrot scans. `false`, `off`, `no`, or `disabled` disables periodic deep scans; `0`, `true`, `on`, or `yes` runs deep mode every scanner cycle. |
-| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and foreground-read backoff. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
+| `scanner.idle_mode` | `RUSTFS_SCANNER_IDLE_MODE` | boolean | `true` (`DEFAULT_SCANNER_IDLE_MODE`) | Enables preset scanner sleeps and the foreground backoff for callers without a disk binding. `false` disables those sleeps; per-disk foreground-priority admission remains active. |
 | `scanner.cache_save_timeout` | `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` | seconds | `14` (`DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS`) | Timeout for saving scanner cache; runtime enforces a minimum of `1` and keeps the default persistence budget within the distributed publication lease. |
 | `scanner.recovery_backoff_cap` | `RUSTFS_SCANNER_RECOVERY_BACKOFF_CAP_SECS` | seconds | `600` (`DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS`) | Caps the wait between automatic startup recovery retries. Must be a positive integer; `0` and malformed values are rejected, so retries never stop. Applies to the next computed retry, not to a wait already in progress. The retry count resets after 24 hours, which bounds a single wait at about one day even with a larger cap. |
 | `scanner.max_concurrent_set_scans` | `RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS` | count | `4` (`DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS`) | Caps concurrent set-level scanner tasks. `0` keeps topology-derived concurrency. |
@@ -63,7 +63,7 @@ Speed presets (`crates/config/src/constants/scanner.rs`) set the base sleep mult
 | `slow` | 10x | 15s | 60s |
 | `slowest` | 100x | 15s | 30m |

-Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is `time spent on the object x factor`, both capped at `max_wait`; a foreground-read floor of `FOREGROUND_READ_BACKOFF_PER_REQUEST_MS` (10ms) per concurrent GetObject/streaming read, capped at `FOREGROUND_READ_BACKOFF_MAX_MS` (250ms), is applied on top and can exceed the preset's `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate is independent of `idle_mode` and remains active when sleeps are disabled.
+Use `scanner.delay`, `scanner.max_wait`, and `scanner.cycle` when the preset is close but one axis needs a precise override. With `idle_mode=true`, directory-level sleep is `1ms x factor` and object-level sleep is admitted work time multiplied by the factor, capped at `max_wait`; object pacing batches small waits and settles remaining debt at bucket completion. Local walkers bound with `DynamicSleeper::for_disk` use disk-priority admission and their shared I/O budget without an additional global foreground-request sleep. A caller without that binding retains a foreground-workload floor of 10ms per active request, capped at 250ms, which can exceed `max_wait` (`crates/scanner/src/sleeper.rs`). The per-disk foreground-priority gate and disk budget remain active when preset sleeps are disabled. Removing redundant global backoff does not guarantee a fixed snapshot age: scans still compete for admitted resources and may checkpoint at their cycle budget.

 ### Startup and cycle budgets

@@ -1882,9 +1882,16 @@ impl DefaultObjectUsecase {

        reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
        if let Some(control) = progress_control {
-            use rustfs_rio::HashReaderMut;
-            let inner = reader.take_inner();
-            reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
+            if !should_compress
+                && (use_zero_copy_eager_put_path || use_empty_or_small_eager_put_path)
+                && trailing_headers.is_none()
+            {
+                control.finish();
+            } else {
+                use rustfs_rio::HashReaderMut;
+                let inner = reader.take_inner();
+                reader.inner = rustfs_rio::boxed_reader(super::request_body::DemandReader::new(inner, control));
+            }
        }
        rustfs_io_metrics::record_put_object_stage_duration_from("app_encryption_prepare", encryption_stage_start);

@@ -160,7 +160,7 @@ impl BodyReadControl {
        state.waited = Duration::ZERO;
    }

-    fn finish(&self) {
+    pub(super) fn finish(&self) {
        let mut state = self.0.budget.lock();
        self.0.active.store(false, Ordering::Release);
        state.finished = true;
@@ -0,0 +1,111 @@
+#!/usr/bin/env python3
+"""Exercise the release download entry point without fetching external tools."""
+
+import hashlib
+import json
+import os
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+
+SCRIPT = Path(__file__).resolve().parents[1] / ".github/actions/setup/download_release_asset.py"
+
+
+class ReleaseAssetTests(unittest.TestCase):
+    def setUp(self):
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.body = b"verified tool fixture"
+        self.sha256 = hashlib.sha256(self.body).hexdigest()
+        self.asset = {"name": "tool.zip", "id": 17, "size": len(self.body), "digest": "sha256:" + self.sha256}
+        self.metadata = {"tag_name": "v1.0", "assets": [self.asset]}
+        self.output = self.root / "tool.zip"
+        self.output.write_bytes(b"previous file")
+        self.calls = self.root / "calls.jsonl"
+        gh = self.root / "gh"
+        gh.write_text(f"#!{sys.executable}\n" + '''import json,os,pathlib,sys
+args=sys.argv[1:]
+with pathlib.Path(os.environ['CALLS']).open('a') as out:out.write(json.dumps(args)+'\\n')
+if '/releases/tags/' in args[-1]:
+    if os.environ.get('FAIL_METADATA'):raise SystemExit(19)
+    print(os.environ['METADATA'])
+else:
+    if os.environ.get('FAIL_DOWNLOAD'):raise SystemExit(23)
+    sys.stdout.buffer.write(bytes.fromhex(os.environ['BODY']))
+''')
+        gh.chmod(0o755)
+        self.env = dict(os.environ, PATH=str(self.root) + os.pathsep + os.environ["PATH"], CALLS=str(self.calls), GH_TOKEN="synthetic-do-not-log-token")
+
+    def invoke(self, extra=(), **env):
+        environment = dict(self.env, METADATA=json.dumps(self.metadata), BODY=self.body.hex(), **env)
+        return subprocess.run([sys.executable, str(SCRIPT), "--repository", "owner/repo", "--tag", "v1.0", "--asset", "tool.zip", "--output", str(self.output), *extra], env=environment, capture_output=True, text=True)
+
+    def assert_failed_atomically(self, result):
+        self.assertNotEqual(result.returncode, 0)
+        self.assertEqual(self.output.read_bytes(), b"previous file")
+        self.assertEqual(list(self.root.glob(".release-asset-*")), [])
+        self.assertNotIn(self.env["GH_TOKEN"], result.stdout + result.stderr)
+
+    def test_valid_asset_uses_api_and_installs_only_verified_bytes(self):
+        result = self.invoke(("--sha256", self.sha256))
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertEqual(self.output.read_bytes(), self.body)
+        calls = [json.loads(line) for line in self.calls.read_text().splitlines()]
+        self.assertEqual(calls[0], ["api", "--hostname", "github.com", "repos/owner/repo/releases/tags/v1.0"])
+        self.assertEqual(calls[1], ["api", "--hostname", "github.com", "--header", "Accept: application/octet-stream", "repos/owner/repo/releases/assets/17"])
+        self.assertNotIn(self.env["GH_TOKEN"], json.dumps(calls) + result.stdout + result.stderr)
+
+    def test_download_failure_preserves_destination(self):
+        self.assert_failed_atomically(self.invoke(FAIL_DOWNLOAD="1"))
+
+    def test_missing_gh_fails_without_replacing_destination(self):
+        (self.root / "gh").unlink()
+        self.env["PATH"] = str(self.root)
+        self.assert_failed_atomically(self.invoke())
+
+    def test_metadata_failure_preserves_destination(self):
+        self.assert_failed_atomically(self.invoke(FAIL_METADATA="1"))
+
+    def test_corrupt_download_fails_digest_validation(self):
+        self.body = b"modified tool fixture"
+        self.assert_failed_atomically(self.invoke())
+
+    def test_truncated_download_fails_size_validation(self):
+        self.asset["size"] += 1
+        self.assert_failed_atomically(self.invoke())
+
+    def test_pinned_digest_cannot_be_replaced_by_release_metadata(self):
+        self.assert_failed_atomically(self.invoke(("--sha256", "0" * 64)))
+        self.assertEqual(len(self.calls.read_text().splitlines()), 1)
+
+    def test_ambiguous_or_missing_assets_fail_before_download(self):
+        for assets in ([], [self.asset, self.asset]):
+            with self.subTest(assets=len(assets)):
+                self.metadata["assets"] = assets
+                self.assert_failed_atomically(self.invoke())
+
+    def test_metadata_tag_must_match_requested_version(self):
+        self.metadata["tag_name"] = "v2.0"
+        self.assert_failed_atomically(self.invoke())
+
+    def test_malformed_release_inventory_fails_before_download(self):
+        for metadata in ([], {"tag_name": "v1.0", "assets": None}, {"tag_name": "v1.0", "assets": [None]}):
+            with self.subTest(metadata=metadata):
+                self.metadata = metadata
+                self.assert_failed_atomically(self.invoke())
+
+    def test_missing_digest_or_invalid_identity_fails_before_download(self):
+        for field, value in [("digest", None), ("digest", "sha256:invalid"), ("id", True), ("id", 0), ("size", 0), ("size", 513 * 1024 * 1024)]:
+            with self.subTest(field=field, value=value):
+                self.asset[field] = value
+                self.assert_failed_atomically(self.invoke())
+                self.asset = {"name": "tool.zip", "id": 17, "size": len(self.body), "digest": "sha256:" + self.sha256}
+                self.metadata["assets"] = [self.asset]
+
+
+if __name__ == "__main__":
+    unittest.main()
Author	SHA1	Message	Date
cxymds	1dea00ffa7	ci: repair e2e selection and verify tool downloads via API (#9240 ) * ci: align full e2e selection with retired catalog test * ci: verify pinned tools through GitHub release asset API * ci: fetch pinned typos through the release asset API * ci: download pinned actionlint through the release asset API * ci: ensure GitHub CLI exists before pinned mc download * ci: budget full quick checks after tool installation	2026-11-08 14:09:26 +00:00
cxymds	addf7a6219	fix: reduce scanner backoff and classify downstream closures (#9235 )	2026-11-08 02:18:17 -05:00
Ramakrishna Chilaka	420cf32b2a	perf(ecstore): optimize PUT, DELETE, and multipart commit hotpaths (#9227 )	2026-11-08 02:12:40 -05:00