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

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
-sha256-darwin=219ac11e720b3f54dfee4d5d5e95171785ba159b2fe01db3037a8f5061c1ad32
-sha256-linux=2d34698cc8ec52d5b6c37ac612919a1d21f36cd56a1af539a7522a4eea518e26
+sha256-darwin=9901a4950099b94d3d2618524d540f1ed4bf6023044903886f98b6c4bf5f30fb
+sha256-linux=09fd5f48f5556c9fc576357222dccd7ebb0b90b790798f1834db2ec5d67b4952

.config/make/tests.mak

@@ -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

.github/actions/quick-checks/action.yml

@@ -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"

.github/actions/setup/action.yml

@@ -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"

.github/actions/setup/download_release_asset.py

@@ -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())

.github/workflows/ci.yml

@@ -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"

Cargo.lock

@@ -10759,6 +10759,7 @@ dependencies = [
  "arc-swap",
  "async-compression",
  "async-trait",
+ "brotli 9.1.1",
  "bytes",
  "chrono",
  "crc-fast",
@@ -10768,6 +10769,7 @@ dependencies = [
  "futures-core",
  "hotpath",
  "http 1.5.0",
+ "lz4",
  "memchr",
  "metrics",
  "parking_lot 0.13.0",

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -127,9 +127,14 @@ const HTTP_PUT_FILE_CAPABILITY_AUTH_DOMAIN: &[u8] = b"rustfs-http-put-file-capab
 const UNSIGNED_PAYLOAD: &str = "UNSIGNED-PAYLOAD";
 const UNSIGNED_PAYLOAD_NONCE: &str = "unsigned";
 const SIGNATURE_VALID_DURATION: i64 = 300; // 5 minutes
-const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(601);
-const REPLAY_CACHE_RETENTION_SECS: usize = 601;
-const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: u64 = 128;
+// Timestamp validity and the wall-clock high-water mark still gate eviction,
+// including future-dated signatures and replay attempts after clock rollback.
+const REPLAY_CACHE_RETENTION: Duration = Duration::from_secs(301);
+// Preserve capacity for the full validity window at maximum accepted clock skew.
+const REPLAY_CACHE_CAPACITY_WINDOW_SECS: usize = 601;
+// A reserved index avoids overlapping old/new tables during a rehash. Include
+// index slack, compact expiry storage, and page/allocator bookkeeping.
+const REPLAY_CACHE_ENTRY_BYTES_ESTIMATE: usize = 2 * std::mem::size_of::<Uuid>() + std::mem::size_of::<RpcNonceExpiration>() + 16;
 // Keep 16 CPU / 32 GiB field nodes at the 32M cap without requiring an env override.
 const REPLAY_CACHE_AUTO_MEMORY_PERCENT: u64 = 13;
 const REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU: usize = 4096;
@@ -222,11 +227,11 @@ fn replay_cache_capacity_from_resources(cpu_count: usize, memory_limit_bytes: Op
     let cpu_count = cpu_count.max(1);
     let cpu_based_capacity = cpu_count
         .saturating_mul(REPLAY_CACHE_AUTO_RPC_RPS_PER_CPU)
-        .saturating_mul(REPLAY_CACHE_RETENTION_SECS);
+        .saturating_mul(REPLAY_CACHE_CAPACITY_WINDOW_SECS);
     let memory_based_capacity = memory_limit_bytes
         .map(|bytes| {
             let budget = bytes.saturating_mul(REPLAY_CACHE_AUTO_MEMORY_PERCENT) / 100;
-            saturating_usize_from_u64(budget / REPLAY_CACHE_ENTRY_BYTES_ESTIMATE)
+            saturating_usize_from_u64(budget / u64::try_from(REPLAY_CACHE_ENTRY_BYTES_ESTIMATE).unwrap_or(u64::MAX))
         })
         .unwrap_or(REPLAY_CACHE_AUTO_MAX_CAPACITY);
     let capacity = memory_based_capacity
@@ -497,6 +502,12 @@ fn publish_nonce_cache_metrics(metrics: Option<RpcNonceCacheMetrics<'_>>) {
 }
 
 impl RpcNonceCache {
+    fn reserve_index(&mut self, capacity: usize) -> std::io::Result<()> {
+        self.nonces
+            .try_reserve(capacity)
+            .map_err(|error| std::io::Error::new(std::io::ErrorKind::OutOfMemory, error))
+    }
+
     fn remove_expired(&mut self, now: Instant, wall_time: i64) -> usize {
         let mut removed = 0;
         while let Some(nonce) = self.expirations.pop_expired(now, wall_time) {
@@ -537,6 +548,11 @@ impl RpcNonceCache {
                 }),
             );
         }
+        if self.nonces.is_empty()
+            && let Err(error) = self.reserve_index(record.capacity.min(REPLAY_CACHE_AUTO_MAX_CAPACITY))
+        {
+            return (Err(error), Some(metrics));
+        }
         self.nonces.insert(record.nonce);
         self.expirations
             .push_back(record.expires_at, record.signed_at.saturating_add(SIGNATURE_VALID_DURATION), record.nonce);
@@ -1699,6 +1715,9 @@ pub fn verify_tonic_rpc_signature_with_bootstrap(
 }
 
 pub fn tonic_rpc_auth_failure_reason(error: &std::io::Error) -> &'static str {
+    if error.kind() == std::io::ErrorKind::OutOfMemory {
+        return "replay_cache_allocation";
+    }
     match error.to_string().as_str() {
         "Missing RPC audience" => "missing_audience",
         "Invalid RPC request path" => "invalid_request_path",
@@ -3138,9 +3157,9 @@ mod tests {
 
         assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
         assert_eq!(decision.memory_basis, Some(MemoryBasis::Host));
-        assert_eq!(decision.memory_based_capacity, 17_448_304);
+        assert_eq!(decision.memory_based_capacity, 31_019_208);
         assert_eq!(decision.cpu_based_capacity, 19_693_568);
-        assert_eq!(decision.capacity, 17_448_304);
+        assert_eq!(decision.capacity, 19_693_568);
     }
 
     #[test]
@@ -3150,13 +3169,13 @@ mod tests {
             replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(32 * gib), Some(MemoryBasis::Host));
 
         assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
-        assert_eq!(decision.memory_based_capacity, 34_896_609);
+        assert_eq!(decision.memory_based_capacity, 62_038_416);
         assert_eq!(decision.cpu_based_capacity, 39_387_136);
         assert_eq!(decision.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY);
 
         let observed_field_node =
             replay_cache_capacity_decision(rustfs_utils::EnvParseOutcome::Absent, 16, Some(31 * gib), Some(MemoryBasis::Host));
-        assert_eq!(observed_field_node.memory_based_capacity, 33_806_090);
+        assert_eq!(observed_field_node.memory_based_capacity, 60_099_715);
         assert_eq!(observed_field_node.cpu_based_capacity, 39_387_136);
         assert_eq!(observed_field_node.capacity, REPLAY_CACHE_AUTO_MAX_CAPACITY);
     }
@@ -3229,6 +3248,138 @@ mod tests {
         }
     }
 
+    #[test]
+    fn nonce_cache_releases_expired_timestamp_capacity_without_allowing_clock_rollback_replay() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let later = now + Duration::from_secs(302);
+        let nonce = Uuid::new_v4();
+        let replacement = Uuid::new_v4();
+        let mut cache = RpcNonceCache::default();
+        cache
+            .check_and_record(test_nonce_record(nonce, 1_000, now, 1_000, expiry, 1))
+            .0
+            .expect("record initial nonce");
+
+        let boundary = cache
+            .check_and_record(test_nonce_record(nonce, 1_000, later, 1_300, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("a signature at the inclusive freshness boundary must remain protected");
+        assert_eq!(boundary.to_string(), "RPC request replay detected");
+        cache
+            .check_and_record(test_nonce_record(replacement, 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect("an expired signature must release bounded cache capacity");
+        assert!(!cache.nonces.contains(&nonce));
+        assert!(cache.nonces.contains(&replacement));
+
+        let rollback = cache
+            .check_and_record(test_nonce_record(nonce, 1_000, later, 1_000, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("clock rollback must not resurrect an evicted signature");
+        assert_eq!(rollback.to_string(), "RPC request timestamp expired after clock regression");
+    }
+
+    #[test]
+    fn nonce_cache_keeps_future_timestamp_reserved_until_its_full_validity_window_closes() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let later = now + Duration::from_secs(302);
+        let nonce = Uuid::new_v4();
+        let mut cache = RpcNonceCache::default();
+        cache
+            .check_and_record(test_nonce_record(nonce, 1_300, now, 1_000, expiry, 1))
+            .0
+            .expect("record nonce signed at maximum allowed forward skew");
+        let replay = cache
+            .check_and_record(test_nonce_record(nonce, 1_300, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("monotonic expiry must not remove a still-valid future signature");
+        assert_eq!(replay.to_string(), "RPC request replay detected");
+        let full = cache
+            .check_and_record(test_nonce_record(Uuid::new_v4(), 1_302, later, 1_302, later + REPLAY_CACHE_RETENTION, 1))
+            .0
+            .expect_err("a still-valid nonce must continue consuming capacity");
+        assert_eq!(full.to_string(), "RPC replay cache capacity exceeded");
+        let expired = now + Duration::from_secs(602);
+        cache
+            .check_and_record(test_nonce_record(
+                Uuid::new_v4(),
+                1_602,
+                expired,
+                1_602,
+                expired + REPLAY_CACHE_RETENTION,
+                1,
+            ))
+            .0
+            .expect("release capacity only once future timestamp validity has closed");
+    }
+
+    #[test]
+    fn nonce_index_rejects_unallocatable_capacity_with_typed_error() {
+        let mut cache = RpcNonceCache::default();
+        let error = cache
+            .reserve_index(usize::MAX)
+            .expect_err("unallocatable capacity must fail without a panic");
+        assert_eq!(error.kind(), std::io::ErrorKind::OutOfMemory);
+        assert!(
+            error
+                .get_ref()
+                .is_some_and(|source| source.is::<std::collections::TryReserveError>())
+        );
+        assert_eq!(tonic_rpc_auth_failure_reason(&error), "replay_cache_allocation");
+        assert!(cache.nonces.is_empty());
+        assert!(cache.expirations.pages.is_empty());
+    }
+
+    #[test]
+    fn nonce_cache_reserves_the_bounded_index_before_sustained_traffic() {
+        let now = Instant::now();
+        let expiry = now + REPLAY_CACHE_RETENTION;
+        let mut cache = RpcNonceCache::default();
+        let capacity = 4096;
+        cache
+            .check_and_record(test_nonce_record(Uuid::from_u128(1), 1000, now, 1000, expiry, capacity))
+            .0
+            .expect("record first nonce");
+        let allocated = cache.nonces.capacity();
+        assert!(allocated >= capacity, "the hash index must not rehash while serving the bounded window");
+        for n in 2..=capacity {
+            cache
+                .check_and_record(test_nonce_record(
+                    Uuid::from_u128(u128::try_from(n).expect("nonce index")),
+                    1000,
+                    now,
+                    1000,
+                    expiry,
+                    capacity,
+                ))
+                .0
+                .expect("record within bounded capacity");
+        }
+        assert_eq!(cache.nonces.capacity(), allocated);
+        let full = cache
+            .check_and_record(test_nonce_record(Uuid::from_u128(9999), 1000, now, 1000, expiry, capacity))
+            .0
+            .expect_err("preallocation must not relax the logical capacity fence");
+        assert_eq!(full.to_string(), "RPC replay cache capacity exceeded");
+        assert_eq!(cache.nonces.len(), capacity);
+    }
+
+    #[test]
+    fn replay_cache_compact_storage_fits_two_cpu_window_in_2800_mib() {
+        let decision = replay_cache_capacity_decision(
+            rustfs_utils::EnvParseOutcome::Absent,
+            2,
+            Some(2800 * 1024 * 1024),
+            Some(MemoryBasis::Cgroup),
+        );
+        assert_eq!(decision.cpu_based_capacity, 4_923_392);
+        assert_eq!(decision.capacity, decision.cpu_based_capacity);
+        assert!(decision.memory_based_capacity >= decision.capacity);
+        assert_eq!(decision.source, ReplayCacheCapacitySource::Auto);
+    }
+
     #[test]
     fn nonce_cache_expires_by_monotonic_deadline_and_fails_closed_at_capacity() {
         let now = Instant::now();

crates/ecstore/src/disk/local.rs

@@ -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)

crates/ecstore/src/disk/local/commit.rs

@@ -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,

crates/ecstore/src/disk/local/delete_recovery.rs

@@ -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;

crates/ecstore/src/disk/local/multipart_index.rs

@@ -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);

crates/ecstore/src/disk/local/multipart_source_state.rs

@@ -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

crates/ecstore/src/disk/os.rs

@@ -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)?;

crates/object-data-cache/src/runtime_memory.rs

@@ -66,11 +66,26 @@ pub fn select_effective_memory(host_total: u64, host_available: u64, cgroup: Opt
 
 /// Resolves the effective memory from an already-refreshed system handle.
 ///
-/// `cgroup_limits()` is computed fresh on each call and is only implemented on
-/// Linux (it returns `None` elsewhere), so non-Linux hosts always use host
-/// values.
+/// On Linux, the current process cgroup and its ancestors are read on each call,
+/// including limits imposed by systemd services. System-wide cgroup limits are
+/// used if process information is unavailable. Other platforms use host values.
 pub fn effective_memory_from_system(system: &System) -> EffectiveMemory {
-    let cgroup = system.cgroup_limits().map(|limits| (limits.total_memory, limits.free_memory));
+    #[cfg(target_os = "linux")]
+    let limits = sysinfo::get_current_pid()
+        .ok()
+        .and_then(|pid| {
+            let mut processes = System::new().ok()?;
+            processes.refresh_processes_specifics(
+                sysinfo::ProcessesToUpdate::Some(&[pid]),
+                false,
+                sysinfo::ProcessRefreshKind::nothing().with_memory(),
+            );
+            processes.process(pid)?.cgroup_limits()
+        })
+        .or_else(|| system.cgroup_limits());
+    #[cfg(not(target_os = "linux"))]
+    let limits = system.cgroup_limits();
+    let cgroup = limits.map(|limits| (limits.total_memory, limits.free_memory));
     select_effective_memory(system.total_memory(), system.available_memory(), cgroup)
 }
 
@@ -89,6 +104,32 @@ mod tests {
 
     const GIB: u64 = 1024 * 1024 * 1024;
 
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn effective_memory_uses_current_process_cgroup() {
+        use sysinfo::{ProcessRefreshKind, ProcessesToUpdate, System, get_current_pid};
+
+        let pid = get_current_pid().expect("current process ID");
+        let mut system = System::new().expect("system memory handle");
+        system.refresh_memory();
+        let mut process_system = System::new().expect("process memory handle");
+        process_system.refresh_processes_specifics(
+            ProcessesToUpdate::Some(&[pid]),
+            false,
+            ProcessRefreshKind::nothing().with_memory(),
+        );
+        let limits = process_system.process(pid).expect("current process entry").cgroup_limits();
+        let expected = select_effective_memory(
+            system.total_memory(),
+            system.available_memory(),
+            limits.map(|limits| (limits.total_memory, limits.free_memory)),
+        );
+        let actual = super::effective_memory_from_system(&system);
+        assert_eq!(actual.basis, expected.basis);
+        assert_eq!(actual.total_bytes, expected.total_bytes);
+        assert!(actual.available_bytes <= actual.total_bytes);
+    }
+
     #[test]
     fn select_effective_memory_prefers_constraining_cgroup() {
         let effective = select_effective_memory(64 * GIB, 40 * GIB, Some((2 * GIB, GIB)));

crates/s3select-api/Cargo.toml

@@ -73,6 +73,8 @@ rustfs-storage-api.workspace = true
 futures = { workspace = true }
 futures-core = { workspace = true }
 flate2.workspace = true
+brotli.workspace = true
+lz4.workspace = true
 http.workspace = true
 memchr.workspace = true
 s3s = { workspace = true, features = ["minio"] }

crates/s3select-api/src/lib.rs

@@ -27,6 +27,7 @@ mod csv_input;
 mod input_stream;
 mod metrics;
 pub mod object_store;
+pub mod parquet_budget;
 pub mod query;
 pub mod server;
 mod storage_api;
@@ -218,6 +219,14 @@ fn classify_select_error_source(err: &(dyn StdError + 'static)) -> Option<Select
     if let Some(error) = err.downcast_ref::<SelectError>() {
         return Some(error.clone());
     }
+    if let Some(error) = err.downcast_ref::<parquet_budget::ParquetBudgetError>() {
+        return Some(match error {
+            parquet_budget::ParquetBudgetError::Exceeded => SelectError::ResourceExhausted,
+            parquet_budget::ParquetBudgetError::Invalid(_) | parquet_budget::ParquetBudgetError::Codec(_) => {
+                SelectError::ParquetParsingError
+            }
+        });
+    }
     if let Some(error) = err.downcast_ref::<object_store::SelectObjectStoreError>() {
         return Some(error.select_error());
     }
@@ -332,9 +341,32 @@ impl Display for ResolvedTable {
 #[cfg(test)]
 mod tests {
     use super::*;
+
     use datafusion::common::DataFusionError;
     use datafusion::sql::sqlparser::parser::ParserError;
 
+    #[test]
+    fn parquet_admission_errors_preserve_select_error_codes() {
+        for (source, expected) in [
+            (parquet_budget::ParquetBudgetError::Exceeded, SelectError::ResourceExhausted),
+            (
+                parquet_budget::ParquetBudgetError::Invalid("malformed footer".into()),
+                SelectError::ParquetParsingError,
+            ),
+            (
+                parquet_budget::ParquetBudgetError::Codec(std::io::Error::from(std::io::ErrorKind::InvalidData)),
+                SelectError::ParquetParsingError,
+            ),
+        ] {
+            let error = QueryError::from(DataFusionError::External(Box::new(source)));
+            assert_eq!(error.select_error(), expected);
+        }
+        let nested = QueryError::from(DataFusionError::ParquetError(Box::new(ParquetError::External(Box::new(
+            parquet_budget::ParquetBudgetError::Exceeded,
+        )))));
+        assert_eq!(nested.select_error(), SelectError::ResourceExhausted);
+    }
+
     #[test]
     fn test_query_error_display() {
         let err = QueryError::NotImplemented {

crates/s3select-query/src/dispatcher/manager.rs

@@ -424,7 +424,8 @@ impl SimpleQueryDispatcher {
         tracker: &QueryExecutionTracker,
     ) -> QueryResult<MetadataProvider> {
         if self.input.request.input_serialization.parquet.is_some() {
-            let provider = ParquetSelectTable::try_new(session.inner(), self.input.as_ref()).await?;
+            let query_guard = tracker.query_guard().ok_or(QueryError::Cancel)?;
+            let provider = ParquetSelectTable::try_new(session.inner(), self.input.as_ref(), Some(query_guard)).await?;
             let current_session_table_provider = self.build_table_handle_provider()?;
             let metadata_provider =
                 MetadataProvider::new(provider, current_session_table_provider, self.func_manager.clone(), session.clone());

crates/s3select-query/src/dispatcher/parquet_table.rs

@@ -27,7 +27,10 @@ use datafusion::{
     datasource::{
         TableProvider,
         listing::{ListingTableUrl, PartitionedFile},
-        physical_plan::{FileScanConfigBuilder, ParquetSource, parquet::ParquetAccessPlan},
+        physical_plan::{
+            FileScanConfigBuilder, ParquetSource,
+            parquet::{ParquetAccessPlan, ParquetFileReaderFactory},
+        },
         source::DataSourceExec,
     },
     execution::{memory_pool::MemoryLimit, object_store::ObjectStoreUrl},
@@ -40,6 +43,7 @@ use datafusion::{
     },
     physical_plan::{
         ExecutionPlan, SendableRecordBatchStream,
+        metrics::ExecutionPlanMetricsSet,
         stream::RecordBatchStreamAdapter,
         streaming::{PartitionStream, StreamingTableExec},
     },
@@ -48,6 +52,8 @@ use futures::{FutureExt, TryFutureExt, TryStreamExt, future::BoxFuture, stream};
 use rustfs_s3select_api::{
     QueryResult,
     object_store::{SelectScanRange, scan_range_from_bounds},
+    parquet_budget::{ParquetBudgetError, page_decode_budget, validate_footer},
+    query::session::QueryExecutionGuard,
 };
 use s3s::dto::SelectObjectContentInput;
 
@@ -59,6 +65,7 @@ pub struct ParquetSelectTable {
     object_size: u64,
     access_plan: Option<Arc<ParquetAccessPlan>>,
     expanded_row_budget: usize,
+    reader_factory: Arc<BudgetParquetReaderFactory>,
 }
 
 impl fmt::Debug for ParquetSelectTable {
@@ -72,11 +79,30 @@ impl fmt::Debug for ParquetSelectTable {
     }
 }
 
+#[derive(Debug)]
 struct ObjectStoreParquetReader {
     store: Arc<dyn ObjectStore>,
     path: Path,
     file_size: u64,
     metadata_read_limit: usize,
+    metadata_decode_limit: usize,
+    metadata: Option<Arc<ParquetMetaData>>,
+    query_guard: Option<QueryExecutionGuard>,
+}
+
+async fn validate_in_worker<T: Send + 'static>(
+    query_guard: Option<QueryExecutionGuard>,
+    operation: impl FnOnce() -> Result<T, ParquetBudgetError> + Send + 'static,
+) -> ParquetResult<T> {
+    tokio::task::spawn_blocking(move || {
+        // Cancellation cannot release admission while the decoder preflight
+        // still owns input buffers or codec allocations.
+        let _query_guard = query_guard;
+        operation()
+    })
+    .await
+    .map_err(|error| ParquetError::External(Box::new(error)))?
+    .map_err(|error| ParquetError::External(Box::new(error)))
 }
 
 impl AsyncFileReader for ObjectStoreParquetReader {
@@ -94,7 +120,16 @@ impl AsyncFileReader for ObjectStoreParquetReader {
         if total.is_none_or(|total| total > self.metadata_read_limit as u64) {
             return async { Err(ParquetError::External(Box::new(parquet_budget_error()))) }.boxed();
         }
-        async move { self.store.get_ranges(&self.path, &ranges).await.map_err(parquet_store_error) }.boxed()
+        async move {
+            // Read requested ranges directly so sparse offsets cannot amplify
+            // a small request into a large coalesced allocation.
+            let mut bytes = Vec::with_capacity(ranges.len());
+            for range in ranges {
+                bytes.push(self.store.get_range(&self.path, range).await.map_err(parquet_store_error)?);
+            }
+            Ok(bytes)
+        }
+        .boxed()
     }
 
     fn get_metadata<'a>(
@@ -102,6 +137,27 @@ impl AsyncFileReader for ObjectStoreParquetReader {
         options: Option<&'a ArrowReaderOptions>,
     ) -> BoxFuture<'a, ParquetResult<Arc<ParquetMetaData>>> {
         async move {
+            if let Some(metadata) = self.metadata.as_ref() {
+                return Ok(Arc::clone(metadata));
+            }
+            let tail_start = self
+                .file_size
+                .checked_sub(8)
+                .ok_or_else(|| ParquetError::EOF("Parquet footer is truncated".into()))?;
+            let tail = self.get_bytes(tail_start..self.file_size).await?;
+            let tail: &[u8; 8] = tail
+                .as_ref()
+                .try_into()
+                .map_err(|_| ParquetError::EOF("Parquet footer is truncated".into()))?;
+            let footer = datafusion::parquet::file::metadata::FooterTail::try_new(tail)?;
+            let length =
+                u64::try_from(footer.metadata_length()).map_err(|_| ParquetError::External(Box::new(parquet_budget_error())))?;
+            let start = tail_start
+                .checked_sub(length)
+                .ok_or_else(|| ParquetError::EOF("Parquet metadata is truncated".into()))?;
+            let bytes = self.get_bytes(start..tail_start).await?;
+            let limit = self.metadata_decode_limit;
+            validate_in_worker(self.query_guard.clone(), move || validate_footer(&bytes, limit)).await?;
             let metadata_options = options.map(|options| options.metadata_options().clone());
             let mut metadata_reader = ParquetMetaDataReader::new().with_metadata_options(metadata_options);
 
@@ -119,8 +175,84 @@ impl AsyncFileReader for ObjectStoreParquetReader {
     }
 }
 
+#[derive(Debug)]
+struct BudgetParquetReaderFactory {
+    store: Arc<dyn ObjectStore>,
+    path: Path,
+    file_size: u64,
+    metadata: Arc<ParquetMetaData>,
+    read_limit: usize,
+}
+
+impl ParquetFileReaderFactory for BudgetParquetReaderFactory {
+    fn create_reader(
+        &self,
+        _partition_index: usize,
+        file: PartitionedFile,
+        _metadata_size_hint: Option<usize>,
+        _metrics: &ExecutionPlanMetricsSet,
+    ) -> DFResult<Box<dyn AsyncFileReader + Send>> {
+        if file.object_meta.location != self.path || file.object_meta.size != self.file_size {
+            return Err(DataFusionError::Execution("Parquet reader input differs from its admitted object".into()));
+        }
+        Ok(Box::new(ObjectStoreParquetReader {
+            store: Arc::clone(&self.store),
+            path: self.path.clone(),
+            file_size: self.file_size,
+            metadata_read_limit: self.read_limit,
+            metadata_decode_limit: self.read_limit,
+            metadata: Some(Arc::clone(&self.metadata)),
+            query_guard: None,
+        }))
+    }
+}
+
+async fn validate_parquet_pages(
+    store: &Arc<dyn ObjectStore>,
+    path: &Path,
+    file_size: u64,
+    metadata: &ParquetMetaData,
+    plan: Option<&ParquetAccessPlan>,
+    limit: usize,
+    query_guard: &Option<QueryExecutionGuard>,
+) -> DFResult<()> {
+    for (index, group) in metadata.row_groups().iter().enumerate() {
+        if plan.is_some_and(|plan| !plan.should_scan(index)) {
+            continue;
+        }
+        let mut resident = 0usize;
+        for column in group.columns() {
+            let start = u64::try_from(column.dictionary_page_offset().unwrap_or(column.data_page_offset()))
+                .map_err(|_| parquet_budget_error())?;
+            let length = u64::try_from(column.compressed_size()).map_err(|_| parquet_budget_error())?;
+            let end = start
+                .checked_add(length)
+                .filter(|end| *end <= file_size)
+                .ok_or_else(parquet_budget_error)?;
+            if length > u64::try_from(limit).map_err(|_| parquet_budget_error())? {
+                return Err(parquet_budget_error());
+            }
+            let bytes = store.get_range(path, start..end).await?;
+            let declared = usize::try_from(column.uncompressed_size()).map_err(|_| parquet_budget_error())?;
+            let codec = column.compression();
+            let required = validate_in_worker(query_guard.clone(), move || page_decode_budget(&bytes, declared, limit, codec))
+                .await
+                .map_err(DataFusionError::from)?;
+            resident = resident
+                .checked_add(required)
+                .filter(|total| *total <= limit)
+                .ok_or_else(parquet_budget_error)?;
+        }
+    }
+    Ok(())
+}
+
 impl ParquetSelectTable {
-    pub async fn try_new(state: &dyn Session, input: &SelectObjectContentInput) -> QueryResult<Arc<dyn TableProvider>> {
+    pub async fn try_new(
+        state: &dyn Session,
+        input: &SelectObjectContentInput,
+        query_guard: Option<QueryExecutionGuard>,
+    ) -> QueryResult<Arc<dyn TableProvider>> {
         let table_path = ListingTableUrl::parse(format!("s3://{}/{}", input.bucket, input.key))?;
         let object_store_url = table_path.object_store();
         let object_location = Path::from(input.key.clone());
@@ -136,9 +268,12 @@ impl ParquetSelectTable {
         };
         let reader = ObjectStoreParquetReader {
             store: Arc::clone(&store),
-            path: object_location,
+            path: object_location.clone(),
             file_size: object_meta.size,
             metadata_read_limit: (memory_budget / 16).max(1),
+            metadata_decode_limit: (memory_budget / 4).max(1),
+            metadata: None,
+            query_guard: query_guard.clone(),
         };
         let builder = ParquetRecordBatchStreamBuilder::new(reader)
             .await
@@ -153,11 +288,26 @@ impl ParquetSelectTable {
             builder.schema().metadata().clone(),
         ));
         let metadata = Arc::clone(builder.metadata());
-        // This is admission for valid Parquet declarations, not a replacement
-        // for limits inside the footer parser or native compression codecs.
         let expanded_row_budget = (memory_budget / 16).max(1);
         let access_plan = parquet_access_plan(input, object_meta.size, metadata.as_ref())?;
         admit_parquet_decode(metadata.as_ref(), access_plan.as_deref(), memory_budget / 4)?;
+        validate_parquet_pages(
+            &store,
+            &object_location,
+            object_meta.size,
+            &metadata,
+            access_plan.as_deref(),
+            memory_budget / 4,
+            &query_guard,
+        )
+        .await?;
+        let reader_factory = Arc::new(BudgetParquetReaderFactory {
+            store,
+            path: object_location,
+            file_size: object_meta.size,
+            metadata,
+            read_limit: (memory_budget / 4).max(1),
+        });
 
         Ok(Arc::new(Self {
             schema,
@@ -166,6 +316,7 @@ impl ParquetSelectTable {
             object_size: object_meta.size,
             access_plan,
             expanded_row_budget,
+            reader_factory,
         }))
     }
 
@@ -202,7 +353,12 @@ impl TableProvider for ParquetSelectTable {
             ));
         }
         let scan_limit = if filters.is_empty() { limit } else { None };
-        let file_source = Arc::new(ParquetSource::new(Arc::clone(&self.schema)));
+        let file_source = Arc::new(
+            ParquetSource::new(Arc::clone(&self.schema))
+                .with_parquet_file_reader_factory(self.reader_factory.clone())
+                .with_enable_page_index(false)
+                .with_bloom_filter_on_read(false),
+        );
         let config = FileScanConfigBuilder::new(self.object_store_url.clone(), file_source)
             .with_file(self.partitioned_file())
             .with_projection_indices(projection.cloned())?
@@ -439,6 +595,39 @@ pub(super) fn dictionary_parquet_for_test(value_bytes: usize, rows: usize) -> Ve
 #[cfg(test)]
 mod tests {
     use super::*;
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn parquet_cancelled_preflight_retains_admission_until_worker_finishes() {
+        let admission = Arc::new(tokio::sync::Semaphore::new(1));
+        let permit = Arc::clone(&admission).acquire_owned().await.expect("initial admission");
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let task = tokio::spawn(validate_in_worker(Some(Arc::new(permit)), move || {
+            started_tx.send(()).expect("report running worker");
+            release_rx
+                .recv_timeout(std::time::Duration::from_secs(5))
+                .expect("worker release");
+            Ok(())
+        }));
+        tokio::time::timeout(std::time::Duration::from_secs(5), started_rx)
+            .await
+            .expect("worker starts before deadline")
+            .expect("worker started");
+        task.abort();
+        assert!(task.await.expect_err("cancelled request task").is_cancelled());
+        assert!(Arc::clone(&admission).try_acquire_owned().is_err(), "running worker still owns admission");
+        release_tx.send(()).expect("release worker");
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            loop {
+                if Arc::clone(&admission).try_acquire_owned().is_ok() {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("worker releases admission after completion");
+    }
     use datafusion::{
         arrow::{
             array::Int32Array,
@@ -519,7 +708,7 @@ mod tests {
         let context = parquet_session(store);
         let state = context.state();
 
-        let error = match ParquetSelectTable::try_new(&state, &parquet_input("missing.parquet")).await {
+        let error = match ParquetSelectTable::try_new(&state, &parquet_input("missing.parquet"), None).await {
             Ok(_) => panic!("missing parquet object must fail"),
             Err(error) => error,
         };
@@ -538,7 +727,7 @@ mod tests {
         let context = parquet_session(store);
         let state = context.state();
 
-        let error = match ParquetSelectTable::try_new(&state, &parquet_input(object.as_ref())).await {
+        let error = match ParquetSelectTable::try_new(&state, &parquet_input(object.as_ref()), None).await {
             Ok(_) => panic!("corrupt parquet metadata must fail"),
             Err(error) => error,
         };
@@ -581,7 +770,7 @@ mod tests {
             .put(&Path::from("dictionary.parquet"), Bytes::from(bytes).into())
             .await
             .expect("fixture");
-        let table = ParquetSelectTable::try_new(&context.state(), &parquet_input("dictionary.parquet"))
+        let table = ParquetSelectTable::try_new(&context.state(), &parquet_input("dictionary.parquet"), None)
             .await
             .expect("valid dictionary metadata");
         context.register_table("source", table).expect("table");
@@ -752,7 +941,7 @@ mod tests {
             start: Some(starts[0]),
             end: Some(starts[0]),
         });
-        let table = ParquetSelectTable::try_new(&context.state(), &input)
+        let table = ParquetSelectTable::try_new(&context.state(), &input, None)
             .await
             .expect("an unselected large row group must not reject the small range");
         context.register_table("source", table).expect("table");
@@ -770,7 +959,7 @@ mod tests {
             start: Some(starts[1]),
             end: Some(starts[1]),
         });
-        let error = match ParquetSelectTable::try_new(&context.state(), &input).await {
+        let error = match ParquetSelectTable::try_new(&context.state(), &input, None).await {
             Ok(_) => panic!("the selected oversized row group must still be rejected before decoding"),
             Err(error) => error,
         };
@@ -806,13 +995,117 @@ mod tests {
             .put(&Path::from("large.parquet"), Bytes::from(bytes).into())
             .await
             .expect("fixture");
-        let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("large.parquet")).await {
+        let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("large.parquet"), None).await {
             Ok(_) => panic!("valid declared decoded size must be admitted before page decode"),
             Err(error) => error,
         };
         assert!(error.to_string().contains("query memory budget"));
     }
 
+    #[tokio::test]
+    async fn parquet_footer_collection_budget_is_checked_before_metadata_decode() {
+        // A root schema with no leaves and 50 empty, valid row groups. The
+        // encoded footer fits the read budget; its reservations do not.
+        let mut footer = vec![0x15, 2, 0x19, 0x1c, 0x48, 4];
+        footer.extend(b"root");
+        footer.extend([0x15, 0, 0, 0x16, 0, 0x19, 0xfc, 50]);
+        for _ in 0..50 {
+            footer.extend([0x19, 0x0c, 0x16, 0, 0x16, 0, 0]);
+        }
+        footer.push(0);
+        let decoded = ParquetMetaDataReader::decode_metadata(&footer).expect("valid compact footer fixture");
+        assert_eq!(decoded.num_row_groups(), 50);
+        let mut bytes = b"PAR1".to_vec();
+        bytes.extend(&footer);
+        bytes.extend(u32::try_from(footer.len()).expect("small footer").to_le_bytes());
+        bytes.extend(b"PAR1");
+        let store = Arc::new(InMemory::new());
+        let context = parquet_session_with_memory(store.clone(), 8192);
+        store
+            .put(&Path::from("footer.parquet"), Bytes::from(bytes).into())
+            .await
+            .expect("fixture");
+        let error = match ParquetSelectTable::try_new(&context.state(), &parquet_input("footer.parquet"), None).await {
+            Ok(_) => panic!("footer collection reservations exceed the query budget"),
+            Err(error) => error,
+        };
+        assert_eq!(error.select_error(), SelectError::ResourceExhausted);
+    }
+
+    #[tokio::test]
+    async fn parquet_forged_footer_cannot_hide_oversized_compressed_pages() {
+        use datafusion::parquet::file::metadata::ParquetMetaDataWriter;
+
+        let schema = Arc::new(Schema::new(vec![Field::new("v", DataType::Utf8, false)]));
+        let batch = RecordBatch::try_new(
+            Arc::clone(&schema),
+            vec![Arc::new(datafusion::arrow::array::StringArray::from(vec![
+                "x".repeat(32 * 1024 * 1024),
+            ]))],
+        )
+        .expect("large compressible value");
+        let properties = datafusion::parquet::file::properties::WriterProperties::builder()
+            .set_dictionary_page_size_limit(64 * 1024 * 1024)
+            .set_compression(datafusion::parquet::basic::Compression::ZSTD(Default::default()))
+            .build();
+        let mut encoded = Vec::new();
+        let mut writer = ArrowWriter::try_new(&mut encoded, schema, Some(properties)).expect("writer");
+        writer.write(&batch).expect("compressed value");
+        writer.close().expect("footer");
+        let original = Bytes::from(encoded);
+        let builder = ParquetRecordBatchReaderBuilder::try_new(original.clone()).expect("valid compressed fixture");
+        let metadata = builder.metadata();
+        let groups = metadata
+            .row_groups()
+            .iter()
+            .map(|group| {
+                let columns = group
+                    .columns()
+                    .iter()
+                    .cloned()
+                    .map(|column| {
+                        column
+                            .into_builder()
+                            .set_total_uncompressed_size(64)
+                            .build()
+                            .expect("forged column")
+                    })
+                    .collect();
+                group
+                    .clone()
+                    .into_builder()
+                    .set_column_metadata(columns)
+                    .build()
+                    .expect("forged group")
+            })
+            .collect();
+        let forged = metadata.as_ref().clone().into_builder().set_row_groups(groups).build();
+        let footer_size = u32::from_le_bytes(
+            original[original.len() - 8..original.len() - 4]
+                .try_into()
+                .expect("footer size"),
+        );
+        let mut bytes = original[..original.len() - 8 - usize::try_from(footer_size).expect("footer length")].to_vec();
+        ParquetMetaDataWriter::new(&mut bytes, &forged)
+            .finish()
+            .expect("forged footer");
+        assert!(bytes.len() < 256 * 1024, "compressed input remains small");
+        admit_parquet_decode(&forged, None, 4 * 1024 * 1024).expect("footer-only admission misses the page size");
+
+        let store = Arc::new(InMemory::new());
+        let context = parquet_session_with_memory(store.clone(), 16 * 1024 * 1024);
+        store
+            .put(&Path::from("forged.parquet"), Bytes::from(bytes).into())
+            .await
+            .expect("fixture");
+        let result = ParquetSelectTable::try_new(&context.state(), &parquet_input("forged.parquet"), None).await;
+        let error = match result {
+            Ok(_) => panic!("actual query admission must reject oversized page buffers before decoding"),
+            Err(error) => error,
+        };
+        assert_eq!(error.select_error(), SelectError::ResourceExhausted);
+    }
+
     #[tokio::test]
     async fn parquet_nested_views_decode_through_the_actual_file_schema() {
         use datafusion::arrow::array::{ListArray, StringArray};

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "nixpkgs": {
       "locked": {
-        "lastModified": 1793466162,
-        "narHash": "sha256-pt+peAEoOK3lLLdtbnQ1AGIBTZbdB6jrkgmurfcQwig=",
+        "lastModified": 1793977579,
+        "narHash": "sha256-WrSODdZFVIs/Hh2NzC0bY+ROByi0bATg8WfePX8dPj8=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "a0788123ada57ec92a79ab2581cda6d3c6cca75f",
+        "rev": "4fe25e12eb9f4c744966dd3de5a05246cf098601",
         "type": "github"
       },
       "original": {
@@ -29,11 +29,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1793434469,
-        "narHash": "sha256-+sgVNI2c67di+XIeMaz5KSei4T65kX68s5Nl0E5Gzbc=",
+        "lastModified": 1794040562,
+        "narHash": "sha256-7eEiezSku3lbZPUOXW8we1a7+xd5axkCQdyI3rzOKjo=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "6ecf534a7c16e7c969f26838fdd8c93597ae93f3",
+        "rev": "d3fe74269f2a6652437e21e78eb7e83be81f6982",
         "type": "github"
       },
       "original": {

rustfs/src/app/object/put.rs

@@ -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);
 

rustfs/src/app/object/request_body.rs

@@ -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;

rustfs/src/table_catalog/maintenance/compaction.rs

@@ -21,8 +21,7 @@ use datafusion::arrow::datatypes::{DataType, Field, Schema};
 use datafusion::parquet::arrow::{
     add_encoded_arrow_schema_to_metadata, arrow_reader::ArrowReaderOptions, arrow_writer::ArrowWriterOptions,
 };
-use datafusion::parquet::file::FOOTER_SIZE;
-use datafusion::parquet::file::metadata::{ColumnChunkMetaData, FooterTail, RowGroupMetaData};
+use datafusion::parquet::file::metadata::ColumnChunkMetaData;
 use datafusion::parquet::file::properties::{EnabledStatistics, WriterProperties};
 use std::io::Write;
 use std::sync::OnceLock;
@@ -74,279 +73,54 @@ fn add_budget(total: &mut usize, amount: usize, limit: usize) -> TableCatalogSto
     Ok(())
 }
 
-/// Sizes read from one Parquet page header (Thrift compact encoding).
-#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
-struct PageHeaderSizes {
-    uncompressed: usize,
-    compressed: usize,
-}
-
+#[cfg(test)]
 const THRIFT_COMPACT_STOP: u8 = 0;
+#[cfg(test)]
 const THRIFT_COMPACT_BOOL_TRUE: u8 = 1;
-const THRIFT_COMPACT_BOOL_FALSE: u8 = 2;
-const THRIFT_COMPACT_BYTE: u8 = 3;
+#[cfg(test)]
 const THRIFT_COMPACT_I16: u8 = 4;
+#[cfg(test)]
 const THRIFT_COMPACT_I32: u8 = 5;
+#[cfg(test)]
 const THRIFT_COMPACT_I64: u8 = 6;
-const THRIFT_COMPACT_DOUBLE: u8 = 7;
+#[cfg(test)]
 const THRIFT_COMPACT_BINARY: u8 = 8;
+#[cfg(test)]
 const THRIFT_COMPACT_LIST: u8 = 9;
-const THRIFT_COMPACT_SET: u8 = 10;
+#[cfg(test)]
 const THRIFT_COMPACT_MAP: u8 = 11;
+#[cfg(test)]
 const THRIFT_COMPACT_STRUCT: u8 = 12;
-const THRIFT_MAX_NESTING: usize = 16;
-const PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE: i16 = 2;
-const PAGE_HEADER_FIELD_COMPRESSED_SIZE: i16 = 3;
-const FILE_METADATA_FIELD_ROW_GROUPS: i16 = 4;
 
-/// Minimal Thrift compact-protocol cursor: enough to read a `PageHeader`'s
-/// size fields and skip everything else so the next header can be located.
-struct ThriftCompactCursor<'a> {
-    bytes: &'a [u8],
-    pos: usize,
-}
-
-impl ThriftCompactCursor<'_> {
-    fn error() -> TableCatalogStoreError {
-        TableCatalogStoreError::Invalid("compaction parquet thrift metadata is malformed".to_owned())
-    }
-
-    fn byte(&mut self) -> TableCatalogStoreResult<u8> {
-        let byte = *self.bytes.get(self.pos).ok_or_else(Self::error)?;
-        self.pos += 1;
-        Ok(byte)
-    }
-
-    fn skip_bytes(&mut self, count: usize) -> TableCatalogStoreResult<()> {
-        self.pos = self
-            .pos
-            .checked_add(count)
-            .filter(|end| *end <= self.bytes.len())
-            .ok_or_else(Self::error)?;
-        Ok(())
-    }
-
-    fn varint(&mut self) -> TableCatalogStoreResult<u64> {
-        let mut value = 0_u64;
-        for shift in (0..64).step_by(7) {
-            let byte = self.byte()?;
-            value |= u64::from(byte & 0x7f) << shift;
-            if byte & 0x80 == 0 {
-                return Ok(value);
-            }
-        }
-        Err(Self::error())
-    }
-
-    fn zigzag(&mut self) -> TableCatalogStoreResult<i64> {
-        let raw = self.varint()?;
-        Ok(((raw >> 1) as i64) ^ -((raw & 1) as i64))
-    }
-
-    /// Every collection element occupies at least one byte, so a declared
-    /// count beyond the remaining bytes is malformed. This bounds the walk to
-    /// the chunk's size instead of the count a hostile header declares.
-    fn collection_count(&self, count: u64) -> TableCatalogStoreResult<usize> {
-        usize::try_from(count)
-            .ok()
-            .filter(|count| *count <= self.bytes.len() - self.pos)
-            .ok_or_else(Self::error)
-    }
-
-    fn list_header(&mut self) -> TableCatalogStoreResult<(u8, usize)> {
-        let header = self.byte()?;
-        let count = match header >> 4 {
-            15 => self.varint()?,
-            short => u64::from(short),
-        };
-        Ok((header & 0x0f, self.collection_count(count)?))
-    }
-
-    fn field_id(&mut self, last: i16, header: u8) -> TableCatalogStoreResult<i16> {
-        match header >> 4 {
-            0 => i16::try_from(self.zigzag()?).map_err(|_| Self::error()),
-            delta => last.checked_add(i16::from(delta)).ok_or_else(Self::error),
-        }
-    }
-
-    /// parquet reads the page size fields by id as a zigzag varint whatever
-    /// the wire kind says. Follow the id; a kind that is not an integer is
-    /// rejected rather than skipped, so the walker never records a smaller
-    /// size than the decoder will allocate.
-    fn size_field(&mut self, kind: u8) -> TableCatalogStoreResult<usize> {
-        match kind {
-            THRIFT_COMPACT_I16 | THRIFT_COMPACT_I32 | THRIFT_COMPACT_I64 => {
-                usize::try_from(self.zigzag()?).map_err(|_| Self::error())
-            }
-            _ => Err(Self::error()),
-        }
-    }
-
-    fn skip_value(&mut self, kind: u8, depth: usize) -> TableCatalogStoreResult<()> {
-        if depth > THRIFT_MAX_NESTING {
-            return Err(Self::error());
-        }
-        match kind {
-            // A struct field's bool value lives in its header byte.
-            THRIFT_COMPACT_BOOL_TRUE | THRIFT_COMPACT_BOOL_FALSE => Ok(()),
-            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 => {
-                let len = usize::try_from(self.varint()?).map_err(|_| Self::error())?;
-                self.skip_bytes(len)
-            }
-            THRIFT_COMPACT_LIST | THRIFT_COMPACT_SET => {
-                let (element, count) = self.list_header()?;
-                for _ in 0..count {
-                    self.skip_element(element, depth + 1)?;
-                }
-                Ok(())
-            }
-            THRIFT_COMPACT_MAP => {
-                let count = self.varint()?;
-                let count = self.collection_count(count)?;
-                if count == 0 {
-                    return Ok(());
-                }
-                let kinds = self.byte()?;
-                for _ in 0..count {
-                    self.skip_element(kinds >> 4, depth + 1)?;
-                    self.skip_element(kinds & 0x0f, depth + 1)?;
-                }
-                Ok(())
-            }
-            THRIFT_COMPACT_STRUCT => self.skip_struct(depth + 1),
-            _ => Err(Self::error()),
-        }
-    }
-
-    /// A collection element carries its own bytes, but parquet skips a bool
-    /// element without consuming any. Either width would let the walker and
-    /// the decoder disagree on where the next header starts; no Parquet
-    /// struct holds a bool collection, so fail closed instead.
-    fn skip_element(&mut self, kind: u8, depth: usize) -> TableCatalogStoreResult<()> {
-        match kind {
-            THRIFT_COMPACT_BOOL_TRUE | THRIFT_COMPACT_BOOL_FALSE => Err(Self::error()),
-            kind => self.skip_value(kind, depth),
-        }
-    }
-
-    fn skip_struct(&mut self, depth: usize) -> TableCatalogStoreResult<()> {
-        let mut field_id = 0_i16;
-        loop {
-            let header = self.byte()?;
-            if header == THRIFT_COMPACT_STOP {
-                return Ok(());
-            }
-            field_id = self.field_id(field_id, header)?;
-            self.skip_value(header & 0x0f, depth)?;
-        }
-    }
-
-    /// Read a `PageHeader` struct, returning its sizes and leaving the cursor at
-    /// the first page byte.
-    fn page_header(&mut self) -> TableCatalogStoreResult<PageHeaderSizes> {
-        let mut sizes = PageHeaderSizes::default();
-        let mut field_id = 0_i16;
-        loop {
-            let header = self.byte()?;
-            if header == THRIFT_COMPACT_STOP {
-                return Ok(sizes);
-            }
-            let kind = header & 0x0f;
-            field_id = self.field_id(field_id, header)?;
-            match field_id {
-                PAGE_HEADER_FIELD_UNCOMPRESSED_SIZE => sizes.uncompressed = self.size_field(kind)?,
-                PAGE_HEADER_FIELD_COMPRESSED_SIZE => sizes.compressed = self.size_field(kind)?,
-                _ => self.skip_value(kind, 1)?,
-            }
-        }
-    }
-
-    /// 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) -> TableCatalogStoreResult<()> {
-        let mut field_id = 0_i16;
-        let mut reserved = 0_usize;
-        loop {
-            let header = self.byte()?;
-            if header == THRIFT_COMPACT_STOP {
-                return Ok(());
-            }
-            let kind = header & 0x0f;
-            field_id = self.field_id(field_id, header)?;
-            if field_id == FILE_METADATA_FIELD_ROW_GROUPS && kind == THRIFT_COMPACT_LIST {
-                let (element, count) = self.list_header()?;
-                if element != THRIFT_COMPACT_STRUCT {
-                    return Err(Self::error());
-                }
-                let reservation = count.checked_mul(size_of::<RowGroupMetaData>()).ok_or_else(Self::error)?;
-                add_budget(&mut reserved, reservation, limit)?;
-                for _ in 0..count {
-                    self.skip_struct(2)?;
-                }
-            } else {
-                self.skip_value(kind, 0)?;
-            }
-        }
+fn parquet_admission_error(error: rustfs_s3select_api::parquet_budget::ParquetBudgetError) -> TableCatalogStoreError {
+    match error {
+        rustfs_s3select_api::parquet_budget::ParquetBudgetError::Exceeded => budget_error(),
+        error => TableCatalogStoreError::Invalid(format!("compaction {error}")),
     }
 }
 
-/// Reject footer collection counts the decoder would allocate for before it
-/// reads a single element. Bytes after the struct are ignored, as parquet does.
 fn validate_footer_metadata(data: &Bytes, limit: usize) -> TableCatalogStoreResult<()> {
-    let error = || TableCatalogStoreError::Invalid("compaction parquet footer is malformed".to_owned());
-    let tail_start = data.len().checked_sub(FOOTER_SIZE).ok_or_else(error)?;
-    let tail: &[u8; FOOTER_SIZE] = data[tail_start..].try_into().map_err(|_| error())?;
-    let tail = FooterTail::try_new(tail).map_err(|_| error())?;
-    let metadata_start = tail_start.checked_sub(tail.metadata_length()).ok_or_else(error)?;
-    let mut cursor = ThriftCompactCursor {
-        bytes: &data[metadata_start..tail_start],
-        pos: 0,
-    };
-    cursor.file_metadata(limit)
+    rustfs_s3select_api::parquet_budget::validate_file_footer(data, limit).map_err(parquet_admission_error)
 }
 
-/// Sum the page headers of one column chunk (`[start, end)` in the file) and
-/// hold every page and the chunk total to `limit`. Pages are never decompressed.
-fn validate_page_headers_in_range(chunk: &[u8], chunk_declared: usize, limit: usize) -> TableCatalogStoreResult<usize> {
-    let mut cursor = ThriftCompactCursor { bytes: chunk, pos: 0 };
-    let mut total = 0_usize;
-    while cursor.pos < chunk.len() {
-        let sizes = cursor.page_header()?;
-        add_budget(&mut total, sizes.uncompressed, limit)?;
-        if total > chunk_declared.max(limit.min(chunk_declared.saturating_mul(2))) {
-            return Err(TableCatalogStoreError::Invalid(
-                "compaction parquet page headers exceed the footer's declared chunk size".to_owned(),
-            ));
-        }
-        cursor.skip_bytes(sizes.compressed)?;
-    }
-    Ok(total)
+#[cfg(test)]
+fn validate_page_headers_in_range(chunk: &[u8], declared: usize, limit: usize) -> TableCatalogStoreResult<usize> {
+    rustfs_s3select_api::parquet_budget::validate_page_headers(chunk, declared, limit).map_err(parquet_admission_error)
 }
 
 fn validate_column_chunk_page_headers(
     data: &Bytes,
     column: &ColumnChunkMetaData,
-    chunk_declared: usize,
+    declared: usize,
     limit: usize,
     location: &str,
-) -> TableCatalogStoreResult<()> {
-    // parquet's `byte_range`: the chunk starts at the dictionary page when the
-    // footer declares one, wherever that is. Walk exactly the bytes it decodes.
-    let start = column.dictionary_page_offset().unwrap_or(column.data_page_offset());
-    let start = usize::try_from(start).map_err(|_| budget_error())?;
-    let length = usize::try_from(column.compressed_size()).map_err(|_| budget_error())?;
-    let end = start.checked_add(length).filter(|end| *end <= data.len()).ok_or_else(|| {
-        TableCatalogStoreError::Invalid(format!("compaction parquet {location} column chunk lies outside the file"))
-    })?;
-    validate_page_headers_in_range(&data[start..end], chunk_declared, limit).map_err(|err| match err {
-        TableCatalogStoreError::Invalid(message) => TableCatalogStoreError::Invalid(format!("{location}: {message}")),
-        other => other,
-    })?;
-    Ok(())
+) -> TableCatalogStoreResult<usize> {
+    rustfs_s3select_api::parquet_budget::validate_column_chunk_page_headers(data, column, declared, limit).map_err(|error| {
+        match parquet_admission_error(error) {
+            TableCatalogStoreError::Invalid(message) => TableCatalogStoreError::Invalid(format!("{location}: {message}")),
+            error => error,
+        }
+    })
 }
 
 fn view_field(field: &Field, depth: usize, fields: &mut usize) -> TableCatalogStoreResult<Arc<Field>> {
@@ -542,11 +316,11 @@ fn compact_parquet_with_budgets(
             let mut columns = 0;
             for column in group.columns() {
                 let chunk_declared = usize::try_from(column.uncompressed_size()).map_err(|_| budget_error())?;
-                add_budget(&mut columns, chunk_declared, limit)?;
                 // The footer is one claim; the page headers the decoder actually
-                // sizes its buffers from are another. Walk them without
-                // decompressing and hold both to the budget.
-                validate_column_chunk_page_headers(&data, column, chunk_declared, limit, &location)?;
+                // sizes its buffers from are another. Check the page headers
+                // and codec expansion before admitting the decoder.
+                let resident = validate_column_chunk_page_headers(&data, column, chunk_declared, limit, &location)?;
+                add_budget(&mut columns, chunk_declared.max(resident), limit)?;
             }
             let group_bytes = usize::try_from(group.total_byte_size()).map_err(|_| budget_error())?;
             add_budget(&mut declared, group_bytes.max(columns), limit)?;
@@ -850,9 +624,9 @@ mod tests {
     #[test]
     fn compaction_page_header_walker_bounds_collection_counts_by_remaining_bytes() {
         let limit = 64 * 1024 * 1024;
-        // Unknown field 4: LIST<BOOL> with long-form count 2^63. Parquet's skip
+        // Unknown field 10: LIST<BOOL> with long-form count 2^63. Parquet's skip
         // would spin; the walker must reject it before iterating.
-        let mut list = vec![(1 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_BOOL_TRUE];
+        let mut list = vec![(7 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_BOOL_TRUE];
         list.extend(varint(1 << 63));
         let started = std::time::Instant::now();
         let error = validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit)
@@ -864,15 +638,15 @@ mod tests {
         assert!(started.elapsed() < std::time::Duration::from_secs(1));
         // A bool collection of any size is rejected: parquet consumes no bytes
         // per element, so either width would desynchronize the two parsers.
-        let list = [(1 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_BOOL_TRUE, 1, 1];
+        let list = [(7 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_BOOL_TRUE, 1, 1];
         assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit).is_err());
         // MAP<BOOL, BOOL> with a huge count is rejected the same way.
-        let mut map = vec![(1 << 4) | THRIFT_COMPACT_MAP];
+        let mut map = vec![(7 << 4) | THRIFT_COMPACT_MAP];
         map.extend(varint(1 << 62));
         map.push((THRIFT_COMPACT_BOOL_TRUE << 4) | THRIFT_COMPACT_BOOL_TRUE);
         assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &map), 4096, limit).is_err());
         // Collections whose elements carry bytes still skip when the count fits.
-        let mut list = vec![(1 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_I32];
+        let mut list = vec![(7 << 4) | THRIFT_COMPACT_LIST, (2 << 4) | THRIFT_COMPACT_I32];
         list.extend(compact_i32_field(0, 7)[1..].iter());
         list.extend(compact_i32_field(0, -7)[1..].iter());
         assert_eq!(
@@ -880,7 +654,7 @@ mod tests {
             8
         );
         // A LIST<I32> whose count exceeds the chunk is malformed, not truncated.
-        let list = [(1 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_I32, 0x7f];
+        let list = [(7 << 4) | THRIFT_COMPACT_LIST, 0xf0 | THRIFT_COMPACT_I32, 0x7f];
         assert!(validate_page_headers_in_range(&page_chunk(compact_i32_field(1, 8), &list), 4096, limit).is_err());
     }
 
@@ -956,6 +730,35 @@ mod tests {
         }
     }
 
+    #[test]
+    fn compaction_aggregates_column_decode_working_sets() {
+        let columns =
+            (0..4).map(|index| (format!("value_{index}"), Arc::new(StringArray::from(vec!["v"; 1000])) as Arc<dyn Array>));
+        let batch = RecordBatch::try_from_iter(columns).expect("multi-column batch");
+        let properties = WriterProperties::builder()
+            .set_compression(Compression::ZSTD(ZstdLevel::default()))
+            .build();
+        let mut writer = ArrowWriter::try_new(Vec::new(), batch.schema(), Some(properties)).expect("Parquet writer");
+        writer.write(&batch).expect("multi-column data");
+        let data = Bytes::from(writer.into_inner().expect("Parquet footer"));
+        let limit = 100 * 1024;
+        let reader = ParquetRecordBatchReaderBuilder::try_new(data.clone()).expect("valid fixture metadata");
+        let mut working_set = 0;
+        for column in reader.metadata().row_group(0).columns() {
+            let declared = usize::try_from(column.uncompressed_size()).expect("positive size");
+            working_set += validate_column_chunk_page_headers(&data, column, declared, limit, "columns.parquet")
+                .expect("each column fits individually");
+        }
+        assert!(working_set > limit, "the aggregate exceeds the decode budget");
+        let error = compact_parquet_with_limits(vec![("columns.parquet".to_owned(), data.to_vec())], limit, limit)
+            .err()
+            .expect("aggregate column allocations must be admitted before decoding");
+        assert!(
+            matches!(&error, TableCatalogStoreError::Invalid(message) if message.contains("budget")),
+            "{error:?}"
+        );
+    }
+
     #[test]
     fn compaction_rejects_compressed_expansion_before_decoding_pages() {
         let file = parquet_strings(&"x".repeat(64 * 1024), 32, false);

scripts/test_download_release_asset.py

@@ -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()