test(e2e): fold eleven post-object accept cases into one table-driven test

2026-08-18 10:43:15 +00:00 · 2026-08-18 15:03:07 +08:00
53 changed files with 1997 additions and 2970 deletions
@@ -15,35 +15,28 @@
 # Package Workflow - Build DEB/RPM packages
 #
 # This workflow builds DEB and RPM packages from pre-built Linux binaries
-# and uploads them to Cloudflare R2 and the GitHub release.
+# and uploads them to Cloudflare R2.
 #
 # Trigger:
-# - workflow_run: automatically package after "Build and Release" completes
-#   for a release tag (the mac/windows/linux binaries are already uploaded
-#   to the GitHub release before packaging starts)
-# - workflow_dispatch: manual fallback (backfill / re-run) with optional tag/run_id
+# - release published: automatically package when a GitHub release is published
+# - workflow_dispatch: manual trigger with optional tag/run_id
 #
 # Flow:
-# 1. Resolve the triggering Build workflow run for the release tag
+# 1. Find the Build workflow run for the release tag
 # 2. Download Linux binaries (x86_64-gnu, aarch64-gnu) from build artifacts
 # 3. Build DEB packages for amd64 and arm64
 # 4. Build RPM packages for x86_64 and aarch64
-# 5. Upload all packages to Cloudflare R2 and the GitHub release
+# 5. Upload all packages to Cloudflare R2

 name: Package DEB/RPM

 permissions:
-  # contents: write is required to upload packages to the GitHub release
-  contents: write
+  contents: read
  actions: read

 on:
-  # Follows the same pattern as docker.yml: run after the release build
-  # workflow completes, so packaging is triggered only by release tags
-  # (e.g. 1.0.0-rc.2, 1.0.0-rc.3), never by development builds.
-  workflow_run:
-    workflows: [ "Build and Release" ]
-    types: [ completed ]
+  release:
+    types: [ published ]
  workflow_dispatch:
    inputs:
      tag:
@@ -56,26 +49,13 @@ on:
        type: string

 concurrency:
-  group: ${{ github.workflow }}-${{ github.event.workflow_run.head_branch || github.event.inputs.tag || github.run_id }}
+  group: ${{ github.workflow }}-${{ github.event.release.tag_name || github.event.inputs.tag || github.run_id }}
  cancel-in-progress: true

-env:
-  HEAD_BRANCH: ${{ github.event.workflow_run.head_branch }}
-  WORKFLOW_RUN_ID: ${{ github.event.workflow_run.id }}
-
 jobs:
  # Resolve which build run to use and extract version info
  resolve:
    name: Resolve Build
-    # Auto-trigger only from successful tag builds of "Build and Release".
-    # Tag pushes arrive as event == push with head_branch != main (a
-    # non-main push head_branch is the release tag name). Manual dispatch
-    # stays available as a fallback for backfills and re-runs.
-    if: >-
-      github.event_name == 'workflow_dispatch' ||
-      (github.event.workflow_run.conclusion == 'success' &&
-       github.event.workflow_run.event == 'push' &&
-       github.event.workflow_run.head_branch != 'main')
    runs-on: ubuntu-latest
    timeout-minutes: 10
    outputs:
@@ -95,8 +75,8 @@ jobs:
          set -euo pipefail

          # Determine tag
-          if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
-            TAG="${HEAD_BRANCH}"
+          if [[ "${{ github.event_name }}" == "release" ]]; then
+            TAG="${{ github.event.release.tag_name }}"
          elif [[ -n "$INPUT_TAG" ]]; then
            TAG="$INPUT_TAG"
          else
@@ -113,11 +93,6 @@ jobs:
            BUILD_RUN_ID="$INPUT_RUN_ID"
            echo "Using explicit build run ID: $BUILD_RUN_ID"

-          elif [[ "${{ github.event_name }}" == "workflow_run" ]]; then
-            # Use the Build and Release run that triggered this workflow
-            BUILD_RUN_ID="${WORKFLOW_RUN_ID}"
-            echo "Using triggering workflow run: $BUILD_RUN_ID"
-
          elif [[ -n "$TAG" ]]; then
            # Find the build run that produced this tag
            echo "Looking for build run for tag: $TAG"
@@ -481,54 +456,6 @@ jobs:
            echo "✅ Latest packages updated"
          fi

-      - name: Upload packages to GitHub Release
-        if: needs.resolve.outputs.tag != ''
-        env:
-          GH_TOKEN: ${{ github.token }}
-        shell: bash
-        run: |
-          set -euo pipefail
-
-          TAG="${{ needs.resolve.outputs.tag }}"
-          DEB_FILE="${{ steps.deb.outputs.deb_file }}"
-          RPM_FILE="${{ steps.rpm.outputs.rpm_file }}"
-
-          # Upload the packages, then refresh the release checksums so the new
-          # assets are covered, matching the binary release flow.
-          for f in "$DEB_FILE" "$RPM_FILE"; do
-            if [[ -n "$f" && -f "$f" ]]; then
-              echo "📤 Uploading $(basename "$f") to GitHub release ${TAG}..."
-              gh release upload "$TAG" "$f" --clobber
-            fi
-          done
-
-          CHECKSUM_DIR="$(mktemp -d)"
-          gh release download "$TAG" -p 'SHA256SUMS' -p 'SHA512SUMS' \
-            -D "$CHECKSUM_DIR" --clobber 2>/dev/null || true
-
-          for spec in "SHA256SUMS:sha256sum" "SHA512SUMS:sha512sum"; do
-            asset="${spec%%:*}"
-            checksum_cmd="${spec##*:}"
-            checksum_file="${CHECKSUM_DIR}/${asset}"
-
-            touch "$checksum_file"
-
-            for f in "$DEB_FILE" "$RPM_FILE"; do
-              if [[ -n "$f" && -f "$f" ]]; then
-                base="$(basename "$f")"
-                # Remove any stale entry, then append the fresh digest
-                grep -Fv -- "$base" "$checksum_file" > "${checksum_file}.tmp" || true
-                mv "${checksum_file}.tmp" "$checksum_file"
-                (cd "$(dirname "$f")" && "$checksum_cmd" -- "$base") >> "$checksum_file"
-              fi
-            done
-
-            echo "📤 Updating ${asset} for release ${TAG}..."
-            gh release upload "$TAG" "$checksum_file" --clobber
-          done
-
-          echo "✅ GitHub release assets updated"
-
  # Summary
  summary:
    name: Summary
@@ -964,9 +964,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-kms"
-version = "1.115.0"
+version = "1.114.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d5b034f8b7ceadb873d0bc607c30bb4b0be68e09a84c837174e7c2c6878ff882"
+checksum = "c0b7d906608ee41e7ddea9983577ba82200435644d567d63dc34e822e088b453"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -990,9 +990,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.142.0"
+version = "1.141.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9e15a5c55e05f4b0b7e483160b3c85cccdf77cff02c95504f3e71d460855cd2"
+checksum = "d9f9420d3a2467eed22ed3635ca653653162c386a0b0f65c78189f9bd3c1379e"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1027,9 +1027,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sso"
-version = "1.106.0"
+version = "1.105.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2d0efcee834347b6705eca3eea2defd88242f43774f55d7326604222e3c86260"
+checksum = "6ffd0fbe7873cb548a7aa60f9573c268fff94155397fd4f14dc9f1ecaaab8516"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1053,9 +1053,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.108.0"
+version = "1.107.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a59312a04cf19c962cfee32b64ecfee758f8786407ff6da5b30fff46ae96f201"
+checksum = "175763eb222a46377df7aa257a3bca980ab3e96703fefc8f4d0b8da6ad2e254c"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1079,9 +1079,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sts"
-version = "1.111.0"
+version = "1.110.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "120e7eb63457a9e547f9986fe3b273f77c43679da4d04f46359fa881c5e19b6e"
+checksum = "dd8b14781dfbff48984017d57167b6ea0b6471c6920ec52b44a2677c7feb3c13"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1598,7 +1598,7 @@ version = "0.10.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3078c7629b62d3f0439517fa394996acacc5cbc91c5a20d8c658e77abd503a71"
 dependencies = [
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 ]

 [[package]]
@@ -1617,7 +1617,7 @@ version = "0.3.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a8894febbff9f758034a5b8e12d87918f56dfc64a8e1fe757d65e29041538d93"
 dependencies = [
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 ]

 [[package]]
@@ -1858,9 +1858,9 @@ dependencies = [

 [[package]]
 name = "cc"
-version = "1.4.3"
+version = "1.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "509591b7bcd67f4ef775afad7662703b4935daaa6ec0e5605cfb1090b32a2b6d"
+checksum = "5d262e149917187838d5b42777c8253bcb64500067342904e7d429499a6f277e"
 dependencies = [
 "find-msvc-tools",
 "jobserver",
@@ -1968,7 +1968,7 @@ version = "0.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773f3b9af64447d2ce9850330c473515014aa235e6a783b02db81ff39e4a3dad"
 dependencies = [
- "crypto-common 0.1.7",
+ "crypto-common 0.1.6",
 "inout 0.1.4",
 ]

@@ -2428,7 +2428,7 @@ version = "0.5.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0dc92fb57ca44df6db8059111ab3af99a63d5d0f8375d9972e319a379c6bab76"
 dependencies = [
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 "rand_core 0.6.4",
 "subtle",
 "zeroize",
@@ -2453,11 +2453,11 @@ dependencies = [

 [[package]]
 name = "crypto-common"
-version = "0.1.7"
+version = "0.1.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "78c8292055d1c1df0cce5d180393dc8cce0abec0a7102adb6c7b1eef6016d60a"
+checksum = "1bfb12502f3fc46cca1bb51ac28df9d618d813cdc3d2f25b9fe775a34af26bb3"
 dependencies = [
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 "typenum",
 ]

@@ -3664,7 +3664,7 @@ checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
 "block-buffer 0.10.4",
 "const-oid 0.9.6",
- "crypto-common 0.1.7",
+ "crypto-common 0.1.6",
 "subtle",
 ]

@@ -3924,7 +3924,7 @@ dependencies = [
 "crypto-bigint 0.5.5",
 "digest 0.10.7",
 "ff 0.13.1",
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 "group 0.13.0",
 "hkdf 0.12.4",
 "pem-rfc7468 0.7.0",
@@ -4148,9 +4148,9 @@ dependencies = [

 [[package]]
 name = "find-msvc-tools"
-version = "0.1.11"
+version = "0.1.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d45db016d36b838f563236e9193d0ee6ce38f3f68b6c94e914b4929c96bbb890"
+checksum = "26b73573e6edcd2af0cdf47bd6cb58f0b3839491263c314eaad1ccf24430e1de"

 [[package]]
 name = "findshlibs"
@@ -4369,9 +4369,9 @@ dependencies = [

 [[package]]
 name = "generic-array"
-version = "0.14.7"
+version = "0.14.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "85649ca51fd72272d7821adaf274ad91c288277713d9c18820d8499a7ff69e9a"
+checksum = "4bb6743198531e02858aeaea5398fcc883e71851fcbcb5a2f773e2fb6cb1edf2"
 dependencies = [
 "typenum",
 "version_check",
@@ -4380,11 +4380,11 @@ dependencies = [

 [[package]]
 name = "generic-array"
-version = "1.4.5"
+version = "1.4.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "337d46834ee672ab3e48caca2cb0c78cc174fb12b3a68d0d88f99a0519a5e36e"
+checksum = "ab4e5aa225bc56696909483320f0ff9b600f1a971b52e07a17d70f3d9b43254b"
 dependencies = [
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 "rustversion",
 "typenum",
 ]
@@ -4726,9 +4726,9 @@ dependencies = [

 [[package]]
 name = "h2"
-version = "0.4.16"
+version = "0.4.15"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9f37a958b41b3b19ee2707c06439c0e9e547e847223eb791ecb0cb821c65e27"
+checksum = "6cb093c84e8bd9b188d4c4a8cb6579fc016968d14c99882163cd3ff402a4f155"
 dependencies = [
 "atomic-waker",
 "bytes",
@@ -5028,9 +5028,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.23.3"
+version = "0.23.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dce755d457a63bdd0c95e4c91511daad1b58b33209543b7f38027b676f387e5e"
+checksum = "62e810bedda5a467ef5c9b5c8a20763fefebc89b63ef36f7ee44a143085204a2"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5062,9 +5062,9 @@ dependencies = [

 [[package]]
 name = "hotpath-macros"
-version = "0.23.3"
+version = "0.23.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a903af89a8429cb07790c3818bc15270b394f80af1bc254e5ccf9c7de2961770"
+checksum = "01bdc59bfc1a9984bee2ff5da63b2f6fccbaa57cd9a4119d709524632bddf341"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5073,15 +5073,15 @@ dependencies = [

 [[package]]
 name = "hotpath-macros-meta"
-version = "0.23.3"
+version = "0.23.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bcc0ab94ffbb2ee77f4a897df02b5a137a10cf24d69bda936e59aff4dd456e61"
+checksum = "d9216e8a01abe1e1671c376dc8736fb1bf772d7a889538d25f9e1200120ced38"

 [[package]]
 name = "hotpath-meta"
-version = "0.23.3"
+version = "0.23.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "053481f6cec8f775a3276c7f6e2f21123111d28261e4edc15ea7421c445964bb"
+checksum = "f22a9d20435fb79511b19dae37b3607224cd98f342a410702d84657cc38fc72f"
 dependencies = [
 "hotpath-macros-meta",
 ]
@@ -5280,9 +5280,9 @@ dependencies = [

 [[package]]
 name = "icu_collections"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fa68d21081c4a05d5a901a1c62add574c77048b6a1c67be3b50ce0b60d4ca513"
+checksum = "2984d1cd16c883d7935b9e07e44071dca8d917fd52ecc02c04d5fa0b5a3f191c"
 dependencies = [
 "displaydoc",
 "potential_utf",
@@ -5294,9 +5294,9 @@ dependencies = [

 [[package]]
 name = "icu_locale_core"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d56e28588da92eee5c3201a6eff33fabdd49b62269c8938d4ff050ce4d900deb"
+checksum = "92219b62b3e2b4d88ac5119f8904c10f8f61bf7e95b640d25ba3075e6cac2c29"
 dependencies = [
 "displaydoc",
 "litemap",
@@ -5307,9 +5307,9 @@ dependencies = [

 [[package]]
 name = "icu_normalizer"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "12f9cf5f235641ed274641dd81c3f28d870e276763d0797aeeab72317b1c646f"
+checksum = "c56e5ee99d6e3d33bd91c5d85458b6005a22140021cc324cea84dd0e72cff3b4"
 dependencies = [
 "icu_collections",
 "icu_normalizer_data",
@@ -5321,17 +5321,16 @@ dependencies = [

 [[package]]
 name = "icu_normalizer_data"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1563da1ed3e0b3bf3d74c9b85917ac9c56464d2f57242270c09c9e752f8021a0"
+checksum = "da3be0ae77ea334f4da67c12f149704f19f81d1adf7c51cf482943e84a2bad38"

 [[package]]
 name = "icu_properties"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e7ca276ad3145661a65914e6daf131ca5120cd3dcee8f8f3214b8875184a148"
+checksum = "bee3b67d0ea5c2cca5003417989af8996f8604e34fb9ddf96208a033901e70de"
 dependencies = [
- "displaydoc",
 "icu_collections",
 "icu_locale_core",
 "icu_properties_data",
@@ -5342,15 +5341,15 @@ dependencies = [

 [[package]]
 name = "icu_properties_data"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e590f038c1464a96894fd6d10127e90a8be4509f56ff7ecef851b15cee0b7caa"
+checksum = "8e2bbb201e0c04f7b4b3e14382af113e17ba4f63e2c9d2ee626b720cbce54a14"

 [[package]]
 name = "icu_provider"
-version = "2.3.0"
+version = "2.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "92a7ed671a6aad807a8651a2e1782a6598fda9ce5185dd8158549e95a91c6428"
+checksum = "139c4cf31c8b5f33d7e199446eff9c1e02decfc2f0eec2c8d71f65befa45b421"
 dependencies = [
 "displaydoc",
 "icu_locale_core",
@@ -5418,7 +5417,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "879f10e63c20629ecabbb64a8010319738c66a5cd0c29b02d63d272b03751d01"
 dependencies = [
 "block-padding 0.3.3",
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 ]

 [[package]]
@@ -5969,9 +5968,9 @@ dependencies = [

 [[package]]
 name = "libredox"
-version = "0.1.20"
+version = "0.1.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "28d0a00925a9f930d679b6789b721e3a7f9ed110f41b86d2497caa780c3a070a"
+checksum = "2026a5056764a10b2bf5d56488cba40da507f5493a6a429340e2004d9ed085fa"
 dependencies = [
 "libc",
 ]
@@ -6034,9 +6033,9 @@ checksum = "32a66949e030da00e8c7d4434b251670a91556f4144941d37452769c25d58a53"

 [[package]]
 name = "litemap"
-version = "0.8.3"
+version = "0.8.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "47d9d19d1d6efa0109d2f65ff4c85cddd50bd572e5a00127ab10987290bcefae"
+checksum = "92daf443525c4cce67b150400bc2316076100ce0b3686209eb8cf3c31612e6f0"

 [[package]]
 name = "local-ip-address"
@@ -6483,9 +6482,9 @@ dependencies = [

 [[package]]
 name = "mqttbytes-core-next"
-version = "0.34.0"
+version = "0.33.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "366b6ba2b4209ca4bc5ac731ccddf570d09831981eed07e5fbd63564cf0cf1aa"
+checksum = "3ff7ae19c74aba9e0ed6e4071cd52aa364e020076fa3cc6ef17e43662f756f3c"
 dependencies = [
 "bytes",
 "thiserror 2.0.20",
@@ -6886,7 +6885,7 @@ version = "5.0.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
 dependencies = [
- "base64 0.22.1",
+ "base64 0.21.7",
 "chrono",
 "getrandom 0.2.17",
 "http 1.5.0",
@@ -7345,9 +7344,9 @@ dependencies = [

 [[package]]
 name = "pageant"
-version = "0.2.2"
+version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3adadc44070da6f464b0918655a12f5792c156e088d8c4082d13e27d94c3e791"
+checksum = "4f3a5ae18f65a85c67a77d18d42d3606c07948e3c17c1e5f74852b26589e88a5"
 dependencies = [
 "base16ct 1.0.0",
 "byteorder",
@@ -7729,9 +7728,9 @@ dependencies = [

 [[package]]
 name = "pkg-config"
-version = "0.3.34"
+version = "0.3.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f6b464fbc74e149a392436b17d523f769e057cb6877f6a5c4618bc6f11800548"
+checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"

 [[package]]
 name = "plotters"
@@ -7837,9 +7836,9 @@ dependencies = [

 [[package]]
 name = "potential_utf"
-version = "0.1.6"
+version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d83eb9bc6d8e5cf568e7a1101d60ee05e81ed50ea106026f3d18deeb046d7661"
+checksum = "0103b1cef7ec0cf76490e969665504990193874ea05c85ff9bab8b911d0a0564"
 dependencies = [
 "zerovec",
 ]
@@ -8047,7 +8046,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
 "heck 0.5.0",
- "itertools 0.14.0",
+ "itertools 0.10.5",
 "log",
 "multimap",
 "once_cell",
@@ -8067,7 +8066,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "03da047801ff44bb6a4d407d4860c05fd70bb81714e6b2f3812603d5b145b042"
 dependencies = [
 "heck 0.5.0",
- "itertools 0.14.0",
+ "itertools 0.10.5",
 "log",
 "multimap",
 "petgraph 0.8.3",
@@ -8088,7 +8087,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
 "anyhow",
- "itertools 0.14.0",
+ "itertools 0.10.5",
 "proc-macro2",
 "quote",
 "syn 2.0.119",
@@ -8101,7 +8100,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b570b25f7617e43d59005d0990ccb79e950a423952cea19671b7a876da390adf"
 dependencies = [
 "anyhow",
- "itertools 0.14.0",
+ "itertools 0.10.5",
 "proc-macro2",
 "quote",
 "syn 2.0.119",
@@ -8217,7 +8216,7 @@ dependencies = [
 "reqwest",
 "serde_json",
 "smallvec",
- "spin 0.12.3",
+ "spin 0.12.2",
 "symbolic-demangle",
 "tempfile",
 "thiserror 2.0.20",
@@ -8286,9 +8285,9 @@ dependencies = [

 [[package]]
 name = "quinn-proto"
-version = "0.11.17"
+version = "0.11.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "04759210543be93709136e28212294a659ef5001836ff4eab4d663e4529bba83"
+checksum = "2f4bfc015262b9df63c8845072ce59068853ff5872180c2ce2f13038b970e560"
 dependencies = [
 "aws-lc-rs",
 "bytes",
@@ -8554,9 +8553,9 @@ dependencies = [

 [[package]]
 name = "redis"
-version = "1.6.0"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e37a4ca5c6ca42aa3e6df2fd32b987a65d32a4c2159a6f3fe0fd1df306a2658f"
+checksum = "3257df217f7eab0044627a268c9cc6cdb60c0c421c88f83ac41c4e31520b6b84"
 dependencies = [
 "arc-swap",
 "arcstr",
@@ -8568,7 +8567,7 @@ dependencies = [
 "futures-channel",
 "futures-util",
 "itoa",
- "num-bigint 0.5.1",
+ "num-bigint 0.4.8",
 "percent-encoding",
 "pin-project-lite",
 "rustls",
@@ -8869,9 +8868,9 @@ dependencies = [

 [[package]]
 name = "rumqttc-core-next"
-version = "0.34.0"
+version = "0.33.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "249896ab27ed630590971738264baa8f722f18965d2e387c706c40a3c2a572cc"
+checksum = "7d7d9205738dd41a2546e82d27a634d07d8b303dcf7558565ff70caf3ceb0f9c"
 dependencies = [
 "async-tungstenite",
 "futures-io",
@@ -8887,18 +8886,18 @@ dependencies = [

 [[package]]
 name = "rumqttc-next"
-version = "0.34.0"
+version = "0.33.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "477c9bbfba8f3aecc7aad31c6de2eacb75822efaa18e7aeecb8d3d8e534fbf07"
+checksum = "ed1bad2180ff539da671da9a996152a921bc5316eb6d8a9cc3bd441653138b08"
 dependencies = [
 "rumqttc-v5-next",
 ]

 [[package]]
 name = "rumqttc-v5-next"
-version = "0.34.0"
+version = "0.33.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3dfa6ddcc7a7dd5688f9bf78d8f81cb94f367bce56c055d8d94cf81ecb0518bf"
+checksum = "229576cbedfa9089f90c17c9454e9429ac1e89cdd223bac5cb39d837593f79bc"
 dependencies = [
 "async-tungstenite",
 "bytes",
@@ -8921,9 +8920,9 @@ dependencies = [

 [[package]]
 name = "russh"
-version = "0.62.7"
+version = "0.62.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9decb68e4e44e1079700e54f17c8f23806ec53d7e0db73ab1c71d9dabc666812"
+checksum = "b41043523e0edcbd4e31d00903e26f12994f63b21bae9904f7405c1ed92752a5"
 dependencies = [
 "aes 0.9.2",
 "aws-lc-rs",
@@ -8946,7 +8945,7 @@ dependencies = [
 "enum_dispatch",
 "flate2",
 "futures",
- "generic-array 1.4.5",
+ "generic-array 1.4.4",
 "getrandom 0.4.3",
 "ghash",
 "hex-literal",
@@ -9492,7 +9491,7 @@ dependencies = [
 "parking_lot",
 "rayon",
 "smallvec",
- "spin 0.12.3",
+ "spin 0.12.2",
 ]

 [[package]]
@@ -10834,7 +10833,7 @@ checksum = "d3e97a565f76233a6003f9f5c54be1d9c5bdfa3eccfb189469f11ec4901c47dc"
 dependencies = [
 "base16ct 0.2.0",
 "der 0.7.10",
- "generic-array 0.14.7",
+ "generic-array 0.14.9",
 "pkcs8 0.10.2",
 "subtle",
 "zeroize",
@@ -11398,9 +11397,9 @@ checksum = "023a211cb3138dbc438680b32560ad89f699977624c9f8dbb95a47d5b4c07dd3"

 [[package]]
 name = "spin"
-version = "0.12.3"
+version = "0.12.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0134f9043ed38b087ac4f7d4af44c79e2c9e5094421fe3164f435ce585953b10"
+checksum = "8abadc99fd9c7bbb7d0ca2b31d72a067d0c0dcd7aad25ab8cac71ba91417694b"
 dependencies = [
 "lock_api",
 ]
@@ -11812,7 +11811,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
 "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
 "once_cell",
 "rustix",
 "windows-sys 0.61.2",
@@ -11976,9 +11975,9 @@ dependencies = [

 [[package]]
 name = "tinystr"
-version = "0.8.4"
+version = "0.8.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b1e27c91459209c2986af3dcf603a5a74a4368754ce37414f59acc971167f643"
+checksum = "c8323304221c2a851516f22236c5722a72eaa19749016521d6dff0824447d96d"
 dependencies = [
 "displaydoc",
 "zerovec",
@@ -12652,9 +12651,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"

 [[package]]
 name = "uuid"
-version = "1.24.1"
+version = "1.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2cefc03fd367c0c6d4305de1b312cf00248c4114f4a0418ce6a6af769e3b0bd9"
+checksum = "bf3923a6f5c4c6382e0b653c4117f48d631ea17f38ed86e2a828e6f7412f5239"
 dependencies = [
 "getrandom 0.4.3",
 "js-sys",
@@ -13169,9 +13168,9 @@ dependencies = [

 [[package]]
 name = "writeable"
-version = "0.6.4"
+version = "0.6.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3ad82d2a33cdc9674dc7465672f271e096168fcdbe0f799d9e6db8c5892679dc"
+checksum = "1ffae5123b2d3fc086436f8834ae3ab053a283cfac8fe0a0b8eaae044768a4c4"

 [[package]]
 name = "x509-cert"
@@ -13351,9 +13350,9 @@ dependencies = [

 [[package]]
 name = "zerotrie"
-version = "0.2.5"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4ea269c3bd32f0a32c321907a2ae912ba6f4649bb0fc764a15627e99a7095a3f"
+checksum = "0f9152d31db0792fa83f70fb2f83148effb5c1f5b8c7686c3459e361d9bc20bf"
 dependencies = [
 "displaydoc",
 "yoke",
@@ -13362,9 +13361,9 @@ dependencies = [

 [[package]]
 name = "zerovec"
-version = "0.11.7"
+version = "0.11.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "94b5c6b5976d66c1d703c4fd17d3f5e43c8cedaacf604961b171adc7130896d8"
+checksum = "90f911cbc359ab6af17377d242225f4d75119aec87ea711a880987b18cd7b239"
 dependencies = [
 "yoke",
 "zerofrom",
@@ -13373,13 +13372,13 @@ dependencies = [

 [[package]]
 name = "zerovec-derive"
-version = "0.11.5"
+version = "0.11.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9f212a141d820099d57ffafb9569be9617a6f27d3dc881fbee8fb56642f917a9"
+checksum = "625dc425cab0dca6dc3c3319506e6593dcb08a9f387ea3b284dbd52a92c40555"
 dependencies = [
 "proc-macro2",
 "quote",
- "syn 3.0.3",
+ "syn 2.0.119",
 ]

 [[package]]
@@ -228,9 +228,9 @@ atoi = "3.1.0"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.10.1" }
 aws-credential-types = { version = "1.3.0" }
-aws-sdk-kms = { default-features = false, version = "1.115.0" }
-aws-sdk-s3 = { default-features = false, version = "1.142.0" }
-aws-sdk-sts = { default-features = false, version = "1.111.0" }
+aws-sdk-kms = { default-features = false, version = "1.114.0" }
+aws-sdk-s3 = { default-features = false, version = "1.141.0" }
+aws-sdk-sts = { default-features = false, version = "1.110.0" }
 aws-smithy-http-client = { default-features = false, version = "1.3.0" }
 aws-smithy-runtime-api = { version = "1.14.0" }
 aws-smithy-types = { version = "1.6.2" }
@@ -284,8 +284,8 @@ rayon = "1.12.0"
 reed-solomon-erasure = { package = "rustfs-erasure-codec", version = "8.0.2" }
 reed-solomon-simd = "3.1.0"
 regex = { version = "1.13.1" }
-rumqttc = { package = "rumqttc-next", version = "0.34.0" }
-redis = { version = "1.6.0" }
+rumqttc = { package = "rumqttc-next", version = "0.33.3" }
+redis = { version = "1.5.0" }
 rustify = { version = "0.7", default-features = false }
 rustix = { version = "1.1.4" }
 rust-embed = { version = "8.12.0" }
@@ -313,7 +313,7 @@ tracing-subscriber = { version = "0.3.23" }
 transform-stream = "0.3.1"
 url = "2.5.8"
 urlencoding = "2.1.3"
-uuid = { version = "1.24.1" }
+uuid = { version = "1.24.0" }
 vaultrs = { version = "0.8.0" }
 tar = "0.4.46"
 walkdir = "2.5.0"
@@ -341,7 +341,7 @@ libunftp = { version = "0.23.0" }
 unftp-core = "0.1.0"
 suppaftp = { version = "10.0.1" }
 rcgen = { version = "0.14.9", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
-russh = { version = "0.62.7" }
+russh = { version = "0.62.6" }
 russh-sftp = "2.4.0"

 # WebDAV
@@ -350,7 +350,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 mimalloc = { version = "0.1.52", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11" }
 libmimalloc-sys = { version = "0.1.49", git = "https://github.com/xonatius/mimalloc_rust.git", rev = "6d4c41bb10c6d9da1d1b6f07b38c4cc051667f11", features = ["extended"] }
-hotpath = { version = "0.23.3", default-features = false }
+hotpath = { version = "0.23.2", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

@@ -287,9 +287,6 @@ pub enum HealRequestSource {
    Scanner,
    AutoHeal,
    ReadRepair,
-    /// Mission Repair Feed: intents delivered by error paths and replayed
-    /// from the durable MRF journal.
-    Mrf,
 }

 impl HealRequestSource {
@@ -300,7 +297,6 @@ impl HealRequestSource {
            Self::Scanner => "scanner",
            Self::AutoHeal => "auto_heal",
            Self::ReadRepair => "read_repair",
-            Self::Mrf => "mrf",
        }
    }
 }
@@ -17,7 +17,6 @@ pub mod globals;
 pub mod heal_channel;
 pub mod last_minute;
 pub mod metrics;
-pub mod mrf_channel;
 mod readiness;
 pub mod table_catalog;
 pub mod trace_bus;
@@ -1,203 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Mission Repair Feed (MRF) intent channel.
-//!
-//! Producers on error paths (read decode failure, scanner metadata
-//! corruption, partial-write recovery) hand a lightweight [`MrfIntent`] to the
-//! heal crate through a global bounded channel. Delivery is strictly
-//! non-blocking: `try_send_mrf_intent` never awaits and drops the intent
-//! (counting it) when the channel is full or uninitialized — losing one heal
-//! hint is always preferred over stalling an IO path. Durable replay of
-//! unconsumed intents is the consumer's job (see `rustfs-heal`
-//! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`.
-
-use std::sync::{
-    Arc, OnceLock,
-    atomic::{AtomicBool, Ordering},
-};
-use tokio::sync::mpsc;
-use uuid::Uuid;
-
-/// Bounded capacity of the global MRF channel. Backpressure is resolved by
-/// dropping (and counting) intents, never by blocking the producer.
-const MRF_CHANNEL_CAPACITY: usize = 8192;
-
-/// Why an intent was produced. Drives the heal priority mapping on the
-/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High,
-/// PartialWrite -> Normal).
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum MrfKind {
-    /// Erasure decode failed while serving a read (read path).
-    DecodeFailure,
-    /// Scanner classified object metadata as corrupt.
-    MetadataCorruption,
-    /// A write left the object with fewer committed shards than the set size.
-    PartialWrite,
-}
-
-impl MrfKind {
-    pub const fn as_str(self) -> &'static str {
-        match self {
-            MrfKind::DecodeFailure => "decode-failure",
-            MrfKind::MetadataCorruption => "metadata-corruption",
-            MrfKind::PartialWrite => "partial-write",
-        }
-    }
-}
-
-/// One repair intent. Kept deliberately small so the in-memory queue and the
-/// journal stay bounded; `bucket`/`object` are `Arc<str>` so re-arming an
-/// intent never re-allocates the strings.
-#[derive(Clone, Debug)]
-pub struct MrfIntent {
-    pub bucket: Arc<str>,
-    pub object: Arc<str>,
-    /// Version the intent targets, as raw UUID bytes.
-    pub version_id: Option<[u8; 16]>,
-    pub kind: MrfKind,
-    pub enqueued_at_ms: u64,
-    /// Times this intent has already been offered to the heal manager.
-    /// Dropped by the consumer once it reaches `MRF_MAX_ATTEMPTS`.
-    pub attempts: u8,
-}
-
-/// Consumer-side retry ceiling before an intent is given up on.
-pub const MRF_MAX_ATTEMPTS: u8 = 3;
-
-impl MrfIntent {
-    /// Rough in-memory footprint used by the queue's byte budget.
-    pub fn estimated_bytes(&self) -> usize {
-        // Struct + strings + version bytes; buckets and objects are usually
-        // far below this bound, so rounding up keeps the budget conservative.
-        64 + self.bucket.len() + self.object.len()
-    }
-}
-
-static GLOBAL_MRF_SENDER: OnceLock<mpsc::Sender<MrfIntent>> = OnceLock::new();
-
-/// Delivery kill-switch, set from `RUSTFS_HEAL_MRF_ENABLE`. Producers check
-/// this before touching the channel so the disabled path stays allocation- and
-/// sync-free.
-static MRF_DELIVERY_ENABLED: AtomicBool = AtomicBool::new(true);
-
-/// Override delivery (used at heal-runtime startup from configuration).
-pub fn set_mrf_delivery_enabled(enabled: bool) {
-    MRF_DELIVERY_ENABLED.store(enabled, Ordering::Relaxed);
-}
-
-/// Whether producers currently deliver intents.
-pub fn mrf_delivery_enabled() -> bool {
-    MRF_DELIVERY_ENABLED.load(Ordering::Relaxed)
-}
-
-/// Create the global MRF channel and return the consumer half. Fails if the
-/// channel is already initialized (the heal runtime is a singleton).
-pub fn init_mrf_channel() -> Result<mpsc::Receiver<MrfIntent>, &'static str> {
-    let (sender, receiver) = mpsc::channel(MRF_CHANNEL_CAPACITY);
-    GLOBAL_MRF_SENDER
-        .set(sender)
-        .map_err(|_| "MRF channel sender already initialized")?;
-    Ok(receiver)
-}
-
-/// Best-effort, non-blocking intent delivery from an error path.
-///
-/// Returns `true` when the intent was accepted into the channel. `false`
-/// means the intent was dropped (feature disabled, channel not yet
-/// initialized, or channel full) — callers must not retry or await; the
-/// existing read-repair / scanner heal paths remain the safety net.
-///
-/// This runs on IO error paths, so it stays synchronous and cheap: one
-/// bounded allocation for the two `Arc<str>` handles plus the channel slot.
-pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option<Uuid>) -> bool {
-    if !mrf_delivery_enabled() {
-        return false;
-    }
-    let Some(sender) = GLOBAL_MRF_SENDER.get() else {
-        return false;
-    };
-    let intent = MrfIntent {
-        bucket: Arc::from(bucket),
-        object: Arc::from(object),
-        version_id: version_id.map(|vid| *vid.as_bytes()),
-        kind,
-        enqueued_at_ms: unix_now_ms(),
-        attempts: 0,
-    };
-    sender.try_send(intent).is_ok()
-}
-
-fn unix_now_ms() -> u64 {
-    // Kept trivial: the timestamp is diagnostic metadata only; wall-clock
-    // failure would be a bug rather than something to handle here.
-    std::time::SystemTime::now()
-        .duration_since(std::time::UNIX_EPOCH)
-        .map(|d| d.as_millis() as u64)
-        .unwrap_or(0)
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn intents_estimate_is_conservative() {
-        let intent = MrfIntent {
-            bucket: Arc::from("bucket"),
-            object: Arc::from("object"),
-            version_id: Some([0u8; 16]),
-            kind: MrfKind::DecodeFailure,
-            enqueued_at_ms: 0,
-            attempts: 0,
-        };
-        assert!(intent.estimated_bytes() >= intent.bucket.len() + intent.object.len());
-    }
-
-    #[tokio::test]
-    async fn try_send_delivers_and_respects_capacity() {
-        let mut receiver = init_mrf_channel().expect("first initialization should succeed");
-        assert!(init_mrf_channel().is_err(), "double initialization must fail");
-
-        assert!(try_send_mrf_intent(MrfKind::DecodeFailure, "b", "o", Some(Uuid::nil())));
-        let intent = receiver.recv().await.expect("intent should arrive");
-        assert_eq!(intent.kind, MrfKind::DecodeFailure);
-        assert_eq!(intent.bucket.as_ref(), "b");
-
-        // Disable delivery: producers become no-ops.
-        set_mrf_delivery_enabled(false);
-        assert!(!try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None));
-        set_mrf_delivery_enabled(true);
-
-        // Fill the bounded channel past capacity: excess intents are dropped,
-        // never blocking.
-        let mut accepted = 0;
-        for _ in 0..(MRF_CHANNEL_CAPACITY + 64) {
-            if try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None) {
-                accepted += 1;
-            }
-        }
-        assert_eq!(accepted, MRF_CHANNEL_CAPACITY);
-    }
-
-    #[test]
-    fn try_send_without_channel_is_false() {
-        // This test may run after the tokio test above in the same process;
-        // the singleton semantics make a clean "uninitialized" case hard, so
-        // assert the flag-off behavior only.
-        set_mrf_delivery_enabled(false);
-        assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None));
-        set_mrf_delivery_enabled(true);
-    }
-}
@@ -14,8 +14,9 @@

 //! Shared backpressure policy type.
 //!
-//! This module only carries the watermark policy; the admission primitive it
-//! projects into lives in `rustfs-io-core`.
+//! The runtime backpressure implementation (byte-watermark pipes and
+//! monitors) lives in `rustfs/src/storage/backpressure.rs`; this module only
+//! carries the watermark policy type that implementation shares.

 use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig;

@@ -177,31 +177,3 @@ pub const DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: usize = 80;

 /// Default foreground pressure recheck delay for heal scheduler, in milliseconds.
 pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250;
-
-/// Environment variable that toggles the MRF (mission repair feed) intent
-/// pipeline: error paths deliver repair intents to the heal runtime, and
-/// unconsumed intents are replayed from the durable journal after a restart.
-pub const ENV_HEAL_MRF_ENABLE: &str = "RUSTFS_HEAL_MRF_ENABLE";
-
-/// Environment variable for the MRF in-memory queue capacity (intent count).
-pub const ENV_HEAL_MRF_QUEUE_SIZE: &str = "RUSTFS_HEAL_MRF_QUEUE_SIZE";
-
-/// Environment variable for the MRF journal byte budget. The journal is
-/// compacted once its on-disk size crosses this bound.
-pub const ENV_HEAL_MRF_JOURNAL_MAX_BYTES: &str = "RUSTFS_HEAL_MRF_JOURNAL_MAX_BYTES";
-
-/// Environment variable for the MRF journal replay batch size (intents per
-/// replay push round).
-pub const ENV_HEAL_MRF_REPLAY_BATCH: &str = "RUSTFS_HEAL_MRF_REPLAY_BATCH";
-
-/// Default behavior keeps the MRF intent pipeline enabled.
-pub const DEFAULT_HEAL_MRF_ENABLE: bool = true;
-
-/// Default MRF queue capacity (matches MinIO's 100k MRF list ceiling).
-pub const DEFAULT_HEAL_MRF_QUEUE_SIZE: usize = 100_000;
-
-/// Default MRF journal byte budget (8 MiB), mirroring the channel payload cap.
-pub const DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES: usize = 8 * 1024 * 1024;
-
-/// Default MRF replay batch size.
-pub const DEFAULT_HEAL_MRF_REPLAY_BATCH: usize = 256;
@@ -234,31 +234,6 @@ pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_A
 /// Default absolute per-object erasure write cap in seconds (`0` = disabled).
 pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0;

-/// Enable foreground PutObject request admission.
-///
-/// This is an experimental, default-off foreground write backpressure gate for
-/// strict commit tail investigations. When disabled, PUTs follow the legacy
-/// path and only the existing request counters are updated.
-pub const ENV_PUT_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE";
-pub const DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE: bool = false;
-
-/// Maximum foreground PutObject requests admitted concurrently per process.
-///
-/// The limit is used only when [`ENV_PUT_FOREGROUND_ADMISSION_ENABLE`] is true.
-/// A value of `0` disables the gate even when the enable flag is present, so a
-/// partially configured rollout cannot reject every PUT.
-pub const ENV_PUT_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT";
-pub const DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT: usize = 0;
-
-/// Time in milliseconds a foreground PutObject waits for an admission permit.
-///
-/// Once this timeout expires the request fails before body ingest/storage
-/// mutation with S3 `SlowDown`/503. `0` means fail fast when the limit is full.
-pub const ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
-pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0;
-
-const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE);
-
 /// Environment variable for minimum GetObject timeout in seconds.
 ///
 /// When dynamic timeout calculation is enabled, this is the minimum timeout
@@ -17,6 +17,7 @@
 use crate::common::{RustFSTestEnvironment, init_logging, local_http_client};
 use async_compression::tokio::write::{BzEncoder, XzEncoder};
 use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
+use aws_sdk_s3::operation::head_object::HeadObjectOutput;
 use aws_sdk_s3::primitives::ByteStream;
 use aws_sdk_s3::types::{
    ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule,
@@ -349,6 +350,71 @@ async fn run_post_object_policy_case(
    Ok(())
 }

+/// One accepted POST Object upload driven end-to-end (backlog#1838): starts a
+/// fresh server, allows anonymous PutObject on `bucket`, posts an anonymous
+/// POST Object form whose policy carries `policy_conditions` and whose form
+/// carries `form_field` on top of the mandatory key+policy fields, then asserts
+/// 204 with an empty body, that `read_stored` observes the submitted value on
+/// the stored object, and that the object body round-tripped unchanged.
+/// `case` prefixes every assertion message so a failing table row is
+/// identifiable at a glance.
+#[allow(clippy::too_many_arguments)]
+async fn run_post_object_accept_case(
+    bucket: &str,
+    object_key: &str,
+    policy_conditions: Vec<serde_json::Value>,
+    form_field: (&str, &str),
+    file_mime: &str,
+    file_body: &[u8],
+    read_stored: fn(&HeadObjectOutput) -> Option<&str>,
+    case: &str,
+) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+    let mut env = RustFSTestEnvironment::new().await?;
+    env.start_rustfs_server(vec![]).await?;
+
+    let admin_client = env.create_s3_client();
+    admin_client.create_bucket().bucket(bucket).send().await?;
+    allow_anonymous_put_object(&admin_client, bucket).await?;
+
+    let policy = encode_post_policy(policy_conditions);
+
+    let (field_name, field_value) = form_field;
+    let post_form = reqwest::multipart::Form::new()
+        .text("key", object_key.to_string())
+        .text("policy", policy)
+        .text(field_name.to_string(), field_value.to_string())
+        .part(
+            "file",
+            reqwest::multipart::Part::bytes(file_body.to_vec())
+                .file_name("upload.txt")
+                .mime_str(file_mime)?,
+        );
+
+    let post_resp = local_http_client()
+        .post(format!("{}/{}", env.url, bucket))
+        .multipart(post_form)
+        .send()
+        .await?;
+
+    let status = post_resp.status();
+    let response_body = post_resp.text().await?;
+
+    assert_eq!(status, reqwest::StatusCode::NO_CONTENT, "[{case}] unexpected status");
+    assert!(
+        response_body.is_empty(),
+        "[{case}] 204 response should not contain a body, got: {response_body}"
+    );
+
+    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
+    assert_eq!(read_stored(&head), Some(field_value), "[{case}] stored {field_name} mismatch");
+
+    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
+    let uploaded = get_out.body.collect().await?.into_bytes();
+    assert_eq!(uploaded.as_ref(), file_body, "[{case}] uploaded body mismatch");
+
+    Ok(())
+}
+
 /// Table-driven fold of the nine `*_missing_from_policy_conditions` POST
 /// Object tests (backlog#1838 PR1). Every row keeps its original test's exact
 /// bucket, key, form field, file body, and expected error strings; the shared
@@ -1551,60 +1617,6 @@ async fn test_anonymous_post_object_accepts_sse_s3_missing_from_policy_condition
    Ok(())
 }

-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_storage_class_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-storage-class";
-    let object_key = "post-storage-class-object.txt";
-    let expected_body = b"post-storage-class-body".to_vec();
-    let storage_class = "REDUCED_REDUNDANCY";
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-storage-class": storage_class }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-storage-class", storage_class)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    assert_eq!(post_resp.status(), reqwest::StatusCode::NO_CONTENT);
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.storage_class().map(|value| value.as_str()), Some(storage_class));
-
-    let uploaded = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = uploaded.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
 #[tokio::test]
 #[serial]
 async fn test_anonymous_post_object_rejects_storage_class_missing_from_policy_conditions()
@@ -2620,521 +2632,183 @@ async fn test_anonymous_post_object_rejects_success_action_redirect_missing_from
    Ok(())
 }

+/// Table-driven fold of the eleven accepted POST Object form-field tests
+/// (backlog#1838 PR4). Every row keeps its original test's exact bucket, key,
+/// form field, submitted value, policy condition, file MIME type, and file
+/// body; the shared shape is: the policy covers the field (exact condition or
+/// `starts-with` prefix), the form submits it, the upload returns 204 with an
+/// empty body, and the stored object echoes the submitted value back.
 #[tokio::test]
 #[serial]
-async fn test_anonymous_post_object_accepts_metadata_field_covered_by_starts_with()
+async fn test_anonymous_post_object_accepts_fields_covered_by_policy_conditions()
 -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
    init_logging();

-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
+    // (case, bucket, object_key, field, submitted value, `starts-with` prefix
+    // (`None` pins the field to an exact policy condition), file part MIME type,
+    // file body, stored-value accessor)
+    type Case = (
+        &'static str,
+        &'static str,
+        &'static str,
+        &'static str,
+        &'static str,
+        Option<&'static str>,
+        &'static str,
+        &'static [u8],
+        fn(&HeadObjectOutput) -> Option<&str>,
+    );
+    let cases: &[Case] = &[
+        (
+            "storage-class",
+            "anon-post-storage-class",
+            "post-storage-class-object.txt",
+            "x-amz-storage-class",
+            "REDUCED_REDUNDANCY",
+            None,
+            "text/plain",
+            b"post-storage-class-body",
+            |head: &HeadObjectOutput| head.storage_class().map(|value| value.as_str()),
+        ),
+        (
+            "metadata-starts-with",
+            "anon-post-policy-meta-accept",
+            "uploads/meta-object.txt",
+            "x-amz-meta-project",
+            "alpha-demo",
+            Some("alpha-"),
+            "text/plain",
+            b"post-policy-meta-body",
+            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
+        ),
+        (
+            "content-type",
+            "anon-post-policy-content-type-accept",
+            "uploads/content-type-accept.txt",
+            "Content-Type",
+            "text/plain",
+            None,
+            "text/plain",
+            b"post-policy-content-type-accept",
+            |head: &HeadObjectOutput| head.content_type(),
+        ),
+        (
+            "content-type-starts-with",
+            "anon-post-policy-content-type-accept",
+            "uploads/content-type-object.txt",
+            "Content-Type",
+            "image/png",
+            Some("image/"),
+            "image/png",
+            b"post-policy-content-type-body",
+            |head: &HeadObjectOutput| head.content_type(),
+        ),
+        (
+            "content-disposition",
+            "anon-post-policy-disposition-accept",
+            "uploads/disposition-object.txt",
+            "Content-Disposition",
+            "attachment; filename=\"upload.txt\"",
+            None,
+            "text/plain",
+            b"post-policy-disposition-body",
+            |head: &HeadObjectOutput| head.content_disposition(),
+        ),
+        (
+            "cache-control",
+            "anon-post-policy-cache-control-accept",
+            "uploads/cache-control-object.txt",
+            "Cache-Control",
+            "max-age=60",
+            None,
+            "text/plain",
+            b"post-policy-cache-control-body",
+            |head: &HeadObjectOutput| head.cache_control(),
+        ),
+        (
+            "content-language",
+            "anon-post-policy-content-language-accept",
+            "uploads/content-language-object.txt",
+            "Content-Language",
+            "en-US",
+            None,
+            "text/plain",
+            b"post-policy-content-language-body",
+            |head: &HeadObjectOutput| head.content_language(),
+        ),
+        (
+            "content-encoding",
+            "anon-post-policy-content-encoding-accept",
+            "uploads/content-encoding-object.txt",
+            "Content-Encoding",
+            "gzip",
+            None,
+            "text/plain",
+            b"post-policy-content-encoding-body",
+            |head: &HeadObjectOutput| head.content_encoding(),
+        ),
+        (
+            "website-redirect-location",
+            "anon-post-policy-website-redirect-accept",
+            "uploads/website-redirect-object.txt",
+            "x-amz-website-redirect-location",
+            "/docs/landing.html",
+            None,
+            "text/plain",
+            b"post-policy-website-redirect-body",
+            |head: &HeadObjectOutput| head.website_redirect_location(),
+        ),
+        (
+            "expires",
+            "anon-post-policy-expires-accept",
+            "uploads/expires-object.txt",
+            "Expires",
+            "Wed, 21 Oct 2037 07:28:00 GMT",
+            None,
+            "text/plain",
+            b"post-policy-expires-body",
+            |head: &HeadObjectOutput| head.expires_string(),
+        ),
+        (
+            "metadata-exact",
+            "anon-post-policy-meta-exact-accept",
+            "uploads/meta-exact-accept-object.txt",
+            "x-amz-meta-project",
+            "alpha-demo",
+            None,
+            "text/plain",
+            b"post-policy-meta-exact-body",
+            |head: &HeadObjectOutput| head.metadata().and_then(|meta| meta.get("project")).map(String::as_str),
+        ),
+    ];

-    let bucket = "anon-post-policy-meta-accept";
-    let object_key = "uploads/meta-object.txt";
-    let metadata_value = "alpha-demo";
-    let expected_body = b"post-policy-meta-body".to_vec();
+    for (case, bucket, object_key, field, value, starts_with_prefix, file_mime, file_body, read_stored) in cases {
+        let condition = match starts_with_prefix {
+            Some(prefix) => serde_json::json!(["starts-with", format!("${field}"), prefix]),
+            None => {
+                let mut exact = serde_json::Map::new();
+                exact.insert((*field).to_string(), serde_json::Value::String((*value).to_string()));
+                serde_json::Value::Object(exact)
+            }
+        };

-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!(["starts-with", "$x-amz-meta-project", "alpha-"]),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-meta-project", metadata_value)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
+        run_post_object_accept_case(
+            bucket,
+            object_key,
+            vec![
+                serde_json::json!({ "bucket": bucket }),
+                serde_json::json!({ "key": object_key }),
+                condition,
+                serde_json::json!(["content-length-range", 0, 1024]),
+            ],
+            (field, value),
+            file_mime,
+            file_body,
+            *read_stored,
+            case,
+        )
        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
-    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_content_type_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-type-accept";
-    let object_key = "uploads/content-type-accept.txt";
-    let content_type = "text/plain";
-    let expected_body = b"post-policy-content-type-accept".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Type": content_type }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Type", content_type)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str(content_type)?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_type(), Some(content_type));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_content_type_field_covered_by_starts_with()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-type-accept";
-    let object_key = "uploads/content-type-object.txt";
-    let content_type = "image/png";
-    let expected_body = b"post-policy-content-type-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!(["starts-with", "$Content-Type", "image/"]),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Type", content_type)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str(content_type)?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_type(), Some(content_type));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_content_disposition_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-disposition-accept";
-    let object_key = "uploads/disposition-object.txt";
-    let content_disposition = "attachment; filename=\"upload.txt\"";
-    let expected_body = b"post-policy-disposition-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Disposition": content_disposition }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Disposition", content_disposition)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_disposition(), Some(content_disposition));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_cache_control_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-cache-control-accept";
-    let object_key = "uploads/cache-control-object.txt";
-    let cache_control = "max-age=60";
-    let expected_body = b"post-policy-cache-control-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Cache-Control": cache_control }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Cache-Control", cache_control)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.cache_control(), Some(cache_control));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_content_language_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-language-accept";
-    let object_key = "uploads/content-language-object.txt";
-    let content_language = "en-US";
-    let expected_body = b"post-policy-content-language-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Language": content_language }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Language", content_language)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_language(), Some(content_language));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_content_encoding_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-content-encoding-accept";
-    let object_key = "uploads/content-encoding-object.txt";
-    let content_encoding = "gzip";
-    let expected_body = b"post-policy-content-encoding-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Content-Encoding": content_encoding }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Content-Encoding", content_encoding)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.content_encoding(), Some(content_encoding));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_website_redirect_location_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-website-redirect-accept";
-    let object_key = "uploads/website-redirect-object.txt";
-    let website_redirect_location = "/docs/landing.html";
-    let expected_body = b"post-policy-website-redirect-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-website-redirect-location": website_redirect_location }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-website-redirect-location", website_redirect_location)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.website_redirect_location(), Some(website_redirect_location));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_expires_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-expires-accept";
-    let object_key = "uploads/expires-object.txt";
-    let expires = "Wed, 21 Oct 2037 07:28:00 GMT";
-    let expected_body = b"post-policy-expires-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "Expires": expires }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("Expires", expires)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    assert_eq!(head.expires_string(), Some(expires));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
+    }

    Ok(())
 }
@@ -3491,65 +3165,6 @@ async fn test_anonymous_post_object_accepts_tagging_field_exact_policy_match()
    Ok(())
 }

-#[tokio::test]
-#[serial]
-async fn test_anonymous_post_object_accepts_metadata_field_exact_policy_match()
-> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    init_logging();
-
-    let mut env = RustFSTestEnvironment::new().await?;
-    env.start_rustfs_server(vec![]).await?;
-
-    let bucket = "anon-post-policy-meta-exact-accept";
-    let object_key = "uploads/meta-exact-accept-object.txt";
-    let metadata_value = "alpha-demo";
-    let expected_body = b"post-policy-meta-exact-body".to_vec();
-
-    let admin_client = env.create_s3_client();
-    admin_client.create_bucket().bucket(bucket).send().await?;
-    allow_anonymous_put_object(&admin_client, bucket).await?;
-
-    let policy = encode_post_policy(vec![
-        serde_json::json!({ "bucket": bucket }),
-        serde_json::json!({ "key": object_key }),
-        serde_json::json!({ "x-amz-meta-project": metadata_value }),
-        serde_json::json!(["content-length-range", 0, 1024]),
-    ]);
-
-    let post_form = reqwest::multipart::Form::new()
-        .text("key", object_key.to_string())
-        .text("policy", policy)
-        .text("x-amz-meta-project", metadata_value)
-        .part(
-            "file",
-            reqwest::multipart::Part::bytes(expected_body.clone())
-                .file_name("upload.txt")
-                .mime_str("text/plain")?,
-        );
-
-    let post_resp = local_http_client()
-        .post(format!("{}/{}", env.url, bucket))
-        .multipart(post_form)
-        .send()
-        .await?;
-
-    let status = post_resp.status();
-    let response_body = post_resp.text().await?;
-
-    assert_eq!(status, reqwest::StatusCode::NO_CONTENT);
-    assert!(response_body.is_empty(), "204 response should not contain a body, got: {response_body}");
-
-    let head = admin_client.head_object().bucket(bucket).key(object_key).send().await?;
-    let metadata = head.metadata().expect("head_object should expose uploaded metadata");
-    assert_eq!(metadata.get("project").map(String::as_str), Some(metadata_value));
-
-    let get_out = admin_client.get_object().bucket(bucket).key(object_key).send().await?;
-    let uploaded = get_out.body.collect().await?.into_bytes();
-    assert_eq!(uploaded.as_ref(), expected_body.as_slice());
-
-    Ok(())
-}
-
 #[tokio::test]
 #[serial]
 async fn test_anonymous_post_object_allows_x_ignore_fields_outside_policy_conditions()
@@ -1783,7 +1783,7 @@ impl TransitionState {
                                .await;
                            }
                            global_metrics().record_scanner_transition_failed(1);
-                            if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) {
+                            if !is_err_version_not_found(&err) && !is_err_object_not_found(&err) && !is_network_or_host_down(&err.to_string(), false) && !err.to_string().contains("use of closed network connection") {
                                error!(
                                    event = EVENT_LIFECYCLE_TIER_OPERATION_FAILED,
                                    component = LOG_COMPONENT_ECSTORE,
@@ -20,7 +20,6 @@
 #![allow(clippy::all)]

 use http::{HeaderMap, HeaderName, HeaderValue};
-use rustfs_utils::http::headers::AMZ_CHECKSUM_MODE;
 use std::collections::HashMap;
 use time::OffsetDateTime;
 use tracing::warn;
@@ -77,7 +76,7 @@ impl GetObjectOptions {
            }
        }
        if self.checksum {
-            headers.insert(HeaderName::from_static(AMZ_CHECKSUM_MODE), HeaderValue::from_static("ENABLED"));
+            headers.insert(HeaderName::from_static("x-amz-checksum-mode"), HeaderValue::from_static("ENABLED"));
        }
        headers
    }
@@ -54,10 +54,6 @@ use rustfs_config::MAX_S3_CLIENT_RESPONSE_SIZE;
 use rustfs_rio::HashReader;
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::{
-    http::headers::{
-        AMZ_CHECKSUM_CRC32, AMZ_CHECKSUM_CRC32C, AMZ_CHECKSUM_CRC64NVME, AMZ_CHECKSUM_MODE, AMZ_CHECKSUM_SHA1,
-        AMZ_CHECKSUM_SHA256,
-    },
    net::get_endpoint_url,
    retry::{DEFAULT_RETRY_CAP, DEFAULT_RETRY_UNIT, MAX_JITTER, MAX_RETRY, RetryTimer},
 };
@@ -1387,12 +1383,12 @@ pub(crate) fn to_object_info_for_provider(
    };

    // Extract checksums
-    let checksum_crc32 = get_header(AMZ_CHECKSUM_CRC32);
-    let checksum_crc32c = get_header(AMZ_CHECKSUM_CRC32C);
-    let checksum_sha1 = get_header(AMZ_CHECKSUM_SHA1);
-    let checksum_sha256 = get_header(AMZ_CHECKSUM_SHA256);
-    let checksum_crc64nvme = get_header(AMZ_CHECKSUM_CRC64NVME);
-    let checksum_mode = get_header(AMZ_CHECKSUM_MODE);
+    let checksum_crc32 = get_header("x-amz-checksum-crc32");
+    let checksum_crc32c = get_header("x-amz-checksum-crc32c");
+    let checksum_sha1 = get_header("x-amz-checksum-sha1");
+    let checksum_sha256 = get_header("x-amz-checksum-sha256");
+    let checksum_crc64nvme = get_header("x-amz-checksum-crc64nvme");
+    let checksum_mode = get_header("x-amz-checksum-mode");

    // Build and return the ObjectInfo struct
    Ok(ObjectInfo {
@@ -86,25 +86,6 @@ const PEER_REST_RECOVERY_MAX_BACKOFF: Duration = Duration::from_secs(30);
 const SCANNER_ACTIVITY_MAX_MESSAGE_SIZE: usize = 1024;
 const REPLICATION_STATS_MAX_MESSAGE_SIZE: usize = 8 * 1024 * 1024;

-/// Error for a peer that reported `success = false` without an `error_info` payload.
-///
-/// Same shape as `peer_s3_client::peer_failure_without_details`, over `StorageError`
-/// instead of `DiskError`. The message names the operation (and the bucket, where the
-/// operation has one) and nothing else, for two reasons:
-///
-/// - `finalize_result` classifies failures by message substring, so any text matching
-///   `message_has_network_needle` would take an answering peer offline and evict its
-///   connection over a plain application-level rejection.
-/// - Quorum aggregation (`reduce_errs`) buckets `Io` errors by kind plus rendered
-///   message, so a per-peer detail such as the peer address would split one shared
-///   failure into single-count buckets and downgrade the dominant error.
-fn peer_failure_without_details(op: &str, bucket: Option<&str>) -> Error {
-    match bucket {
-        Some(bucket) => Error::other(format!("{op}({bucket}): peer returned failure without error details")),
-        None => Error::other(format!("{op}: peer returned failure without error details")),
-    }
-}
-
 fn decode_bucket_stats_response(response: GetBucketStatsDataResponse) -> Result<BucketStats> {
    if !response.success {
        return Err(Error::other(
@@ -715,7 +696,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("local_storage_info", None));
+            return Err(Error::other(""));
        }
        let data = response.storage_info;

@@ -738,7 +719,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("server_info", None));
+            return Err(Error::other(""));
        }
        let data = response.server_properties;

@@ -761,7 +742,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("get_cpus", None));
+            return Err(Error::other(""));
        }
        let data = response.cpus;

@@ -784,7 +765,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("get_net_info", None));
+            return Err(Error::other(""));
        }
        let data = response.net_info;

@@ -807,7 +788,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("get_partitions", None));
+            return Err(Error::other(""));
        }
        let data = response.partitions;

@@ -830,7 +811,7 @@ impl PeerRestClient {
            if let Some(msg) = response.error_info {
                return Err(Error::other(msg));
            }
-            return Err(peer_failure_without_details("get_os_info", None));
+            return Err(Error::other(""));
        }
        let data = response.os_info;

@@ -851,7 +832,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_se_linux_info", None));
+                    return Err(Error::other(""));
                }
                let data = response.sys_services;

@@ -876,7 +857,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_sys_config", None));
+                    return Err(Error::other(""));
                }
                let data = response.sys_config;

@@ -901,7 +882,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_sys_errors", None));
+                    return Err(Error::other(""));
                }
                let data = response.sys_errors;

@@ -926,7 +907,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_mem_info", None));
+                    return Err(Error::other(""));
                }
                let data = response.mem_info;

@@ -958,7 +939,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_metrics", None));
+                    return Err(Error::other(""));
                }
                let data = response.realtime_metrics;

@@ -983,7 +964,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_live_events", None));
+                    return Err(Error::other(""));
                }

                Ok(PeerLiveEventsBatch {
@@ -1008,7 +989,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("get_proc_info", None));
+                    return Err(Error::other(""));
                }
                let data = response.proc_info;

@@ -1035,7 +1016,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("start_profiling", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1342,7 +1323,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_bucket_metadata", Some(bucket)));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1365,7 +1346,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("delete_bucket_metadata", Some(bucket)));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1388,7 +1369,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("delete_policy", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1411,7 +1392,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_policy", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1436,7 +1417,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_policy_mapping", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1459,7 +1440,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("delete_user", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1482,7 +1463,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("delete_service_account", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1506,7 +1487,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_user", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1529,7 +1510,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_service_account", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1552,7 +1533,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_group", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1573,7 +1554,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("reload_site_replication_config", None));
+                    return Err(Error::other(""));
                }
                Ok(())
            }
@@ -1616,7 +1597,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("signal_service", None));
+                    return Err(Error::other(""));
                }
                validate_signal_service_protocol(sig, sub_sys, response.protocol_version)?;
                Ok(response)
@@ -1686,7 +1667,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("reload_pool_meta", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1710,7 +1691,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("stop_rebalance", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1744,7 +1725,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("load_rebalance_meta", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1772,7 +1753,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("start_decommission", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1796,7 +1777,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("decommission_cancel", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1820,7 +1801,7 @@ impl PeerRestClient {
                    if let Some(msg) = response.error_info {
                        return Err(Error::other(msg));
                    }
-                    return Err(peer_failure_without_details("clear_decommission", None));
+                    return Err(Error::other(""));
                }

                Ok(())
@@ -1966,8 +1947,6 @@ fn tier_config_reload_status_outcome(status: tonic::Status) -> TierConfigReloadO
 mod tests {
    use super::*;
    use crate::config::com::STORAGE_CLASS_SUB_SYS;
-    use crate::disk::error::DiskError;
-    use crate::disk::error_reduce::reduce_errs;
    use crate::layout::{disks_layout::DisksLayout, endpoints::SetupType};
    use rustfs_config::{ENV_KUBERNETES_SERVICE_HOST, ENV_LOCAL_ENDPOINT_HOST, ENV_STARTUP_TOPOLOGY_WAIT_MODE};
    use serde_json::Value;
@@ -3119,115 +3098,4 @@ mod tests {
                && span.get("request_id").and_then(Value::as_str) == Some("req-peer-rest")
        }));
    }
-
-    /// Every operation name passed to `peer_failure_without_details` in this file.
-    const PEER_FAILURE_OPS: &[&str] = &[
-        "local_storage_info",
-        "server_info",
-        "get_cpus",
-        "get_net_info",
-        "get_partitions",
-        "get_os_info",
-        "get_se_linux_info",
-        "get_sys_config",
-        "get_sys_errors",
-        "get_mem_info",
-        "get_metrics",
-        "get_live_events",
-        "get_proc_info",
-        "start_profiling",
-        "load_bucket_metadata",
-        "delete_bucket_metadata",
-        "delete_policy",
-        "load_policy",
-        "load_policy_mapping",
-        "delete_user",
-        "delete_service_account",
-        "load_user",
-        "load_service_account",
-        "load_group",
-        "reload_site_replication_config",
-        "signal_service",
-        "reload_pool_meta",
-        "stop_rebalance",
-        "load_rebalance_meta",
-        "start_decommission",
-        "decommission_cancel",
-        "clear_decommission",
-    ];
-
-    #[test]
-    fn peer_failure_without_details_names_operation_and_bucket() {
-        for op in PEER_FAILURE_OPS {
-            let message = peer_failure_without_details(op, None).to_string();
-            assert!(message.contains(op), "{op} message must name the operation: {message}");
-        }
-
-        for op in ["load_bucket_metadata", "delete_bucket_metadata"] {
-            let message = peer_failure_without_details(op, Some("ops-bucket")).to_string();
-            assert!(message.contains(op), "{op} message must name the operation: {message}");
-            assert!(message.contains("ops-bucket"), "{op} message must name the bucket: {message}");
-        }
-    }
-
-    #[test]
-    fn peer_failure_without_details_keeps_one_reduce_errs_bucket_per_operation() {
-        // reduce_errs groups Io errors by kind plus rendered message: peers failing the
-        // same operation must stay a single dominant error instead of one bucket per peer.
-        let per_peer_errs = (0..4)
-            .map(|_| Some(DiskError::from(peer_failure_without_details("load_bucket_metadata", Some("shared")))))
-            .collect::<Vec<_>>();
-        let (count, dominant) = reduce_errs(&per_peer_errs, &[]);
-        assert_eq!(count, 4, "one shared failure must not split into per-peer buckets");
-        assert_eq!(
-            dominant,
-            Some(DiskError::from(peer_failure_without_details("load_bucket_metadata", Some("shared"))))
-        );
-
-        assert_ne!(
-            peer_failure_without_details("load_bucket_metadata", Some("shared")).to_string(),
-            peer_failure_without_details("delete_bucket_metadata", Some("shared")).to_string()
-        );
-        assert_ne!(
-            peer_failure_without_details("load_bucket_metadata", Some("bucket-a")).to_string(),
-            peer_failure_without_details("load_bucket_metadata", Some("bucket-b")).to_string()
-        );
-    }
-
-    #[test]
-    fn peer_failure_without_details_never_reads_as_a_network_failure() {
-        // `finalize_result` marks the peer offline and evicts its connection whenever the
-        // message matches a network needle. A peer that answered `success = false` is alive,
-        // so no operation or bucket name may push this text over that classifier.
-        for op in PEER_FAILURE_OPS {
-            let err = peer_failure_without_details(op, None);
-            assert!(
-                !PeerRestClient::is_network_like_error(&err),
-                "{op} must not read as a transport failure: {err}"
-            );
-
-            let scoped = peer_failure_without_details(op, Some("bucket-name"));
-            assert!(
-                !PeerRestClient::is_network_like_error(&scoped),
-                "{op} must not read as a transport failure: {scoped}"
-            );
-        }
-
-        // The bucket name is caller-supplied. Every needle carries a space, which S3 bucket
-        // names cannot, and the name is closed by `)` before the literal text resumes, so no
-        // needle can straddle the boundary either.
-        for bucket in [
-            "timed-out",
-            "connection-reset",
-            "transport-error",
-            "broken-pipe",
-            "unavailable-logs",
-        ] {
-            let err = peer_failure_without_details("load_bucket_metadata", Some(bucket));
-            assert!(
-                !PeerRestClient::is_network_like_error(&err),
-                "bucket {bucket} must not push the message over the network classifier: {err}"
-            );
-        }
-    }
 }
@@ -220,8 +220,6 @@ fn pool_write_quorum(participant_count: usize) -> usize {
 /// buckets `Error::Io` by kind plus rendered message, so any per-peer detail (address,
 /// timing) would split one shared failure into single-count buckets and downgrade a real
 /// dominant error into `ErasureWriteQuorum`.
-///
-/// `peer_rest_client` carries the same helper over `StorageError` for the same response shape.
 fn peer_failure_without_details(op: &str, bucket: Option<&str>) -> Error {
    match bucket {
        Some(bucket) => Error::other(format!("{op}({bucket}): peer returned failure without error details")),
@@ -1116,14 +1116,6 @@ mod tests {
        assert!(encoder_source.is::<reed_solomon_erasure::Error>());
    }

-    // The lifecycle transition worker relies on this arm alone to suppress the
-    // closed-connection noise (`bucket_lifecycle_ops.rs`); dropping it here would
-    // silently turn shutdown races back into `error!` log spam.
-    #[test]
-    fn is_network_or_host_down_covers_closed_network_connection() {
-        assert!(is_network_or_host_down("transition failed: use of closed network connection", false));
-    }
-
    // Regression for #952 (ECA-11): an all-`DiskNotFound` slice (every drive in
    // every set unreachable) must NOT be classified as "all not found",
    // otherwise ListObjects silently returns an empty listing and masks a full
@@ -5845,14 +5845,6 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {

    #[tracing::instrument(skip(self))]
    async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> {
-        // MRF journal intent: partial-write recovery must survive a restart
-        // (HS-01); the heal request below remains the in-memory fast path.
-        rustfs_common::mrf_channel::try_send_mrf_intent(
-            rustfs_common::mrf_channel::MrfKind::PartialWrite,
-            bucket,
-            object,
-            uuid::Uuid::try_parse(version_id).ok(),
-        );
        let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
            bucket.to_string(),
            Some(object.to_string()),
@@ -1077,15 +1077,6 @@ impl SetDisks {
                            "Recoverable decode error triggered read repair"
                        );
                        let version_id = fi.version_id.as_ref().map(ToString::to_string);
-                        // MRF journal intent: keeps a durable Urgent ECDecode
-                        // request alive across restarts even when the in-memory
-                        // read-repair request is dropped or lost (HS-01).
-                        rustfs_common::mrf_channel::try_send_mrf_intent(
-                            rustfs_common::mrf_channel::MrfKind::DecodeFailure,
-                            bucket,
-                            object,
-                            fi.version_id,
-                        );
                        submit_read_repair_heal(
                            bucket,
                            object,
@@ -89,8 +89,6 @@ async-trait = { workspace = true }
 futures = { workspace = true }
 metrics = { workspace = true }
 base64 = { workspace = true }
-bytes = { workspace = true }
-crc-fast = { workspace = true }

 [dev-dependencies]
 serde_json = { workspace = true, features = ["raw_value"] }
@@ -612,8 +612,7 @@ impl HealChannelProcessor {
            HealRequestSource::Admin
            | HealRequestSource::AutoHeal
            | HealRequestSource::Internal
-            | HealRequestSource::ReadRepair
-            | HealRequestSource::Mrf => true,
+            | HealRequestSource::ReadRepair => true,
        });

        // Build HealOptions with all available fields
@@ -270,8 +270,6 @@ pub struct HealSourceCounts {
    pub auto_heal: u64,
    pub internal: u64,
    pub read_repair: u64,
-    #[serde(default)]
-    pub mrf: u64,
 }

 impl HealSourceCounts {
@@ -282,7 +280,6 @@ impl HealSourceCounts {
            HealRequestSource::AutoHeal => self.auto_heal += 1,
            HealRequestSource::Internal => self.internal += 1,
            HealRequestSource::ReadRepair => self.read_repair += 1,
-            HealRequestSource::Mrf => self.mrf += 1,
        }
    }
 }
@@ -16,7 +16,6 @@ pub mod channel;
 pub mod erasure_healer;
 pub mod event;
 pub mod manager;
-pub mod mrf_queue;
 pub mod progress;
 pub(crate) mod replacement_readiness;
 pub mod resume;
@@ -1,682 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! Mission Repair Feed (MRF) queue, journal, and consumer.
-//!
-//! Intents arriving on the global channel (see `rustfs_common::mrf_channel`)
-//! are buffered in a bounded in-memory queue, translated into prioritized
-//! heal requests, and — while they are not yet accepted by the heal manager —
-//! mirrored into a durable journal so a crash or restart can replay them.
-//! This is the RustFS counterpart of MinIO's `.heal/mrf/list.bin` replay,
-//! layered on top of (not replacing) read-repair and scanner heal.
-//!
-//! Durability model: the journal is a snapshot of the *unaccepted* pending
-//! set, rewritten on a group-commit cadence (every flush interval or flush
-//! threshold new intents). A rewrite is atomic at the record level only — a
-//! torn tail simply truncates during replay because every record carries its
-//! own CRC32. Losing the last flush window (≤500 ms) is acceptable: replayed
-//! duplicates are merged by the manager's dedup key, and read-repair remains
-//! the safety net.
-
-use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
-use crate::heal::manager::HealManager;
-use metrics::{counter, gauge};
-use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
-use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent};
-use std::collections::VecDeque;
-use std::sync::Arc;
-use std::time::Duration;
-use tokio::sync::mpsc;
-use uuid::Uuid;
-
-use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType};
-
-/// Journal location inside the metadata bucket, following the resume-state
-/// layout.
-pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin";
-
-/// Record format tag.
-const MRF_JOURNAL_FORMAT: u8 = 1;
-/// Record layout version.
-const MRF_JOURNAL_VERSION: u8 = 1;
-
-/// Fixed header size: format, version, kind, attempts, enqueued_at_ms,
-/// has_version flag.
-const MRF_RECORD_FIXED_HEAD: usize = 1 + 1 + 1 + 1 + 8 + 1;
-
-#[derive(Debug, Clone)]
-pub(crate) struct MrfConsumerConfig {
-    /// In-memory queue capacity in intents.
-    pub queue_capacity: usize,
-    /// Journal byte budget; a pending snapshot above this bound is rejected
-    /// oldest-first so the journal can never grow unbounded.
-    pub journal_max_bytes: usize,
-    /// How many journal intents to re-arm per replay round.
-    pub replay_batch: usize,
-    /// Group-commit cadence for the journal snapshot.
-    pub flush_interval: Duration,
-    /// New intents between flushes that force an early snapshot.
-    pub flush_threshold: usize,
-    /// Backoff after the heal manager reports a full admission.
-    pub admission_backoff: Duration,
-}
-
-impl Default for MrfConsumerConfig {
-    fn default() -> Self {
-        Self {
-            queue_capacity: rustfs_utils::get_env_usize(
-                rustfs_config::ENV_HEAL_MRF_QUEUE_SIZE,
-                rustfs_config::DEFAULT_HEAL_MRF_QUEUE_SIZE,
-            ),
-            journal_max_bytes: rustfs_utils::get_env_usize(
-                rustfs_config::ENV_HEAL_MRF_JOURNAL_MAX_BYTES,
-                rustfs_config::DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES,
-            ),
-            replay_batch: rustfs_utils::get_env_usize(
-                rustfs_config::ENV_HEAL_MRF_REPLAY_BATCH,
-                rustfs_config::DEFAULT_HEAL_MRF_REPLAY_BATCH,
-            ),
-            flush_interval: Duration::from_millis(500),
-            flush_threshold: 1000,
-            admission_backoff: Duration::from_secs(5),
-        }
-    }
-}
-
-/// Bounded pending set with count and byte ceilings. Overflow drops the
-/// incoming intent (never a resident one) and counts the loss.
-pub(crate) struct MrfQueue {
-    pending: VecDeque<MrfIntent>,
-    bytes: usize,
-    capacity: usize,
-    byte_budget: usize,
-}
-
-impl MrfQueue {
-    pub(crate) fn new(capacity: usize, byte_budget: usize) -> Self {
-        Self {
-            pending: VecDeque::new(),
-            bytes: 0,
-            capacity,
-            byte_budget,
-        }
-    }
-
-    /// Returns `false` (after counting) when either ceiling would be crossed.
-    pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool {
-        let cost = intent.estimated_bytes();
-        if self.pending.len() >= self.capacity || self.bytes + cost > self.byte_budget {
-            counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1);
-            return false;
-        }
-        self.bytes += cost;
-        self.pending.push_back(intent);
-        true
-    }
-
-    pub(crate) fn pop_front(&mut self) -> Option<MrfIntent> {
-        let intent = self.pending.pop_front()?;
-        self.bytes = self.bytes.saturating_sub(intent.estimated_bytes());
-        Some(intent)
-    }
-
-    pub(crate) fn push_back(&mut self, intent: MrfIntent) {
-        self.bytes += intent.estimated_bytes();
-        self.pending.push_back(intent);
-    }
-
-    pub(crate) fn depth(&self) -> usize {
-        self.pending.len()
-    }
-
-    pub(crate) fn bytes(&self) -> usize {
-        self.bytes
-    }
-
-    pub(crate) fn intents(&self) -> impl Iterator<Item = &MrfIntent> {
-        self.pending.iter()
-    }
-}
-
-// ---------------------------------------------------------------------------
-// Journal record codec
-// ---------------------------------------------------------------------------
-
-/// Append one encoded record to `out`.
-pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) {
-    let start = out.len();
-    out.push(MRF_JOURNAL_FORMAT);
-    out.push(MRF_JOURNAL_VERSION);
-    out.push(match intent.kind {
-        rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1,
-        rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2,
-        rustfs_common::mrf_channel::MrfKind::PartialWrite => 3,
-    });
-    out.push(intent.attempts);
-    out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes());
-    match intent.version_id {
-        Some(bytes) => {
-            out.push(1);
-            out.extend_from_slice(&bytes);
-        }
-        None => out.push(0),
-    }
-    out.extend_from_slice(&(intent.bucket.len() as u32).to_le_bytes());
-    out.extend_from_slice(&(intent.object.len() as u32).to_le_bytes());
-    out.extend_from_slice(intent.bucket.as_bytes());
-    out.extend_from_slice(intent.object.as_bytes());
-    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
-    hasher.update(&out[start..]);
-    out.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes());
-}
-
-fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> {
-    if data.len() < MRF_RECORD_FIXED_HEAD + 8 {
-        return None;
-    }
-    if data[0] != MRF_JOURNAL_FORMAT || data[1] != MRF_JOURNAL_VERSION {
-        return None;
-    }
-    let kind = match data[2] {
-        1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure,
-        2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
-        3 => rustfs_common::mrf_channel::MrfKind::PartialWrite,
-        _ => return None,
-    };
-    let attempts = data[3];
-    let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().expect("slice length checked"));
-    let has_version = data[12] != 0;
-    let mut cursor = MRF_RECORD_FIXED_HEAD;
-    let version_id = if has_version {
-        if data.len() < cursor + 16 {
-            return None;
-        }
-        let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().expect("slice length checked");
-        cursor += 16;
-        Some(bytes)
-    } else {
-        None
-    };
-    if data.len() < cursor + 8 {
-        return None;
-    }
-    let bucket_len = u32::from_le_bytes(data[cursor..cursor + 4].try_into().expect("slice length checked")) as usize;
-    let object_len = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().expect("slice length checked")) as usize;
-    cursor += 8;
-    let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?;
-    let record_end = body_end.checked_add(4)?;
-    if data.len() < record_end {
-        return None;
-    }
-    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
-    hasher.update(&data[..body_end]);
-    if (hasher.finalize() as u32) != u32::from_le_bytes(data[body_end..record_end].try_into().expect("slice length checked")) {
-        return None;
-    }
-    let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?);
-    let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?);
-    Some((
-        MrfIntent {
-            bucket,
-            object,
-            version_id,
-            kind,
-            enqueued_at_ms,
-            attempts,
-        },
-        record_end,
-    ))
-}
-
-/// Decode a whole journal, stopping at the first torn or corrupt record.
-/// Returns the decoded intents and the number of trailing bytes discarded.
-pub(crate) fn decode_journal(data: &[u8]) -> (Vec<MrfIntent>, usize) {
-    let mut intents = Vec::new();
-    let mut cursor = 0usize;
-    while cursor < data.len() {
-        match decode_one(&data[cursor..]) {
-            Some((intent, consumed)) => {
-                intents.push(intent);
-                cursor += consumed;
-            }
-            None => break,
-        }
-    }
-    let truncated = data.len() - cursor;
-    (intents, truncated)
-}
-
-// ---------------------------------------------------------------------------
-// Journal disk IO (all local disks, first successful read wins)
-// ---------------------------------------------------------------------------
-
-async fn journal_disks() -> Vec<DiskStore> {
-    let map = local_disk_map_read().await;
-    map.values().flatten().cloned().collect()
-}
-
-async fn read_journal() -> Option<Vec<u8>> {
-    for disk in journal_disks().await {
-        match disk.read_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH).await {
-            Ok(bytes) => return Some(bytes.to_vec()),
-            Err(_) => continue,
-        }
-    }
-    None
-}
-
-async fn write_journal(data: &[u8]) {
-    let payload = bytes::Bytes::copy_from_slice(data);
-    for disk in journal_disks().await {
-        if let Err(err) = disk
-            .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone())
-            .await
-        {
-            warn_mrf_journal_write(&err);
-        }
-    }
-    if !data.is_empty() {
-        counter!("rustfs_heal_mrf_journal_fsync_total").increment(1);
-    }
-    gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64);
-}
-
-async fn delete_journal() {
-    for disk in journal_disks().await {
-        let _ = disk
-            .delete(
-                super::RUSTFS_META_BUCKET,
-                MRF_JOURNAL_PATH,
-                crate::heal::storage_api::owner::EcstoreDeleteOptions::default(),
-            )
-            .await;
-    }
-}
-
-fn warn_mrf_journal_write(err: &super::DiskError) {
-    tracing::warn!(
-        target: "rustfs::heal::mrf",
-        error = %err,
-        "MRF journal write failed; unconsumed intents may be lost on restart"
-    );
-}
-
-// ---------------------------------------------------------------------------
-// Consumer
-// ---------------------------------------------------------------------------
-
-/// Translate an intent into the prioritized heal request the issue specifies:
-/// decode failures go Urgent ECDecode, metadata corruption goes High
-/// Metadata, partial writes go Normal object heal.
-pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest {
-    let bucket = intent.bucket.to_string();
-    let object = intent.object.to_string();
-    let version_id = intent.version_id.map(|bytes| Uuid::from_bytes(bytes).to_string());
-    let (heal_type, priority) = match intent.kind {
-        rustfs_common::mrf_channel::MrfKind::DecodeFailure => (
-            HealType::ECDecode {
-                bucket,
-                object,
-                version_id,
-            },
-            HealPriority::Urgent,
-        ),
-        rustfs_common::mrf_channel::MrfKind::MetadataCorruption => (HealType::Metadata { bucket, object }, HealPriority::High),
-        rustfs_common::mrf_channel::MrfKind::PartialWrite => (
-            HealType::Object {
-                bucket,
-                object,
-                version_id,
-            },
-            HealPriority::Normal,
-        ),
-    };
-    let mut request = HealRequest::new(heal_type, HealOptions::default(), priority);
-    request.source = rustfs_common::heal_channel::HealRequestSource::Mrf;
-    request
-}
-
-struct MrfRuntime {
-    queue: MrfQueue,
-    config: MrfConsumerConfig,
-    new_since_flush: usize,
-    /// True while a journal snapshot exists on disk that no longer reflects
-    /// an all-consumed pending set; the next idle tick removes it (MinIO
-    /// deletes its `list.bin` after replay for the same reason).
-    journal_on_disk: bool,
-    /// Earliest instant a full-admission retry may proceed.
-    backoff_until: Option<tokio::time::Instant>,
-}
-
-impl MrfRuntime {
-    fn record_accept(&mut self) {
-        // Accepted intents leave the pending set; the next flush persists the
-        // smaller snapshot, which is the journal's compaction.
-    }
-
-    fn snapshot(&self) -> Vec<u8> {
-        let mut buf = Vec::new();
-        for intent in self.queue.intents() {
-            encode_intent(intent, &mut buf);
-        }
-        buf
-    }
-
-    async fn flush(&mut self) {
-        write_journal(&self.snapshot()).await;
-        self.new_since_flush = 0;
-        self.journal_on_disk = true;
-    }
-
-    /// Drain pending intents into the heal manager until it is full, the
-    /// queue empties, or attempts are exhausted.
-    async fn dispatch(&mut self, manager: &HealManager) {
-        if let Some(until) = self.backoff_until {
-            if tokio::time::Instant::now() < until {
-                return;
-            }
-            self.backoff_until = None;
-        }
-        while let Some(mut intent) = self.queue.pop_front() {
-            let request = build_heal_request(&intent);
-            match manager.submit_heal_request(request).await {
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => self.record_accept(),
-                Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
-                    intent.attempts = intent.attempts.saturating_add(1);
-                    if intent.attempts >= MRF_MAX_ATTEMPTS {
-                        counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1);
-                        continue;
-                    }
-                    self.queue.push_back(intent);
-                    self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff);
-                    break;
-                }
-                Ok(HealAdmissionResult::Dropped(_)) => {
-                    counter!("rustfs_heal_mrf_dropped_total", "reason" => "admission_policy").increment(1);
-                }
-                Err(_) => {
-                    intent.attempts = intent.attempts.saturating_add(1);
-                    if intent.attempts >= MRF_MAX_ATTEMPTS {
-                        counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1);
-                        continue;
-                    }
-                    self.queue.push_back(intent);
-                    self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff);
-                    break;
-                }
-            }
-        }
-        gauge!("rustfs_heal_mrf_queue_depth").set(self.queue.depth() as f64);
-        gauge!("rustfs_heal_mrf_queue_bytes").set(self.queue.bytes() as f64);
-    }
-}
-
-/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and
-/// spawn the consumer task. Called once from the heal runtime bootstrap right
-/// after the manager started; a disabled feature or a double call is a no-op.
-/// Public for integration tests that drive the real consumer loop.
-pub fn spawn_mrf_consumer(manager: Arc<HealManager>) {
-    let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE);
-    rustfs_common::mrf_channel::set_mrf_delivery_enabled(enabled);
-    if !enabled {
-        tracing::info!(
-            target: "rustfs::heal::mrf",
-            "MRF intent pipeline disabled by configuration; producers will not deliver"
-        );
-        return;
-    }
-    let receiver = match rustfs_common::mrf_channel::init_mrf_channel() {
-        Ok(receiver) => receiver,
-        Err(err) => {
-            tracing::warn!(
-                target: "rustfs::heal::mrf",
-                error = err,
-                "MRF channel initialization failed; intents will be dropped at producers"
-            );
-            return;
-        }
-    };
-    tokio::spawn(async move {
-        run_mrf_consumer(manager, receiver).await;
-    });
-    tracing::info!(target: "rustfs::heal::mrf", "MRF intent consumer started");
-}
-
-/// Replay the durable journal into a fresh pending queue and submit whatever
-/// it armed. Returns the number of intact intents replayed. Duplicates are
-/// merged by the manager's dedup key; the journal file is removed once read
-/// (torn tails truncate via the per-record CRC). Public for integration tests;
-/// the live consumer invokes this through [`replay_into`] at startup.
-pub async fn replay_journal_once(manager: &Arc<HealManager>) -> usize {
-    let config = MrfConsumerConfig::default();
-    let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
-    let mut backoff_until: Option<tokio::time::Instant> = None;
-    replay_into(manager, &mut queue, &mut backoff_until).await
-}
-
-/// Shared replay core: read + decode + re-arm + delete, then drain what fits.
-async fn replay_into(
-    manager: &Arc<HealManager>,
-    queue: &mut MrfQueue,
-    backoff_until: &mut Option<tokio::time::Instant>,
-) -> usize {
-    let Some(data) = read_journal().await else {
-        return 0;
-    };
-    let (intents, truncated) = decode_journal(&data);
-    if truncated > 0 {
-        tracing::warn!(
-            target: "rustfs::heal::mrf",
-            truncated_bytes = truncated,
-            "MRF journal had a torn tail; truncated records were discarded"
-        );
-    }
-    counter!("rustfs_heal_mrf_replayed_total").increment(intents.len() as u64);
-    let replayed = intents.len();
-    for intent in intents {
-        queue.try_push(intent);
-    }
-    delete_journal().await;
-
-    // Drain the replayed intents immediately; whatever the manager refuses
-    // stays armed in `queue` for the consumer's retry loop.
-    if backoff_until.is_none() {
-        while let Some(mut intent) = queue.pop_front() {
-            let request = build_heal_request(&intent);
-            match manager.submit_heal_request(request).await {
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
-                Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
-                    intent.attempts = intent.attempts.saturating_add(1);
-                    if intent.attempts < MRF_MAX_ATTEMPTS {
-                        queue.push_back(intent);
-                        *backoff_until = Some(tokio::time::Instant::now());
-                    }
-                    break;
-                }
-                Ok(HealAdmissionResult::Dropped(_)) | Err(_) => {}
-            }
-        }
-    }
-    replayed
-}
-
-/// Replay the journal, then keep draining the channel into the heal manager
-/// while persisting the pending snapshot.
-async fn run_mrf_consumer(manager: Arc<HealManager>, mut receiver: mpsc::Receiver<MrfIntent>) {
-    let config = MrfConsumerConfig::default();
-    let mut runtime = MrfRuntime {
-        queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes),
-        config: config.clone(),
-        new_since_flush: 0,
-        journal_on_disk: false,
-        backoff_until: None,
-    };
-
-    // Replay: read the journal, re-arm intents (duplicates are merged by the
-    // manager's dedup key), then drop the file so the next flush starts clean.
-    replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await;
-
-    let mut flush_tick = tokio::time::interval(runtime.config.flush_interval);
-    flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
-    let mut batch: Vec<MrfIntent> = Vec::with_capacity(runtime.config.replay_batch);
-
-    loop {
-        tokio::select! {
-            received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => {
-                if received == 0 {
-                    // Channel closed: flush once more and stop.
-                    runtime.flush().await;
-                    tracing::info!(
-                        target: "rustfs::heal::mrf",
-                        "MRF channel closed; consumer stopped after final flush"
-                    );
-                    return;
-                }
-                for intent in batch.drain(..) {
-                    runtime.queue.try_push(intent);
-                    runtime.new_since_flush += 1;
-                }
-                runtime.dispatch(manager.as_ref()).await;
-                if runtime.new_since_flush >= runtime.config.flush_threshold {
-                    runtime.flush().await;
-                }
-            }
-            _ = flush_tick.tick() => {
-                if runtime.new_since_flush > 0 || runtime.queue.depth() > 0 {
-                    runtime.flush().await;
-                    runtime.dispatch(manager.as_ref()).await;
-                } else if runtime.journal_on_disk {
-                    // All intents consumed: remove the journal so a restart
-                    // replays nothing (mirrors MinIO's post-replay unlink).
-                    delete_journal().await;
-                    runtime.journal_on_disk = false;
-                    gauge!("rustfs_heal_mrf_journal_bytes").set(0.0);
-                }
-                gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64);
-            }
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use rustfs_common::mrf_channel::{MrfIntent, MrfKind};
-    use std::sync::Arc as StdArc;
-
-    fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent {
-        MrfIntent {
-            bucket: StdArc::from(bucket),
-            object: StdArc::from(object),
-            version_id: Some([7u8; 16]),
-            kind: MrfKind::DecodeFailure,
-            enqueued_at_ms: 1_700_000_000_000,
-            attempts,
-        }
-    }
-
-    #[test]
-    fn queue_enforces_count_and_byte_ceilings() {
-        let mut queue = MrfQueue::new(2, usize::MAX);
-        assert!(queue.try_push(intent("b", "o", 0)));
-        assert!(queue.try_push(intent("b", "o", 0)));
-        assert!(!queue.try_push(intent("b", "o", 0)), "count ceiling must drop");
-
-        let mut tiny = MrfQueue::new(usize::MAX, intent("bucket", "object", 0).estimated_bytes());
-        assert!(tiny.try_push(intent("bucket", "object", 0)));
-        assert!(
-            !tiny.try_push(intent("bucket", "object", 0)),
-            "byte budget must drop before the second intent fits"
-        );
-    }
-
-    #[test]
-    fn journal_roundtrip_preserves_intents() {
-        let intents = vec![
-            intent("bucket-a", "object/a", 0),
-            intent("bucket-b", "object/b", 2),
-            MrfIntent {
-                bucket: StdArc::from("bucket-c"),
-                object: StdArc::from("object/c"),
-                version_id: None,
-                kind: MrfKind::MetadataCorruption,
-                enqueued_at_ms: 5,
-                attempts: 1,
-            },
-        ];
-        let mut buf = Vec::new();
-        for intent in &intents {
-            encode_intent(intent, &mut buf);
-        }
-        let (decoded, truncated) = decode_journal(&buf);
-        assert_eq!(truncated, 0);
-        assert_eq!(decoded.len(), intents.len());
-        for (left, right) in decoded.iter().zip(intents.iter()) {
-            assert_eq!(left.bucket, right.bucket);
-            assert_eq!(left.object, right.object);
-            assert_eq!(left.version_id, right.version_id);
-            assert_eq!(left.kind, right.kind);
-            assert_eq!(left.attempts, right.attempts);
-        }
-    }
-
-    #[test]
-    fn journal_torn_tail_is_truncated() {
-        let mut buf = Vec::new();
-        encode_intent(&intent("b", "o", 0), &mut buf);
-        let mut torn = buf.clone();
-        torn.extend_from_slice(&buf[..buf.len() / 2]);
-
-        let (decoded, truncated) = decode_journal(&torn);
-        assert_eq!(decoded.len(), 1, "the intact record must survive");
-        assert!(truncated > 0, "the partial tail must be discarded");
-
-        // A corrupted body (CRC mismatch) also truncates from that record on.
-        let mut corrupt = buf.clone();
-        let mid = MRF_RECORD_FIXED_HEAD + 4;
-        corrupt[mid] ^= 0xff;
-        let (decoded, truncated) = decode_journal(&corrupt);
-        assert!(decoded.is_empty());
-        assert_eq!(truncated, corrupt.len());
-    }
-
-    #[test]
-    fn heal_request_mapping_follows_priority_matrix() {
-        let decode = build_heal_request(&intent("b", "o", 0));
-        assert!(matches!(decode.heal_type, HealType::ECDecode { .. }));
-        assert_eq!(decode.priority, HealPriority::Urgent);
-
-        let metadata = build_heal_request(&MrfIntent {
-            bucket: StdArc::from("b"),
-            object: StdArc::from("o"),
-            version_id: None,
-            kind: MrfKind::MetadataCorruption,
-            enqueued_at_ms: 0,
-            attempts: 0,
-        });
-        assert!(matches!(metadata.heal_type, HealType::Metadata { .. }));
-        assert_eq!(metadata.priority, HealPriority::High);
-
-        let partial = build_heal_request(&MrfIntent {
-            bucket: StdArc::from("b"),
-            object: StdArc::from("o"),
-            version_id: None,
-            kind: MrfKind::PartialWrite,
-            enqueued_at_ms: 0,
-            attempts: 0,
-        });
-        assert!(matches!(partial.heal_type, HealType::Object { .. }));
-        assert_eq!(partial.priority, HealPriority::Normal);
-    }
-}
@@ -158,10 +158,6 @@ pub async fn init_heal_manager_with_workload_provider(
            return Err(err);
        }

-        // Start the MRF intent consumer (error-path repair intents + durable
-        // journal replay) now that the manager can accept submissions.
-        heal::mrf_queue::spawn_mrf_consumer(heal_manager.clone());
-
        #[cfg(test)]
        test_hook_after_manager_start().await;

@@ -1,189 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-
-//! HS-01 (rustfs/backlog#1865): MRF intent pipeline integration tests.
-//!
-//! Drives the real consumer loop (`spawn_mrf_consumer`) against a real
-//! 4-disk `ECStore` heal storage and a `HealManager` that has not started its
-//! scheduler, so submitted intents stay observable in the admission queue.
-//! Under `cargo nextest` each test runs in its own process, which keeps the
-//! process-global MRF channel singleton safe.
-
-use rustfs_common::mrf_channel::{self, MrfKind};
-use rustfs_heal::heal::{
-    manager::{HealConfig, HealManager},
-    mrf_queue,
-    storage::{ECStoreHealStorage, HealStorageAPI},
-};
-use serial_test::serial;
-use std::{path::Path, sync::Arc, time::Duration};
-
-mod storage_api;
-
-use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, init_local_disks};
-
-const META_BUCKET: &str = ".rustfs.sys";
-const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin";
-
-async fn heal_env() -> (Vec<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
-    let env = rustfs_test_utils::TestECStoreEnv::builder()
-        .prefix("rustfs_heal_mrf_test")
-        .build()
-        .await;
-    let heal_storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
-    (env.disk_paths, heal_storage)
-}
-
-fn make_manager(storage: Arc<dyn HealStorageAPI>) -> Arc<HealManager> {
-    Arc::new(HealManager::new(
-        storage,
-        Some(HealConfig {
-            // Keep the scheduler from draining the queue before assertions.
-            heal_interval: Duration::from_secs(3600),
-            enable_auto_heal: false,
-            ..Default::default()
-        }),
-    ))
-}
-
-/// Encode one journal record independently of the implementation, so a format
-/// drift between writer and this fixture fails loudly here.
-fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec<u8> {
-    let mut body = vec![1u8, 1, kind, attempts];
-    body.extend_from_slice(&1_700_000_000_000u64.to_le_bytes());
-    match version {
-        Some(bytes) => {
-            body.push(1);
-            body.extend_from_slice(&bytes);
-        }
-        None => body.push(0),
-    }
-    body.extend_from_slice(&(bucket.len() as u32).to_le_bytes());
-    body.extend_from_slice(&(object.len() as u32).to_le_bytes());
-    body.extend_from_slice(bucket.as_bytes());
-    body.extend_from_slice(object.as_bytes());
-    let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc);
-    hasher.update(&body);
-    body.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes());
-    body
-}
-
-fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) {
-    for path in disk_paths {
-        let journal = path.join(META_BUCKET).join(JOURNAL_REL);
-        std::fs::create_dir_all(journal.parent().expect("journal parent")).expect("create journal dir");
-        std::fs::write(&journal, data).expect("write journal fixture");
-    }
-}
-
-async fn wait_until<F, Fut>(deadline: Duration, mut probe: F) -> bool
-where
-    F: FnMut() -> Fut,
-    Fut: std::future::Future<Output = bool>,
-{
-    let start = std::time::Instant::now();
-    while start.elapsed() < deadline {
-        if probe().await {
-            return true;
-        }
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-    false
-}
-
-/// A decode-failure intent delivered on the global channel must surface in the
-/// heal manager as an Urgent request attributed to the MRF source.
-#[tokio::test]
-#[serial]
-async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
-    let (_disk_paths, storage) = heal_env().await;
-    let manager = make_manager(storage);
-
-    mrf_queue::spawn_mrf_consumer(manager.clone());
-
-    assert!(
-        mrf_channel::try_send_mrf_intent(MrfKind::DecodeFailure, "mrf-bucket", "mrf-object", None),
-        "intent should be accepted while the consumer holds the channel"
-    );
-
-    let appeared = wait_until(Duration::from_secs(10), || async {
-        let snapshot = manager.operations_snapshot().await;
-        snapshot.queued_by_source.mrf >= 1 && snapshot.queued_by_priority.urgent >= 1
-    })
-    .await;
-    assert!(
-        appeared,
-        "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})",
-        manager.operations_snapshot().await
-    );
-}
-
-/// A journal left behind by a previous process must be replayed into the
-/// manager queue and then removed, and a torn tail must not block replay of
-/// the intact records.
-#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
-#[serial]
-async fn journal_replay_arms_intents_and_deletes_the_file() {
-    let (disk_paths, storage) = heal_env().await;
-
-    // The journal reader resolves disks through the process-local disk map;
-    // register the environment's disks the same way server startup does.
-    let mut endpoints: Vec<Endpoint> = disk_paths
-        .iter()
-        .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path"))
-        .collect();
-    for (i, endpoint) in endpoints.iter_mut().enumerate() {
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(i);
-    }
-    let pool = PoolEndpoints {
-        legacy: false,
-        set_count: 1,
-        drives_per_set: endpoints.len(),
-        endpoints: Endpoints::from(endpoints),
-        cmd_line: "mrf-test".to_string(),
-        platform: String::new(),
-    };
-    init_local_disks(EndpointServerPools::from(vec![pool]))
-        .await
-        .expect("local disks should register");
-
-    let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0);
-    journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1));
-    // Torn tail: a third record truncated mid-way must not block the two
-    // intact records above.
-    journal.extend_from_slice(&journal_record(2, "replay-bucket", "metadata-object", None, 0)[..8]);
-    write_journal_to_disks(&disk_paths, &journal);
-
-    let manager = make_manager(storage);
-    // Replay directly (not via the process-global channel consumer, which the
-    // sibling test already claimed in this process under plain `cargo test`).
-    let replayed = mrf_queue::replay_journal_once(&manager).await;
-    assert_eq!(replayed, 2, "the two intact records must be replayed");
-
-    let snapshot = manager.operations_snapshot().await;
-    assert_eq!(snapshot.queued_by_source.mrf, 2, "replayed intents must be attributed to the MRF source");
-
-    assert!(
-        disk_paths
-            .iter()
-            .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()),
-        "the journal file must be removed after a successful replay"
-    );
-
-    let snapshot = manager.operations_snapshot().await;
-    assert_eq!(snapshot.queued_by_priority.urgent, 1, "the decode-failure record must replay as Urgent");
-    assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal");
-}
@@ -26,6 +26,7 @@ pub enum StorageMedia {
 }

 impl StorageMedia {
+    #[allow(dead_code)]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Nvme => "nvme",
@@ -59,6 +60,7 @@ pub enum AccessPattern {
 }

 impl AccessPattern {
+    #[allow(dead_code)]
    pub fn as_str(&self) -> &'static str {
        match self {
            Self::Sequential => "sequential",
@@ -69,21 +71,25 @@ impl AccessPattern {
    }

    /// Check if this is a sequential access pattern.
+    #[allow(dead_code)]
    pub fn is_sequential(&self) -> bool {
        matches!(self, Self::Sequential)
    }

    /// Check if this is a random access pattern.
+    #[allow(dead_code)]
    pub fn is_random(&self) -> bool {
        matches!(self, Self::Random)
    }

    /// Check if this is a mixed access pattern.
+    #[allow(dead_code)]
    pub fn is_mixed(&self) -> bool {
        matches!(self, Self::Mixed)
    }

    /// Check if this pattern is unknown.
+    #[allow(dead_code)]
    pub fn is_unknown(&self) -> bool {
        matches!(self, Self::Unknown)
    }
@@ -427,6 +427,7 @@ pub enum DataSource {
    /// Write triggered
    WriteTriggered,
    /// Fallback value
+    #[allow(dead_code)]
    Fallback,
 }

@@ -602,6 +603,7 @@ impl WriteRecord {

 /// Hybrid strategy configuration
 #[derive(Debug, Clone)]
+#[allow(dead_code)]
 pub struct HybridStrategyConfig {
    /// Scheduled update interval
    pub scheduled_update_interval: Duration,
@@ -996,12 +998,14 @@ impl HybridCapacityManager {
    }

    /// Get cache age
+    #[allow(dead_code)]
    pub async fn get_cache_age(&self) -> Option<Duration> {
        let cache = self.cache.read().await;
        cache.as_ref().map(|c| c.last_update.elapsed())
    }

    /// Get write frequency (writes/minute)
+    #[allow(dead_code)]
    pub async fn get_write_frequency(&self) -> usize {
        let record = &self.write_record;
        record.recent_write_count(record.monotonic_second())
@@ -1296,6 +1300,7 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
 ///     .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
 ///     .await;
 /// ```
+#[allow(dead_code)]
 pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc<HybridCapacityManager> {
    Arc::new(HybridCapacityManager::new(config))
 }
@@ -49,6 +49,7 @@ pub struct IndexInfo {
    pub uncompressed_offset: i64,
 }

+#[allow(dead_code)]
 impl Index {
    pub fn new() -> Self {
        Self {
@@ -59,6 +60,14 @@ impl Index {
        }
    }

+    #[allow(dead_code)]
+    fn reset(&mut self, max_block: usize) {
+        self.est_block_uncomp = max_block as i64;
+        self.total_compressed = -1;
+        self.total_uncompressed = -1;
+        self.info.clear();
+    }
+
    pub fn len(&self) -> usize {
        self.info.len()
    }
@@ -502,6 +511,47 @@ fn read_varint(buf: &[u8]) -> io::Result<(i64, usize)> {
    Err(io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF"))
 }

+// Helper functions for index header manipulation
+#[allow(dead_code)]
+pub fn remove_index_headers(b: &[u8]) -> Option<&[u8]> {
+    if b.len() < 4 + S2_INDEX_TRAILER.len() {
+        return None;
+    }
+
+    // Skip size
+    let b = &b[4..];
+
+    // Check trailer
+    if !b.starts_with(S2_INDEX_TRAILER) {
+        return None;
+    }
+
+    Some(&b[S2_INDEX_TRAILER.len()..])
+}
+
+#[allow(dead_code)]
+pub fn restore_index_headers(in_data: &[u8]) -> Vec<u8> {
+    if in_data.is_empty() {
+        return Vec::new();
+    }
+
+    let mut b = Vec::with_capacity(4 + S2_INDEX_HEADER.len() + in_data.len() + S2_INDEX_TRAILER.len() + 4);
+    b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]);
+    b.extend_from_slice(S2_INDEX_HEADER);
+    b.extend_from_slice(in_data);
+
+    let total_size = (b.len() + 4 + S2_INDEX_TRAILER.len()) as u32;
+    b.extend_from_slice(&total_size.to_le_bytes());
+    b.extend_from_slice(S2_INDEX_TRAILER);
+
+    let chunk_len = b.len() - 4;
+    b[1] = chunk_len as u8;
+    b[2] = (chunk_len >> 8) as u8;
+    b[3] = (chunk_len >> 16) as u8;
+
+    b
+}
+
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -6,8 +6,6 @@ admission, scanner-driven heal/bitrot checks, and namespace alerts.

 For operator-facing runtime controls, status fields, and tuning workflows, see
 [Scanner Runtime Controls](../../docs/operations/scanner-runtime-controls.md).
-For a MinIO `data-scanner` comparison and the improvement contracts, see
-[MinIO Scanner Compatibility](../../docs/architecture/minio-scanner-compat.md).
 For repeatable scanner-pressure validation, see
 [Scanner Benchmark Runbook](../../docs/operations/scanner-benchmark-runbook.md).

@@ -4,8 +4,6 @@ RustFS Scanner 是后台维护扫描循环，负责用量统计、生命周期

 面向运维人员的运行时控制项、状态字段和调参流程，请参考
 [Scanner Runtime Controls](../../docs/operations/scanner-runtime-controls.md)。
-与 MinIO `data-scanner` 的对照和补齐约定请参考
-[MinIO Scanner Compatibility](../../docs/architecture/minio-scanner-compat.md)。
 可复现的 scanner 压力验证流程请参考
 [Scanner Benchmark Runbook](../../docs/operations/scanner-benchmark-runbook.md)。

@@ -2478,15 +2478,6 @@ impl FolderScanner {
                        }

                        if let GetSizeFailureAction::HealMetadata { object } = failure_action {
-                            // MRF journal intent: durable High-priority Metadata
-                            // heal across restarts (HS-01); the scanner heal
-                            // request below stays as the immediate path.
-                            rustfs_common::mrf_channel::try_send_mrf_intent(
-                                rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
-                                &item.bucket,
-                                &object,
-                                None,
-                            );
                            self.send_required_scanner_heal_request(
                                PendingScannerHealKind::Object,
                                item.bucket.clone(),
@@ -206,6 +206,7 @@ async fn setup_isolated_test_env(init_expiry: bool) -> (Vec<PathBuf>, Arc<ECStor
 }

 /// Test helper: Create a test bucket
+#[allow(dead_code)]
 async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
    (**ecstore)
        .make_bucket(bucket_name, &Default::default())
@@ -250,6 +251,7 @@ async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOpti
 }

 /// Test helper: Set bucket lifecycle configuration
+#[allow(dead_code)]
 async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
    let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -272,6 +274,7 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::erro
 }

 /// Test helper: Set bucket lifecycle configuration
+#[allow(dead_code)]
 async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
    // Create lifecycle rule that targets delete-marker cleanup only.
    // Keep Expiration.Days unset to avoid expiring live transitioned object versions.
@@ -294,6 +297,7 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<
    Ok(())
 }

+#[allow(dead_code)]
 async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box<dyn std::error::Error>> {
    let lifecycle_xml = format!(
        r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -316,6 +320,7 @@ async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64)
    Ok(())
 }

+#[allow(dead_code)]
 async fn set_bucket_lifecycle_transition_with_tier(
    bucket_name: &str,
    storage_class: &str,
@@ -363,6 +368,7 @@ async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bo
 }

 /// Test helper: Check if object exists
+#[allow(dead_code)]
 async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
    if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
        println!("oi: {oi:?}");
@@ -373,6 +379,7 @@ async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &
    }
 }

+#[allow(dead_code)]
 async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
    let deadline = tokio::time::Instant::now() + timeout;

@@ -428,6 +428,7 @@ pub fn parse_url(s: &str) -> Result<ParsedURL, NetError> {
    Ok(ParsedURL(uu))
 }

+#[allow(dead_code)]
 pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
    let u = parse_url(s)?;
    match u.0.scheme() {
@@ -436,6 +437,7 @@ pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
    }
 }

+#[allow(dead_code)]
 pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> bool {
    if err.kind() == std::io::ErrorKind::TimedOut {
        return !expect_timeouts;
@@ -447,10 +449,12 @@ pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> b
        || err_str.contains("use of closed network connection")
 }

+#[allow(dead_code)]
 pub fn is_conn_reset_err(err: &std::io::Error) -> bool {
    err.to_string().contains("connection reset by peer") || matches!(err.raw_os_error(), Some(libc::ECONNRESET))
 }

+#[allow(dead_code)]
 pub fn is_conn_refused_err(err: &std::io::Error) -> bool {
    err.to_string().contains("connection refused") || matches!(err.raw_os_error(), Some(libc::ECONNREFUSED))
 }
@@ -32,8 +32,8 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use hyper_rustls::ConfigBuilderExt;
 use rumqttc::{
-    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, ProtocolViolation,
-    PublishNoticeError, PublishOptions, QoS, Transport, mqttbytes::Error as MqttBytesError,
+    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS,
+    Transport, mqttbytes::Error as MqttBytesError,
 };
 use rustfs_config::{
    EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
@@ -791,7 +791,7 @@ where
                .as_ref()
                .ok_or_else(|| TargetError::Configuration("MQTT client not initialized".to_string()))?;
            let notice = client
-                .publish_tracked(&self.args.topic, body, PublishOptions::new(self.args.qos))
+                .publish_tracked(&self.args.topic, self.args.qos, false, body)
                .await
                .map_err(|error| classify_mqtt_client_error(&error))?;
            drop(client_guard);
@@ -1145,7 +1145,7 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
                        );
                        connected_status.store(false, Ordering::SeqCst);
                    }
-                    rumqttc::Event::Incoming(Incoming::PingResp) => {
+                    rumqttc::Event::Incoming(Incoming::PingResp(_)) => {
                        trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive.");
                    }
                    rumqttc::Event::Incoming(Incoming::SubAck(suback)) => {
@@ -1257,11 +1257,7 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
 /// copy is preserved and replayed rather than dropped (backlog#971).
 fn classify_mqtt_client_error(err: &ClientError) -> TargetError {
    match err {
-        ClientError::RequestChannelFull(_) | ClientError::RequestChannelDisconnected(_) | ClientError::TrackingUnavailable => {
-            TargetError::NotConnected
-        }
-        ClientError::InvalidRequest(_) => TargetError::Request(format!("Invalid MQTT publish request: {err}")),
-        _ => TargetError::NotConnected,
+        ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected,
    }
 }

@@ -1274,14 +1270,10 @@ fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError {
        PublishNoticeError::Recv
        | PublishNoticeError::SessionReset
        | PublishNoticeError::Qos0NotFlushed
-        | PublishNoticeError::BrokerOnlySessionResume
-        | PublishNoticeError::SessionPersistence(_)
        | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected,
-        PublishNoticeError::RetainNotSupported => TargetError::Request(format!("MQTT broker rejected publish: {err}")),
        PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => {
            TargetError::Request(format!("MQTT broker rejected publish: {err}"))
        }
-        _ => TargetError::NotConnected,
    }
 }

@@ -1307,13 +1299,12 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
                        | MqttBytesError::MalformedPacket // Package format error
                        | MqttBytesError::PayloadTooLong // Too long load
                        | MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded
-                        | MqttBytesError::TopicNotUtf8 { .. } // Topic Non-UTF-8 (Serious Agreement Violation)
+                        | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
                    )
                }
                // Others that are fatal StateError variants
                rumqttc::StateError::InvalidState          // The internal state machine is in invalid state
-                | rumqttc::StateError::ProtocolViolation(ProtocolViolation::UnexpectedIncomingPacket(_)) // Agreement Violation: Unexpected Data Packet Received
-                | rumqttc::StateError::ProtocolViolation(_) // Agreement Violation
+                | rumqttc::StateError::WrongPacket         // Agreement Violation: Unexpected Data Packet Received
                | rumqttc::StateError::Unsolicited(_)      // Agreement Violation: Unsolicited ACK Received
                | rumqttc::StateError::CollisionTimeout    // Agreement Violation (if this stage occurs)
                | rumqttc::StateError::EmptySubscription   // Agreement violation (if this stage occurs)
@@ -1736,8 +1727,8 @@ where
 mod tests {
    use super::{
        AsyncClient, ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTarget, MQTTTlsConfig,
-        MqttOptions, PublishNoticeError, PublishOptions, QoS, QueuedPayloadMeta, classify_mqtt_client_error,
-        classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
+        MqttOptions, PublishNoticeError, QoS, QueuedPayloadMeta, classify_mqtt_client_error, classify_mqtt_notice_error,
+        next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
    };
    use crate::error::TargetError;
    use crate::target::{REDACTED_SECRET, TargetType};
@@ -1803,7 +1794,7 @@ mod tests {
            .capacity(1)
            .build();
        client
-            .publish("fill", b"fill".as_slice(), PublishOptions::new(QoS::AtLeastOnce))
+            .publish("fill", QoS::AtLeastOnce, false, b"fill".as_slice())
            .await
            .expect("first publish should fill the local channel");
        *target.client.lock().await = Some(client);
@@ -46,7 +46,6 @@ Two rules keep this directory healthy:
 - [s3-tables-support-matrix.md](s3-tables-support-matrix.md)
 - [minio-rustfs-router-compatibility.md](minio-rustfs-router-compatibility.md)
 - [minio-file-format-compat.md](minio-file-format-compat.md)
- [minio-scanner-compat.md](minio-scanner-compat.md) — MinIO data-scanner parity: cycle/heal/ILM/usage gaps and improvement contracts

 ## Inventories & baselines (snapshots that feed migration work)

@@ -105,8 +105,9 @@ inventory. Generic function-local names such as `CACHE`, `LOCK`, `INIT`, and
 | `USE_STARSHARD_CACHE`, `BUCKET_CACHE_SMALL`, `BUCKET_CACHE_LARGE` | `rustfs/src/storage/ecfs_extend.rs` | Cache or constant / owner-local cache | Bucket validation cache backend selection and cache storage stay private to the ECFS extension owner. |
 | `GLOBAL_SSE_DEK_PROVIDER`, `SSE_TEST_LOCK` | `rustfs/src/storage/sse.rs` | Owner-local cache / test state | SSE DEK provider cache and test serialization lock stay private to the SSE owner. |
 | `AUTH_FS` | `rustfs/src/storage/access.rs` | Cache or constant / owner-local cache | Authorization tag-condition lookup keeps its filesystem helper private to the access owner. |
+| `LOCK_STATS` | `rustfs/src/storage/lock_optimizer.rs` | Process-global owner-local metrics | Lock optimization statistics stay private behind lock optimizer helper APIs. |
 | `DEADLOCK_DETECTOR` | `rustfs/src/storage/deadlock_detector.rs` | Process-global owner-local state | Deadlock detector lifecycle state stays private to the storage deadlock detector owner. |
-| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager and request counters remain inside the storage concurrency owner boundary. |
+| `CONCURRENCY_MANAGER`, `ACTIVE_GET_REQUESTS`, `ACTIVE_PUT_REQUESTS`, `IO_PRIORITY_METRICS` | `rustfs/src/storage/concurrency/*` | Process-global owner-local scheduler state | Storage concurrency manager, counters, and metrics remain inside the storage concurrency owner boundary. |
 | `GET_OBJECT_BUFFER_THRESHOLD_WARNED`, `GET_READER_STREAM_BUFFER_SIZE_OVERRIDE`, function-local `ENABLED`, `OBJECT_SEEK_SUPPORT_THRESHOLD`, `OBJECT_SEEK_SUPPORT_CONCURRENCY_THRESHOLDS` | `rustfs/src/app/object_usecase.rs` | Cache or constant / owner-local cache | Object GET/seek tuning caches and warning guards stay private to object usecase helpers. |
 | `SUPPORTED_HEADERS` | `rustfs/src/storage/options.rs` | Cache or constant / owner-local constant | Supported-header lookup state stays private to storage option parsing. |
 | `AUDIT_TARGET_SPECS`, `NOTIFICATION_TARGET_SPECS` | `rustfs/src/admin/handlers/audit.rs`, `rustfs/src/admin/handlers/event.rs`, `rustfs/src/admin/handlers/plugins_instances.rs` | Cache or constant / owner-local constant | Admin target descriptor tables stay private to their handler owners. |
@@ -1,464 +0,0 @@
-# MinIO ↔ RustFS Data Scanner Comparison
-
-Assesses how closely the RustFS background data scanner matches MinIO's
-`cmd/data-scanner.go` implementation: cycle leadership, namespace walk,
-usage accounting, ILM admission, replication repair, heal/bitrot selection,
-alerts, and operator surfaces. This is a **durable gap analysis**. It changes
-no scanner code. Every claim cites the code that backs it.
-
-MinIO sources below are the public `minio/minio` `master` tree as of
-2026-08-18 (`cmd/data-scanner.go`, `cmd/erasure.go`, `cmd/xl-storage.go`,
-`cmd/data-usage-cache.go`, `internal/config/scanner/scanner.go`). They are
-not files in this repository.
-
-Operator runtime knobs already documented here stay in
-[scanner-runtime-controls.md](../operations/scanner-runtime-controls.md).
-This page does not duplicate that runbook.
-
-## Executive Summary
-
-The two scanners share the same skeleton. Both run one cluster-wide leader
-loop, persist a cycle counter in `.bloomcycle.bin`, walk folders with a
-1-in-16 compacted-leaf schedule, select objects for heal with a 1-in-1024
-hash, compact usage trees at the same child thresholds, evaluate ILM through
-a lifecycle evaluator, enqueue replication heals, emit excess-version and
-excess-folder alerts, and throttle with a proportional sleeper.
-
-The remaining gaps are not "the scanner is missing". They are **heal
-fidelity**, **cross-set scheduling**, **on-disk cache interop**, and
-**operator/notification wire names**. The highest-severity difference is
-that MinIO heals a selected object inline and then cleans abandoned parts,
-while RustFS admits a low-priority heal request that can be dropped and
-never calls `check_abandoned_parts` on that path.
-
-RustFS also has several load-bearing additions MinIO does not: dirty-usage
-fast wake, cycle budgets, leader-epoch fencing, remote NS-scanner protocol
-v6, checkpoint resume, and clean-idle backoff. Those should be preserved.
-
-| Area | Verdict | Why it matters |
-|---|---|---|
-| Cycle / leadership / bitrot mode | Close | Same `.bloomcycle.bin` counter, same deep-scan window of `healObjectSelectProb` cycles. |
-| Folder walk, compact, 1/16 + 1/1024 selection | Close | Constants and `mod` / `modAlt` schedule match. |
-| ILM eval + expiry/transition enqueue | Close | Same action set; RustFS additionally gates metrics on queue admission. |
-| Bucket replication repair | Close | Both call `queueReplicationHeal` / `queue_heal`. |
-| Scanner-selected object heal | **Gap** | MinIO `HealObject` is synchronous; RustFS async admission can drop the check. |
-| Abandoned-part cleanup on selected objects | **Gap** | MinIO calls `CheckAbandonedParts` after heal; RustFS scanner path does not. |
-| Bucket order across erasure sets | **Gap** | MinIO shuffles per set; RustFS is deterministic dirty→new→existing. |
-| `.usage-cache.bin` bytes | **Incompatible** | MinIO is zstd+msgp v8; RustFS is raw MessagePack. Reconstructable, not reusable. |
-| Excess-folder default | Differs | MinIO `50000`; RustFS `65538`. |
-| Alert event names | Differs | MinIO `s3:ObjectManyVersions`; RustFS `s3:Scanner:ManyVersions`. |
-
---
-
-## Architecture Overlay
-
-Both stacks are `init → leader lock → cycle → NSScanner → per-set disk walk →
-scanDataFolder → applyActions`.
-
-```text
-initDataScanner / init_data_scanner
-        │
-        ▼
-runDataScanner / run_data_scanner          (cluster leader lock)
-        │
-        ├─ load .bloomcycle.bin cycle state
-        ├─ getCycleScanMode (Normal vs Deep bitrot)
-        └─ NSScanner(wantCycle, scanMode)
-                │
-                ▼
-        per erasure set  (MinIO: er.nsScanner; RustFS: scanner_io)
-                │
-                ├─ load set .usage-cache.bin
-                ├─ bucket order (new first, then existing)
-                └─ per disk: NSScanner / scan_data_folder
-                        │
-                        ├─ lifecycle + replication config
-                        ├─ folder walk, compact, 1/16 skip
-                        ├─ getSize → applyActions
-                        │       ├─ ILM eval
-                        │       ├─ heal selected versions
-                        │       └─ healReplication
-                        └─ abandoned-children heal walk
-```
-
-| Stage | MinIO | RustFS |
-|---|---|---|
-| Startup | `initDataScanner` goroutine, random sleep ≥ 1s between `runDataScanner` calls | `init_data_scanner` in `crates/scanner/src/scanner.rs`; optional cold-cache / replication skip of start delay |
-| Leader | `globalLeaderLock.GetLock` (blocks) | `leader.lock` write lock with timeout; contended cycle returns and retries |
-| Cycle persist | LE `uint64` + msgp `currentScannerCycle` at `.bloomcycle.bin` | LE `uint64` + optional `RSCYC001` epoch header + msgpack `CurrentCycle` |
-| Set walk | `erasureObjects.nsScanner` in MinIO `cmd/erasure.go` | `crates/scanner/src/scanner_io.rs` |
-| Folder walk | `folderScanner.scanFolder` | `crates/scanner/src/scanner_folder.rs` |
-| Object actions | `scannerItem.applyActions` | `ScannerItem::apply_actions` |
-| Usage publish | `storeDataUsageInBackend` ← `.usage.json` | `store_data_usage_in_backend` ← `.usage.v2.json` (legacy `.usage.json` read-only; `scanner-usage-v2` in [compat-cleanup-register.md](compat-cleanup-register.md)) |
-
---
-
-## What Already Matches
-
-These are not gaps. Treat regressions here as MinIO-parity bugs.
-
-### Cycle constants and heal selection
-
-| Constant | MinIO | RustFS | Evidence |
-|---|---|---|---|
-| Folder sleep quantum | 1ms | sleeper `MIN_SLEEP` 1ms | MinIO `dataScannerSleepPerFolder`; `crates/scanner/src/sleeper.rs` |
-| Compacted-leaf visit period | 16 | 16 (`RUSTFS_DATA_USAGE_UPDATE_DIR_CYCLES`) | MinIO `dataUsageUpdateDirCycles`; `crates/scanner/src/scanner_folder.rs` `DATA_USAGE_UPDATE_DIR_CYCLES` |
-| Heal object probability | 1024 | 1024 (`RUSTFS_HEAL_OBJECT_SELECT_PROB`) | MinIO `healObjectSelectProb`; `DEFAULT_HEAL_OBJECT_SELECT_PROB` |
-| Compact least objects | 500 | 500 | both `dataScannerCompactLeastObject` / `DATA_SCANNER_COMPACT_LEAST_OBJECT` |
-| Compact at children | 10000 | 10000 | both |
-| Compact at folders | 2500 | 2500 | `children/4` |
-| Force compact folders | 250000 | 250000 | both |
-| Start delay default | 1 minute | speed-preset derived (default 1 minute) | MinIO `dataScannerStartDelay`; RustFS speed preset |
-| Excess versions | 100 | 100 | MinIO `scannerExcessObjectVersions`; `DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS` |
-| Excess version size | 1 TiB | 1 TiB | MinIO `scannerExcessObjectVersionsTotalSize`; `DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE` |
-
-Folder skip uses `hash.mod(nextCycle, 16)`. Object heal uses
-`hash.modAlt(nextCycle/div, healObjectSelect/div)`. Compacted folders raise
-`objectHealProbDiv` to 16 so the 1/1024 overall probability still holds.
-RustFS copies this in `scan_folder` (`mod_` + `object_heal_prob_div`).
-
-Path identity is the cleaned path string, not a digest. MinIO `hashPath` is
-`path.Clean`; RustFS `hash_path` in `crates/data-usage/src/data_usage.rs`
-cleans the same way. `xxhash` is only used in MinIO `mod` / `modAlt`.
-
-### Bitrot cycle window
-
-Both enter Deep scan when:
-
- bitrot cycle is `0` (always deep), or
- `current - bitrotStartCycle < healObjectSelectProb`, or
- wall time since `bitrotStartTime` exceeds the configured bitrot cycle.
-
-MinIO: `getCycleScanMode` in `cmd/data-scanner.go`. RustFS:
-`get_cycle_scan_mode` in `crates/scanner/src/scanner.rs`. Both persist
-`backgroundHealInfo` / `BackgroundHealInfo` and skip it on single-disk
-(`globalIsErasureSD` / `scanner_is_erasure_sd`).
-
-### ILM action coverage
-
-`apply_actions` covers the same action enum MinIO does: delete, delete
-version, delete restored, delete-all, del-marker-delete-all, transition,
-and none (heal + replication). Evaluator is constructed with lock
-retention and replication config. Free versions are swept through
-`enqueue_free_version` / `enqueue_runtime_free_version`. Noncurrent
-versions batch through `enqueueNoncurrentVersions` /
-`enqueue_runtime_newer_noncurrent`.
-
-### Speed presets
-
-`fastest` / `fast` / `default` / `slow` / `slowest` map to the same delay,
-max-wait, and cycle defaults MinIO `LookupConfig` uses (`0/0/1s`,
-`1/100ms/1m`, `2/1s/1m`, `10/15s/1m`, `100/15s/30m`).
-
-### Abandoned-children folder heal
-
-When a previously cached child is missing from the current directory listing,
-both scanners quorum-list the prefix and enqueue bucket/object heals. RustFS
-keeps this walk in `scan_folder` after the new/existing folder scans.
-
-### Read-path heal still exists
-
-MinIO also heals from GET/HEAD and MRF; the scanner is not the only heal
-source (MinIO PR 18050). RustFS GET decode errors enqueue
-`HealRequestSource::ReadRepair` in `crates/ecstore/src/set_disk/read.rs`.
-Scanner-heal gaps therefore delay *background* repair, not all repair.
-
---
-
-## Gaps
-
-Severity is the operator-visible failure if the gap is left as-is.
-
-### G1 — Scanner object heal is async and droppable (high)
-
-MinIO `applyHealing` calls `ObjectLayer.HealObject` and waits. The folder
-walker then treats `getSize` as having already healed the object
-(`cmd/data-scanner.go`, comment on `abandonedChildren` deletion). After a
-successful heal it always runs `CheckAbandonedParts` with
-`Remove: healDeleteDangling`.
-
-RustFS `heal_actions` always returns the original `actual_size` and, when
-heal is selected, calls `enqueue_heal` → `send_heal_request_with_admission`
-at `HealChannelPriority::Low`. `Full` and `Dropped` admissions are logged
-and skipped. `RUSTFS_SCANNER_INLINE_HEAL_ENABLE` only warns
-`inline_heal_rollback_unsupported` (`warn_inline_heal_compat_requested` in
-`crates/scanner/src/scanner_folder.rs`).
-
-**Failure:** a 1/1024-selected object with a missing shard can remain
-unhealed for many more cycles if the heal channel is full. Bitrot Deep
-selection has the same drop window. Usage accounting is unchanged by heal
-outcome, so a reconstructed size never replaces the pre-heal size in that
-cycle.
-
-**Do not "fix" this by making every scanner heal inline on the walk
-goroutine.** MinIO can afford that because `HealObject` is the storage
-layer. RustFS already has a heal worker pool and admission. The missing
-contract is: scanner-selected heals must be durable (pending_heals retry)
-and must not be silent-dropped without a later guaranteed retry.
-
-Pending heals already exist for some metadata/abandoned-child failures
-(`PendingScannerHeal` in the usage cache). Object-selection heals that hit
-`HealAdmissionResult::Full` do not currently join that retry list.
-
-### G2 — No `CheckAbandonedParts` on the scanner object-heal path (high)
-
-`check_abandoned_parts` is implemented on the store
-(`crates/ecstore/src/store/heal.rs`, `crates/ecstore/src/set_disk/ops/heal.rs`)
-and is in the object API. The scanner never calls it. The heal task
-processor (`crates/heal/src/heal/task.rs`) also does not call it after a
-scanner-originated `heal_object`.
-
-MinIO records this as `scannerMetricCleanAbandoned`. RustFS defines
-`Metric::CleanAbandoned` in `crates/common/src/metrics.rs` but the scanner
-crate never records it.
-
-**Failure:** leftover `part.N` files after a successful object heal stay
-until some other heal path notices them. Disk usage and bitrot surface
-area remain inflated.
-
-### G3 — Erasure-set bucket order is not shuffled (medium)
-
-MinIO `nsScanner` builds a permutation of buckets, emits *new* buckets
-(absent from the old cache) first in that random order, then existing
-buckets in that random order. Comment: otherwise the same buckets are
-scanned across every erasure set at the same time.
-
-RustFS `bucket_usage_scan_order` in `crates/scanner/src/scanner_io.rs` is
-deterministic: dirty buckets, then cache-miss (new) buckets, then
-cache-hit buckets, preserving `ListBuckets` order.
-
-Dirty-first is a RustFS improvement (MinIO has no dirty-usage wake). The
-gap is the *existing* bucket tail: under many buckets and several sets,
-RustFS lock-steps ILM/heal/replication load onto the same prefixes.
-
-### G4 — Alert event names and audit channel (medium)
-
-MinIO emits `event.ObjectManyVersions`, `event.ObjectLargeVersions`,
-`event.PrefixManyFolders`, plus `auditLogInternal` events
-`scanner:manyversions` / `scanner:largeversions` / `scanner:manyprefixes`.
-
-RustFS emits `s3:Scanner:ManyVersions`, `s3:Scanner:LargeVersions`,
-`s3:Scanner:BigPrefix` (`EVENT_SCANNER_*` in
-`crates/scanner/src/scanner_folder.rs`; wire names in
-`crates/s3-types/src/event_name.rs`). Notifications are edge-held 24h
-(MinIO re-emits every cycle). There is no scanner audit-log counterpart.
-
-**Failure:** notification destinations configured for MinIO event names
-miss RustFS scanner alerts. Audit pipelines that key on
-`scanner:manyversions` see nothing.
-
-### G5 — Excess-folder default differs (low)
-
-MinIO `scannerExcessFolders` default is `50000`
-(`internal/config/scanner/scanner.go`). RustFS
-`DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS` is `65538`
-(`crates/config/src/constants/scanner.rs`).
-
-**Failure:** the same prefix is silent on MinIO and noisy on RustFS (or
-the reverse if an operator copied MinIO runbooks).
-
-### G6 — `.usage-cache.bin` is not MinIO-readable (medium for migration, low otherwise)
-
-MinIO writes one version byte (`dataUsageCacheVerCurrent = 8`) plus zstd
-plus msgp (`cmd/data-usage-cache.go` `serializeTo`). RustFS
-`DataUsageCache::save_inner` writes uncompressed `rmp_serde` with no
-version byte (`crates/scanner/src/data_usage_define.rs`).
-
-Both use the same object name `.usage-cache.bin` and a `.bkp` sibling.
-A MinIO disk set attached to RustFS rebuilds the tree on first scan; the
-bytes are not reused. The inverse is also true.
-
-`.bloomcycle.bin` is closer: both start with a little-endian `u64` next
-cycle. RustFS additionally writes `RSCYC001` + leader epoch when fencing
-is active, and still reads a bare 8-byte or 8-byte+msgpack MinIO blob
-(`decode_scanner_cycle_state`). MinIO cannot consume the fenced form.
-
-Cluster usage snapshots diverge on purpose: MinIO `.usage.json`, RustFS
-authoritative `.usage.v2.json`. That is already a compat register item,
-not a scanner-logic bug.
-
-### G7 — Heal-selected usage size ignores heal result (low)
-
-MinIO `healActions` replaces `actualSz` with `HealObject`'s
-`res.ObjectSize` when positive. RustFS `heal_actions` always returns
-`actual_size`. Wrong sizes persist until the *next* cycle that both
-selects the object *and* observes healed metadata.
-
-This is secondary to G1: without a completed heal there is no new size.
-
-### G8 — Operator metric names and `mc admin scanner info` (low)
-
-MinIO `scannerMetric.String()` is PascalCase (`ReadMetadata`, `ScanObject`,
-`ILM`). RustFS `Metric::as_str` is snake_case (`read_metadata`,
-`scan_object`, `ilm`). `mc admin scanner info` against a RustFS
-`ScannerMetrics.life_time_ops` map will not match MinIO dashboard keys.
-
-RustFS exposes a richer `/v3/scanner/status` (freshness, runtime config
-sources, cycle schedule, admission). That is the supported operator
-surface; MinIO `mc` scanner info is not a compatibility target unless
-explicitly added.
-
-### G9 — Unversioned replication heal gate (low)
-
-MinIO `healReplication` returns immediately when `oi.VersionID == ""`.
-RustFS allows the call when the object is a delete marker or has a
-version-purge status even if `version_id` is none/nil
-(`ScannerItem::heal_replication`). This is likely *more* correct for
-purge/delete-marker repair on unversioned-looking entries, but it is a
-behavioral difference worth pinning with a test rather than copying
-MinIO's empty-VersionID skip blindly.
-
---
-
-## RustFS-Only Behavior To Keep
-
-These are not MinIO gaps. Removing them to "match MinIO" would be a
-regression.
-
-| Addition | Where | Why keep it |
-|---|---|---|
-| Dirty-usage fast wake + superseded retry (5s base) | `crates/scanner/src/scanner.rs`, `scanner_io.rs` `record_dirty_usage_bucket` | Quota/usage lag after write bursts; MinIO waits a full cycle. |
-| Cycle object/directory/runtime budgets | `crates/scanner/src/scanner_budget.rs` | Bounds scanner blast radius; MinIO only sleeps. |
-| Leader epoch + CAS persist | `encode_scanner_cycle_state` | Split-brain cycle counters after lock loss. |
-| Remote NS-scanner protocol v6 | `crates/scanner/src/remote_scanner.rs`; compat `ns-scanner-rpc-v3` | Distributed disk walks with fencing. |
-| Scan checkpoints / resume hints | `DataUsageScanCheckpoint` | Partial cycles after budget cancel. |
-| Clean-idle backoff (single-disk / erasure) | `ScannerCleanIdleBackoff` | Stops minute-cadence full walks on idle namespaces. |
-| Heal/replication admission metrics | `HealAdmissionResult`, `ScannerReplicationQueueAdmission` | Makes G1 observable; MinIO has no equivalent queue. |
-| Alert emission cooldown | 24h edge-hold | Avoids notification storms MinIO still has. |
-
---
-
-## Improvement Workstreams
-
-These are contracts, not a checklist. Each workstream is independently
-shippable. Do not couple them into one "make scanner like MinIO" rewrite.
-
-### W1 — Durable scanner-selected heal (closes G1, G7)
-
-**Invariant:** if an object is selected by `modAlt` in a cycle that
-`should_heal()`, that object/version is either healed, recorded in
-`pending_heals` for a later cycle, or the cycle is marked incomplete for
-heal work. Silent `Full`/`Dropped` is not a success.
-
-**Shape:** keep the heal channel. On `Full`/`Dropped`, persist
-`PendingScannerHeal` (object, version, scan mode) the same way abandoned
-metadata heals already persist. Retry at high or at least non-droppable
-priority next cycle. When a heal *completes*, optionally replace the
-accounted size with the healed size (G7).
-
-**Do not:** call `HealObject` inline from `scan_folder` as a default. The
-unsupported `RUSTFS_SCANNER_INLINE_HEAL_ENABLE` warning exists because
-that rollback fights the worker pool. An opt-in inline path is only
-justified if a measured admission-drop rate stays high after durable
-retry.
-
-**Tests:** (a) selected object missing one shard, heal channel full →
-pending_heals non-empty, next cycle heals it; (b) Deep mode + recent
-mtime stays Normal (existing cooldown); (c) usage size updates only after
-heal success; (d) revert of pending_heals-on-drop fails the test.
-
-### W2 — Abandoned-part cleanup after scanner object heal (closes G2)
-
-**Invariant:** a scanner-selected object heal that succeeds (or that the
-heal worker reports as already consistent) runs `check_abandoned_parts`
-with dangling removal, matching MinIO `healDeleteDangling = true`.
-
-**Shape:** call it from the heal worker when `source == Scanner`, not
-from the folder walk. That keeps IO off the scanner hot path. Record
-`Metric::CleanAbandoned` so last-minute scanner metrics are not a dead
-enum.
-
-**Tests:** object with an extra `part.N` after a valid heal → part
-removed; dry-run heal does not delete (existing set_disk tests stay
-authoritative); `CleanAbandoned` lifetime counter increments.
-
-### W3 — Per-set shuffle of existing buckets (closes G3)
-
-**Invariant:** dirty and new buckets still go first (RustFS dirty-usage
-contract). The existing-bucket tail is shuffled per erasure set per
-cycle so sets do not scan the same prefix concurrently.
-
-**Shape:** smallest change is `bucket_usage_scan_order` taking a
-per-set RNG seed (cycle + pool + set). Do not shuffle dirty buckets;
-that would delay quota/usage repair.
-
-**Tests:** two sets, three existing buckets, same cycle → different
-existing tails; dirty bucket always index 0.
-
-### W4 — Notification and audit aliases (closes G4, optionally G5)
-
-**Invariant:** a destination subscribed to MinIO names
-`s3:ObjectManyVersions` / `s3:ObjectLargeVersions` /
-`s3:PrefixManyFolders` receives RustFS scanner alerts. Keep the current
-`s3:Scanner:*` names as aliases, not replacements, until clients migrate.
-
-**Shape:** dual-name parse in `crates/s3-types/src/event_name.rs` (already
-comments "corresponding to Go") plus dual emit, or a compatibility
-mapping at notify dispatch. Audit events are optional and should reuse
-the existing audit pipeline rather than a scanner-specific logger.
-
-Align `DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS` to `50000` only with a
-release note; 65538 is not a bug, it is a silent default drift.
-
-### W5 — Cache-format interop (closes G6 only if migration requires it)
-
-**Invariant for RustFS-only clusters:** none. Rebuilding `.usage-cache.bin`
-on first scan is acceptable.
-
-**Invariant if MinIO disk import is a product goal:** either detect MinIO
-v8 zstd+msgp and ignore/rebuild, or implement a one-shot importer.
-Writing MinIO-shaped cache from RustFS is not required for serving
-objects.
-
-Document in operations that `.usage-cache.bin` is not a migration
-artifact. `.bloomcycle.bin` 8-byte prefix already round-trips.
-
-### W6 — Operator surface (closes G8)
-
-Keep `/v3/scanner/status` as the source of truth. If `mc admin scanner
-info` support is required, add a madmin-shaped projection with PascalCase
-`life_time_ops` keys *in addition to* snake_case, behind a documented
-compat flag. Do not rename RustFS metrics; Prometheus and status JSON
-already use snake_case.
-
---
-
-## Suggested Verification (when a workstream ships)
-
-Scanner changes are high-risk under AGENTS.md (lifecycle/tiering,
-heal, S3-visible usage). A workstream PR should run:
-
- `cargo fmt --all --check`
- `cargo test -p rustfs-scanner` (and heal tests for W2)
- the crate's lifecycle integration tests when ILM admission changes
- `make doc-paths-check` if this file's citations move
-
-Do not run `make pre-pr` for documentation-only edits of this page.
-
---
-
-## Sources
-
-RustFS:
-
- `crates/scanner/src/scanner.rs` — leader loop, cycle fencing, bitrot mode
- `crates/scanner/src/scanner_folder.rs` — folder walk, ILM, heal, alerts
- `crates/scanner/src/scanner_io.rs` — NSScanner, bucket order, dirty usage
- `crates/scanner/src/scanner_budget.rs` — cycle budgets
- `crates/scanner/src/sleeper.rs` — proportional throttle
- `crates/scanner/src/data_usage_define.rs` — cache persist
- `crates/scanner/src/runtime_config.rs` — env/config resolution
- `crates/config/src/constants/scanner.rs` — defaults
- `crates/common/src/metrics.rs` — metric enum (MinIO-shaped)
- `rustfs/src/admin/handlers/scanner.rs` — `/v3/scanner/status`
- [compat-cleanup-register.md](compat-cleanup-register.md) — `scanner-usage-v2`, `ns-scanner-rpc-v3`
-
-MinIO (`minio/minio` master, 2026-08-18):
-
- `cmd/data-scanner.go` — init/run, applyActions, healReplication, sleeper
- `cmd/data-scanner-metric.go` — metric enum and `mc` report
- `cmd/erasure.go` — `nsScanner` shuffle and per-disk walk
- `cmd/xl-storage.go` — disk `NSScanner` / getSize
- `cmd/data-usage-cache.go` — hash mod, zstd+msgp cache
- `cmd/data-usage.go` — `.usage.json` / `.bloomcycle.bin` names
- `internal/config/scanner/scanner.go` — speed presets and alert defaults
@@ -63,7 +63,7 @@
 | list_objects_v2_metadata_extension_test | 1 |  |
 | list_objects_v2_pagination_test | 12 | ✅ |
 | mc_mirror_small_bucket_test | 1 |  |
-| multipart_auth_test | 85 |  |
+| multipart_auth_test | 75 |  |
 | multipart_storage_class_test | 3 | ✅ |
 | namespace_lock_quorum_test | 2 |  |
 | negative_sigv4_test | 6 | ✅ |
@@ -317,7 +317,6 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea
    total.auto_heal = total.auto_heal.saturating_add(next.auto_heal);
    total.internal = total.internal.saturating_add(next.internal);
    total.read_repair = total.read_repair.saturating_add(next.read_repair);
-    total.mrf = total.mrf.saturating_add(next.mrf);
 }

 fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) {
@@ -2354,7 +2353,6 @@ mod tests {
            auto_heal: value,
            internal: value,
            read_repair: value,
-            mrf: value,
        };
        let operations = |value| rustfs_heal::HealOperationsSnapshot {
            queue_length: value,
@@ -22,7 +22,6 @@ use crate::admin::runtime_sources::{
    current_object_store_handle_for_context, current_or_init_kms_runtime_service_manager,
 };
 use crate::admin::storage_api::config::{read_admin_config, save_admin_config};
-use crate::admin::storage_api::error::StorageError;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use hyper::{Method, StatusCode};
@@ -279,11 +278,8 @@ pub async fn load_kms_config() -> Option<KmsConfig> {
            }
        },
        Err(e) => {
-            // Config not found is normal on first run: `read_config` maps a missing or
-            // empty config object to `ConfigNotFound`, so that variant is the only
-            // "absent" signal reaching here. Every other not-found variant (disk,
-            // volume, bucket) means degraded storage and must stay a warning.
-            if matches!(e, StorageError::ConfigNotFound) {
+            // Config not found is normal on first run
+            if e.to_string().contains("ConfigNotFound") || e.to_string().contains("not found") {
                info!(
                    component = LOG_COMPONENT_ADMIN,
                    subsystem = LOG_SUBSYSTEM_KMS,
@@ -16,12 +16,13 @@

 use super::kms_dynamic::current_kms_config_fingerprint;
 use super::kms_keys::{CreateKeyHandler, DescribeKeyHandler, GenerateDataKeyHandler, ListKeysHandler};
-use crate::admin::auth::authorize_admin_request;
+use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
    current_kms_runtime_service_manager, current_notification_system, current_or_init_kms_runtime_service_manager,
 };
-use crate::server::ADMIN_PREFIX;
+use crate::auth::{check_key_valid, get_session_token};
+use crate::server::{ADMIN_PREFIX, RemoteAddr};
 use hyper::{HeaderMap, Method, StatusCode};
 use matchit::Params;
 use rustfs_kms::KmsBackend;
@@ -68,18 +69,6 @@ fn kms_clear_cache_actions() -> Vec<Action> {
    vec![Action::KmsAction(KmsAction::ClearCacheAction)]
 }

-/// Admin gate for the KMS management endpoints, none of which act on a key.
-///
-/// The pre-check keeps these endpoints' historical missing-credentials message;
-/// the shared gate reports "get cred failed".
-async fn authorize_kms_management_request(req: &S3Request<Body>, actions: Vec<Action>) -> S3Result<()> {
-    if req.credentials.is_none() {
-        return Err(s3_error!(InvalidRequest, "authentication required"));
-    }
-    authorize_admin_request(req, actions).await?;
-    Ok(())
-}
-
 /// Response of `POST /kms/clear-cache`.
 ///
 /// Declared rather than built inline so the shape the console already depends
@@ -271,7 +260,22 @@ pub struct KmsStatusHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsStatusHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        authorize_kms_management_request(&req, kms_service_control_actions()).await?;
+        let Some(cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "authentication required"));
+        };
+
+        let (cred, owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
+
+        validate_admin_request(
+            &req.headers,
+            &cred,
+            owner,
+            false,
+            kms_service_control_actions(),
+            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        )
+        .await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -322,7 +326,22 @@ pub struct KmsConfigHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsConfigHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        authorize_kms_management_request(&req, kms_configure_actions()).await?;
+        let Some(cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "authentication required"));
+        };
+
+        let (cred, owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
+
+        validate_admin_request(
+            &req.headers,
+            &cred,
+            owner,
+            false,
+            kms_configure_actions(),
+            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        )
+        .await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -356,7 +375,22 @@ pub struct KmsClearCacheHandler {}
 #[async_trait::async_trait]
 impl Operation for KmsClearCacheHandler {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
-        authorize_kms_management_request(&req, kms_clear_cache_actions()).await?;
+        let Some(cred) = req.credentials else {
+            return Err(s3_error!(InvalidRequest, "authentication required"));
+        };
+
+        let (cred, owner) =
+            check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &cred.access_key).await?;
+
+        validate_admin_request(
+            &req.headers,
+            &cred,
+            owner,
+            false,
+            kms_clear_cache_actions(),
+            req.extensions.get::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+        )
+        .await?;

        let Some(service) = kms_encryption_service_from_context().await else {
            return Err(s3_error!(InternalError, "KMS service not initialized"));
@@ -388,14 +422,9 @@ impl Operation for KmsClearCacheHandler {

 #[cfg(test)]
 mod tests {
-    use super::{
-        KmsClearCacheResponse, authorize_kms_management_request, kms_clear_cache_actions, kms_configure_actions,
-        kms_service_control_actions,
-    };
+    use super::{KmsClearCacheResponse, kms_clear_cache_actions, kms_configure_actions, kms_service_control_actions};
    use crate::admin::handlers::kms_keys::stable_json_value;
-    use hyper::HeaderMap;
    use rustfs_policy::policy::action::{Action, AdminAction, KmsAction};
-    use s3s::{Body, S3Request};

    fn assert_has_action(actions: &[Action], action: Action) {
        assert!(actions.contains(&action), "expected action list to contain {action:?}");
@@ -405,58 +434,6 @@ mod tests {
        assert!(!actions.contains(&action), "expected action list not to contain {action:?}");
    }

-    /// These endpoints authorize through the shared admin gate, which reports
-    /// "get cred failed" for a credential-less request. The pre-check keeps the
-    /// message these endpoints have always returned (rustfs/backlog#1829).
-    #[tokio::test]
-    async fn kms_management_gate_keeps_its_missing_credentials_message() {
-        let req = S3Request {
-            input: Body::from(String::new()),
-            method: http::Method::GET,
-            uri: "/rustfs/admin/v3/kms/status".parse().expect("uri should parse"),
-            headers: HeaderMap::new(),
-            extensions: http::Extensions::new(),
-            credentials: None,
-            region: None,
-            service: None,
-            trailing_headers: None,
-        };
-
-        let err = authorize_kms_management_request(&req, kms_service_control_actions())
-            .await
-            .expect_err("a request without credentials must be rejected");
-        assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
-        assert_eq!(err.message(), Some("authentication required"));
-    }
-
-    /// Every management endpoint must reach the shared gate, each with its own
-    /// action set. The action lists are pinned above, but nothing else checks
-    /// which handler asks for which, and a handler that lost its gate entirely
-    /// would still serve its response.
-    #[test]
-    fn management_handlers_authorize_with_their_dedicated_actions() {
-        let src = include_str!("kms_management.rs");
-
-        for (handler, actions) in [
-            ("KmsStatusHandler", "kms_service_control_actions()"),
-            ("KmsConfigHandler", "kms_configure_actions()"),
-            ("KmsClearCacheHandler", "kms_clear_cache_actions()"),
-        ] {
-            let block = src
-                .split_once(&format!("impl Operation for {handler}"))
-                .unwrap_or_else(|| panic!("{handler} impl should exist"))
-                .1;
-            let end = block
-                .find("\nimpl Operation for")
-                .or_else(|| block.find("\n#[cfg(test)]"))
-                .unwrap_or(block.len());
-            assert!(
-                block[..end].contains(&format!("authorize_kms_management_request(&req, {actions})")),
-                "{handler} must authorize through the shared gate with {actions}"
-            );
-        }
-    }
-
    #[test]
    fn kms_management_auth_actions_use_dedicated_kms_actions() {
        assert_has_action(&kms_service_control_actions(), Action::KmsAction(KmsAction::ServiceControlAction));
@@ -67,9 +67,6 @@ use rustfs_policy::policy::action::{Action, S3Action};
 use rustfs_s3_types::EventName;
 use rustfs_signer::pre_sign_v4;
 use rustfs_utils::egress::{OutboundDnsResolver, OutboundPolicy};
-use rustfs_utils::http::headers::{
-    AMZ_CHECKSUM_CRC32, AMZ_CHECKSUM_CRC32C, AMZ_CHECKSUM_CRC64NVME, AMZ_CHECKSUM_SHA1, AMZ_CHECKSUM_SHA256, AMZ_CHECKSUM_TYPE,
-};
 use rustfs_utils::http::{
    SUFFIX_SOURCE_DELETEMARKER, SUFFIX_SOURCE_MTIME, SUFFIX_SOURCE_REPLICATION_CHECK, SUFFIX_SOURCE_REPLICATION_REQUEST,
    SUFFIX_SOURCE_VERSION_ID, get_source_scheme, insert_header,
@@ -1034,24 +1031,28 @@ fn build_get_object_response_headers(output: &GetObjectOutput, base_headers: &He
        )?;
    }
    if let Some(checksum_crc32) = &output.checksum_crc32 {
-        insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC32), checksum_crc32.clone())?;
+        insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-crc32"), checksum_crc32.clone())?;
    }
    if let Some(checksum_crc32c) = &output.checksum_crc32c {
-        insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC32C), checksum_crc32c.clone())?;
+        insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-crc32c"), checksum_crc32c.clone())?;
    }
    if let Some(checksum_crc64nvme) = &output.checksum_crc64nvme {
-        insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_CRC64NVME), checksum_crc64nvme.clone())?;
+        insert_string_header(
+            &mut headers,
+            HeaderName::from_static("x-amz-checksum-crc64nvme"),
+            checksum_crc64nvme.clone(),
+        )?;
    }
    if let Some(checksum_sha1) = &output.checksum_sha1 {
-        insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_SHA1), checksum_sha1.clone())?;
+        insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-sha1"), checksum_sha1.clone())?;
    }
    if let Some(checksum_sha256) = &output.checksum_sha256 {
-        insert_string_header(&mut headers, HeaderName::from_static(AMZ_CHECKSUM_SHA256), checksum_sha256.clone())?;
+        insert_string_header(&mut headers, HeaderName::from_static("x-amz-checksum-sha256"), checksum_sha256.clone())?;
    }
    if let Some(checksum_type) = &output.checksum_type {
        insert_string_header(
            &mut headers,
-            HeaderName::from_static(AMZ_CHECKSUM_TYPE),
+            HeaderName::from_static("x-amz-checksum-type"),
            checksum_type.as_str().to_string(),
        )?;
    }
@@ -1113,12 +1114,12 @@ fn clear_object_lambda_variant_headers(headers: &mut HeaderMap) {
        http::header::ETAG,
        http::header::LAST_MODIFIED,
        http::header::EXPIRES,
-        HeaderName::from_static(AMZ_CHECKSUM_CRC32),
-        HeaderName::from_static(AMZ_CHECKSUM_CRC32C),
-        HeaderName::from_static(AMZ_CHECKSUM_CRC64NVME),
-        HeaderName::from_static(AMZ_CHECKSUM_SHA1),
-        HeaderName::from_static(AMZ_CHECKSUM_SHA256),
-        HeaderName::from_static(AMZ_CHECKSUM_TYPE),
+        HeaderName::from_static("x-amz-checksum-crc32"),
+        HeaderName::from_static("x-amz-checksum-crc32c"),
+        HeaderName::from_static("x-amz-checksum-crc64nvme"),
+        HeaderName::from_static("x-amz-checksum-sha1"),
+        HeaderName::from_static("x-amz-checksum-sha256"),
+        HeaderName::from_static("x-amz-checksum-type"),
        HeaderName::from_static("x-amz-tagging-count"),
        HeaderName::from_static("x-amz-request-route"),
        HeaderName::from_static("x-amz-request-token"),
@@ -56,8 +56,8 @@ use super::storage_api::object_usecase::bucket::{
 };
 use super::storage_api::object_usecase::compression::{MIN_DISK_COMPRESSIBLE_SIZE, is_disk_compressible};
 use super::storage_api::object_usecase::concurrency::{
-    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectAdmission, PutObjectGuard,
-    get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
+    self, ConcurrencyManager, DiskReadAdmission, GetObjectGuard, PutObjectGuard, get_concurrency_aware_buffer_size,
+    get_concurrency_manager, get_put_concurrency_aware_buffer_size,
 };
 #[cfg(test)]
 use super::storage_api::object_usecase::contract::http::HTTPPreconditions;
@@ -5681,35 +5681,6 @@ impl DefaultObjectUsecase {
        let server_side_encryption_requested =
            server_side_encryption.is_some() || sse_customer_algorithm.is_some() || ssekms_key_id.is_some();

-        // Resolve the store through the request-bound server context
-        // (backlog#1052 S6), not the process-global handle, so an embedded
-        // second server never writes into the first server's store.
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        validate_bucket_exists(&store, &bucket).await?;
-        rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
-
-        let put_admission = match get_concurrency_manager()
-            .admit_put_object()
-            .await
-            .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))?
-        {
-            PutObjectAdmission::Disabled => None,
-            PutObjectAdmission::Admitted(permit) => {
-                counter!("rustfs.put_object.foreground_admission.total", "result" => "admitted").increment(1);
-                Some(permit)
-            }
-            PutObjectAdmission::Rejected => {
-                counter!("rustfs.put_object.foreground_admission.total", "result" => "rejected").increment(1);
-                return Err(s3_error!(
-                    SlowDown,
-                    "foreground write concurrency limit reached, please reduce your request rate"
-                ));
-            }
-        };
-
        let mut put_request_guard = PutObjectGuard::new();
        let concurrent_put_requests = PutObjectGuard::concurrent_requests();

@@ -5762,6 +5733,16 @@ impl DefaultObjectUsecase {
            use_large_put_concurrency_tuning,
        );

+        // Resolve the store through the request-bound server context
+        // (backlog#1052 S6), not the process-global handle, so an embedded
+        // second server never writes into the first server's store.
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        let bucket_validate_stage_start = put_stage_metrics_enabled.then(Instant::now);
+        validate_bucket_exists(&store, &bucket).await?;
+        rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
+
        let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
        rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start);
@@ -6151,9 +6132,7 @@ impl DefaultObjectUsecase {
            let cache_adapter = cache_adapter.clone();
            let request_id = request_id.clone();
            let put_path = put_path.to_string();
-            let put_admission = put_admission;
            async move {
-                let _put_admission = put_admission;
                let object_traffic_progress = object_traffic_health
                    .as_deref()
                    .and_then(ObjectTrafficHealth::track_write_storage);
@@ -6204,7 +6183,6 @@ impl DefaultObjectUsecase {
                    }
                };
                rustfs_io_metrics::record_put_object_stage_duration_from("app_store_put", store_put_stage_start);
-                drop(_put_admission);
                drop(object_traffic_progress);
                #[cfg(test)]
                wait_for_put_post_store_test_hook(&bucket).await;
@@ -936,7 +936,7 @@ pub(crate) mod bucket {

 pub(crate) mod concurrency {
    pub(crate) use crate::storage::storage_api::concurrency_consumer::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
        get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
    };
 }
@@ -0,0 +1,618 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Backpressure Management for Object Data Transfer.
+//!
+//! This module provides backpressure-aware pipes for object data transfer,
+//! preventing buffer overflow and memory exhaustion under high concurrency.
+
+//! # Key Features
+//!
+//! - Configurable buffer size with high/low watermarks
+//! - Backpressure state monitoring and events
+//! - Backpressure metrics emitted through the shared metrics pipeline
+//! - Graceful handling of slow consumers
+//!
+//! # Architecture
+//!
+//! ```text
+//! [Disk Reader] --> [BackpressurePipe] --> [HTTP Response]
+//!                      |
+//!                      v
+//!               [Buffer Monitor]
+//!                      |
+//!                      v
+//!              [High Watermark?] --> Apply Backpressure
+//! ```
+
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::time::{Duration, Instant};
+use tokio::io::{DuplexStream, duplex};
+use tracing::{debug, warn};
+
+use metrics::counter;
+use rustfs_concurrency::PipeBackpressurePolicy;
+use rustfs_io_core::BackpressureConfig as CoreBackpressureConfig;
+
+/// Object-transfer duplex pipe backpressure policy.
+#[derive(Debug, Clone, Copy)]
+pub struct ObjectPipeBackpressurePolicy {
+    /// Buffer size in bytes (default 4MB).
+    pub buffer_size: usize,
+    /// High watermark percentage (default 80%).
+    /// When buffer usage exceeds this, backpressure is applied.
+    pub high_watermark: u32,
+    /// Low watermark percentage (default 50%).
+    /// When buffer usage drops below this after high watermark, backpressure is released.
+    pub low_watermark: u32,
+}
+
+impl Default for ObjectPipeBackpressurePolicy {
+    fn default() -> Self {
+        Self {
+            buffer_size: rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE,
+            high_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK,
+            low_watermark: rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK,
+        }
+    }
+}
+
+impl ObjectPipeBackpressurePolicy {
+    /// Load configuration from environment variables.
+    pub fn from_env() -> Self {
+        let buffer_size = rustfs_utils::get_env_usize(
+            rustfs_config::ENV_OBJECT_DUPLEX_BUFFER_SIZE,
+            rustfs_config::DEFAULT_OBJECT_DUPLEX_BUFFER_SIZE,
+        );
+        let high_watermark = rustfs_utils::get_env_u32(
+            rustfs_config::ENV_OBJECT_BACKPRESSURE_HIGH_WATERMARK,
+            rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_HIGH_WATERMARK,
+        );
+        let low_watermark = rustfs_utils::get_env_u32(
+            rustfs_config::ENV_OBJECT_BACKPRESSURE_LOW_WATERMARK,
+            rustfs_config::DEFAULT_OBJECT_BACKPRESSURE_LOW_WATERMARK,
+        );
+
+        Self {
+            buffer_size,
+            high_watermark,
+            low_watermark,
+        }
+    }
+
+    /// Calculate high watermark threshold in bytes.
+    pub fn high_watermark_bytes(&self) -> usize {
+        (self.buffer_size as u64 * self.high_watermark as u64 / 100) as usize
+    }
+
+    /// Calculate low watermark threshold in bytes.
+    pub fn low_watermark_bytes(&self) -> usize {
+        (self.buffer_size as u64 * self.low_watermark as u64 / 100) as usize
+    }
+
+    /// Project this object-transfer policy into the shared concurrency facade policy.
+    pub fn to_concurrency_policy(&self) -> PipeBackpressurePolicy {
+        PipeBackpressurePolicy {
+            buffer_size: self.buffer_size,
+            high_watermark: self.high_watermark,
+            low_watermark: self.low_watermark,
+        }
+    }
+
+    /// Project this object-transfer policy into the reusable io-core admission config.
+    pub fn to_core_config(&self) -> CoreBackpressureConfig {
+        self.to_concurrency_policy().to_core_config()
+    }
+}
+
+/// Backpressure state.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum BackpressureState {
+    /// Normal operation, buffer usage is below high watermark.
+    Normal,
+    /// Buffer usage is above high watermark, backpressure should be applied.
+    HighWatermark,
+    /// Backpressure is actively being applied to the producer.
+    BackpressureApplied,
+}
+
+impl std::fmt::Display for BackpressureState {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            BackpressureState::Normal => write!(f, "normal"),
+            BackpressureState::HighWatermark => write!(f, "high_watermark"),
+            BackpressureState::BackpressureApplied => write!(f, "backpressure_applied"),
+        }
+    }
+}
+
+/// Compact metadata snapshot for object-transfer backpressure pipes.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct BackpressurePipeMeta {
+    /// Buffer capacity in bytes.
+    pub buffer_capacity: usize,
+    /// Current backpressure state.
+    pub state: BackpressureState,
+    /// Age of the pipe since creation.
+    pub age: Duration,
+}
+
+/// Compact metadata snapshot for the lightweight backpressure monitor.
+#[derive(Debug, Clone, Copy, PartialEq)]
+pub struct BackpressureMonitorMeta {
+    /// Buffer capacity in bytes.
+    pub buffer_capacity: usize,
+    /// Current buffer usage percentage.
+    pub usage_percent: f32,
+    /// Current backpressure state.
+    pub state: BackpressureState,
+}
+
+fn calculate_usage_percent(usage: usize, capacity: usize) -> f32 {
+    if capacity > 0 {
+        (usage as f32 / capacity as f32) * 100.0
+    } else {
+        0.0
+    }
+}
+
+fn apply_watermark_transition(
+    in_high_watermark: &AtomicBool,
+    usage: usize,
+    high: usize,
+    low: usize,
+) -> (BackpressureState, bool) {
+    let current = in_high_watermark.load(Ordering::Acquire);
+    let next_state = if usage >= high {
+        BackpressureState::HighWatermark
+    } else if usage <= low {
+        BackpressureState::Normal
+    } else if current {
+        BackpressureState::HighWatermark
+    } else {
+        BackpressureState::Normal
+    };
+    let next_is_high = matches!(next_state, BackpressureState::HighWatermark);
+    let changed = in_high_watermark.swap(next_is_high, Ordering::AcqRel) != next_is_high;
+    (next_state, changed)
+}
+
+fn saturating_sub_atomic(value: &AtomicUsize, delta: usize) {
+    value
+        .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| Some(current.saturating_sub(delta)))
+        .ok();
+}
+
+/// A backpressure-aware pipe wrapping tokio's duplex.
+///
+/// This provides monitoring and events for backpressure conditions
+/// while maintaining compatibility with the standard duplex interface.
+pub struct BackpressurePipe {
+    /// Reader end of the duplex pipe.
+    reader: DuplexStream,
+    /// Writer end of the duplex pipe.
+    writer: DuplexStream,
+    /// Configuration.
+    config: ObjectPipeBackpressurePolicy,
+    /// Current buffer usage (approximate, updated on write).
+    buffer_usage: AtomicUsize,
+    /// Current backpressure state.
+    state: AtomicBool, // true = in high watermark state
+    /// Total bytes written.
+    total_written: AtomicUsize,
+    /// Total bytes read.
+    total_read: AtomicUsize,
+    /// Cached high watermark threshold in bytes.
+    high_watermark_bytes: usize,
+    /// Cached low watermark threshold in bytes.
+    low_watermark_bytes: usize,
+    /// Pipe creation timestamp.
+    created_at: Instant,
+}
+
+impl BackpressurePipe {
+    /// Create a new backpressure-aware pipe with default configuration.
+    pub fn new() -> Self {
+        Self::with_config(ObjectPipeBackpressurePolicy::from_env())
+    }
+
+    /// Create a new backpressure-aware pipe with custom configuration.
+    pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self {
+        let policy = config.to_concurrency_policy();
+        let (reader, writer) = duplex(policy.buffer_size);
+        let high_watermark_bytes = policy.high_watermark_bytes();
+        let low_watermark_bytes = policy.low_watermark_bytes();
+
+        debug!(
+            buffer_size = config.buffer_size,
+            high_watermark = config.high_watermark,
+            low_watermark = config.low_watermark,
+            high_watermark_bytes,
+            low_watermark_bytes,
+            "Created backpressure pipe"
+        );
+
+        Self {
+            reader,
+            writer,
+            config,
+            buffer_usage: AtomicUsize::new(0),
+            state: AtomicBool::new(false),
+            total_written: AtomicUsize::new(0),
+            total_read: AtomicUsize::new(0),
+            high_watermark_bytes,
+            low_watermark_bytes,
+            created_at: Instant::now(),
+        }
+    }
+
+    /// Take the reader end of the pipe (consumes self).
+    pub fn into_reader(self) -> DuplexStream {
+        self.reader
+    }
+
+    /// Take the writer end of the pipe (consumes self).
+    pub fn into_writer(self) -> DuplexStream {
+        self.writer
+    }
+
+    /// Split into reader and writer (consumes self).
+    pub fn split(self) -> (DuplexStream, DuplexStream) {
+        (self.reader, self.writer)
+    }
+
+    /// Get current backpressure state.
+    pub fn state(&self) -> BackpressureState {
+        if self.state.load(Ordering::Acquire) {
+            BackpressureState::BackpressureApplied
+        } else {
+            BackpressureState::Normal
+        }
+    }
+
+    /// Get a compact metadata snapshot for the pipe.
+    pub fn meta(&self) -> BackpressurePipeMeta {
+        BackpressurePipeMeta {
+            buffer_capacity: self.config.buffer_size,
+            state: self.state(),
+            age: self.age(),
+        }
+    }
+
+    /// Get the age of this pipe.
+    pub fn age(&self) -> Duration {
+        self.created_at.elapsed()
+    }
+
+    /// Get current buffer usage.
+    pub fn usage(&self) -> usize {
+        self.buffer_usage.load(Ordering::Acquire)
+    }
+
+    /// Record bytes written (call after successful write).
+    pub fn record_write(&self, bytes: usize) {
+        self.total_written.fetch_add(bytes, Ordering::Relaxed);
+        self.buffer_usage.fetch_add(bytes, Ordering::Release);
+        self.update_watermark_state();
+    }
+
+    /// Record bytes read (call after successful read).
+    pub fn record_read(&self, bytes: usize) {
+        self.total_read.fetch_add(bytes, Ordering::Relaxed);
+        saturating_sub_atomic(&self.buffer_usage, bytes);
+        self.update_watermark_state();
+    }
+
+    /// Update watermark state and emit transition signals.
+    fn update_watermark_state(&self) {
+        let usage = self.buffer_usage.load(Ordering::Acquire);
+        let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32;
+        let (next_state, changed) =
+            apply_watermark_transition(&self.state, usage, self.high_watermark_bytes, self.low_watermark_bytes);
+
+        if changed {
+            match next_state {
+                BackpressureState::HighWatermark => {
+                    counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1);
+
+                    warn!(
+                        buffer_usage = usage,
+                        buffer_capacity = self.config.buffer_size,
+                        usage_percent,
+                        high_watermark = self.config.high_watermark,
+                        "Backpressure: high watermark reached"
+                    );
+                }
+                BackpressureState::Normal => {
+                    counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1);
+
+                    debug!(
+                        buffer_usage = usage,
+                        buffer_capacity = self.config.buffer_size,
+                        usage_percent,
+                        low_watermark = self.config.low_watermark,
+                        "Backpressure: returned to normal"
+                    );
+                }
+                BackpressureState::BackpressureApplied => {}
+            }
+        }
+    }
+
+    /// Get total bytes written.
+    pub fn total_written(&self) -> usize {
+        self.total_written.load(Ordering::Relaxed)
+    }
+
+    /// Get total bytes read.
+    pub fn total_read(&self) -> usize {
+        self.total_read.load(Ordering::Relaxed)
+    }
+
+    /// Get buffer capacity.
+    pub fn capacity(&self) -> usize {
+        self.config.buffer_size
+    }
+}
+
+impl Default for BackpressurePipe {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// A simple wrapper that provides backpressure monitoring for duplex streams.
+///
+/// This is a lighter-weight alternative to `BackpressurePipe` that doesn't
+/// wrap the streams but provides monitoring capabilities.
+pub struct BackpressureMonitor {
+    /// Configuration.
+    config: ObjectPipeBackpressurePolicy,
+    /// Current buffer usage.
+    buffer_usage: AtomicUsize,
+    /// In high watermark state.
+    in_high_watermark: AtomicBool,
+    /// Cached high watermark threshold in bytes.
+    high_watermark_bytes: usize,
+    /// Cached low watermark threshold in bytes.
+    low_watermark_bytes: usize,
+}
+
+impl BackpressureMonitor {
+    /// Create a new monitor with default configuration.
+    pub fn new() -> Self {
+        Self::with_config(ObjectPipeBackpressurePolicy::from_env())
+    }
+
+    /// Create a new monitor with custom configuration.
+    pub fn with_config(config: ObjectPipeBackpressurePolicy) -> Self {
+        let policy = config.to_concurrency_policy();
+        let high_watermark_bytes = policy.high_watermark_bytes();
+        let low_watermark_bytes = policy.low_watermark_bytes();
+        Self {
+            config,
+            buffer_usage: AtomicUsize::new(0),
+            in_high_watermark: AtomicBool::new(false),
+            high_watermark_bytes,
+            low_watermark_bytes,
+        }
+    }
+
+    /// Record bytes added to buffer.
+    pub fn on_write(&self, bytes: usize) -> BackpressureState {
+        self.buffer_usage.fetch_add(bytes, Ordering::Release);
+        self.update_state()
+    }
+
+    /// Record bytes removed from buffer.
+    pub fn on_read(&self, bytes: usize) -> BackpressureState {
+        saturating_sub_atomic(&self.buffer_usage, bytes);
+        self.update_state()
+    }
+
+    /// Get current state.
+    pub fn state(&self) -> BackpressureState {
+        if self.in_high_watermark.load(Ordering::Acquire) {
+            BackpressureState::HighWatermark
+        } else {
+            BackpressureState::Normal
+        }
+    }
+
+    /// Get current buffer usage.
+    pub fn usage(&self) -> usize {
+        self.buffer_usage.load(Ordering::Acquire)
+    }
+
+    /// Get usage percentage.
+    pub fn usage_percent(&self) -> f32 {
+        let usage = self.buffer_usage.load(Ordering::Acquire);
+        calculate_usage_percent(usage, self.config.buffer_size)
+    }
+
+    /// Get a compact metadata snapshot for the monitor.
+    pub fn meta(&self) -> BackpressureMonitorMeta {
+        let usage = self.buffer_usage.load(Ordering::Acquire);
+        BackpressureMonitorMeta {
+            buffer_capacity: self.config.buffer_size,
+            usage_percent: calculate_usage_percent(usage, self.config.buffer_size),
+            state: self.state(),
+        }
+    }
+
+    /// Update state based on current usage.
+    fn update_state(&self) -> BackpressureState {
+        let usage = self.buffer_usage.load(Ordering::Acquire);
+        let usage_percent = calculate_usage_percent(usage, self.config.buffer_size) as u32;
+        let (next_state, changed) =
+            apply_watermark_transition(&self.in_high_watermark, usage, self.high_watermark_bytes, self.low_watermark_bytes);
+
+        if matches!(next_state, BackpressureState::HighWatermark) {
+            if changed {
+                counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1);
+
+                debug!(usage_percent, "Backpressure: entered high watermark");
+            }
+            BackpressureState::HighWatermark
+        } else {
+            if changed {
+                counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1);
+
+                debug!(usage_percent, "Backpressure: returned to normal");
+            }
+            BackpressureState::Normal
+        }
+    }
+}
+
+impl Default for BackpressureMonitor {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[cfg(test)]
+#[allow(unused_imports)]
+mod tests {
+    use super::{BackpressureMonitor, BackpressurePipe, BackpressureState, ObjectPipeBackpressurePolicy};
+
+    #[test]
+    fn test_backpressure_config_default() {
+        let config = ObjectPipeBackpressurePolicy::default();
+        assert_eq!(config.buffer_size, 4 * 1024 * 1024);
+        assert_eq!(config.high_watermark, 80);
+        assert_eq!(config.low_watermark, 50);
+    }
+
+    #[test]
+    fn test_backpressure_config_watermarks() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 1000,
+            high_watermark: 80,
+            low_watermark: 50,
+        };
+        assert_eq!(config.high_watermark_bytes(), 800);
+        assert_eq!(config.low_watermark_bytes(), 500);
+    }
+
+    #[test]
+    fn test_backpressure_policy_projects_to_concurrency_and_core_config() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let core = config.to_core_config();
+
+        assert_eq!(concurrency.buffer_size, config.buffer_size);
+        assert_eq!(concurrency.high_watermark, config.high_watermark);
+        assert_eq!(concurrency.low_watermark, config.low_watermark);
+        assert_eq!(core.high_water_mark, 0.75);
+        assert_eq!(core.low_water_mark, 0.40);
+        assert!(core.enabled);
+    }
+
+    #[test]
+    fn test_backpressure_pipe_consumes_concurrency_policy_thresholds() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let pipe = BackpressurePipe::with_config(config);
+
+        assert_eq!(pipe.capacity(), concurrency.buffer_size);
+        assert_eq!(pipe.high_watermark_bytes, concurrency.high_watermark_bytes());
+        assert_eq!(pipe.low_watermark_bytes, concurrency.low_watermark_bytes());
+    }
+
+    #[test]
+    fn test_backpressure_monitor_consumes_concurrency_policy_thresholds() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 2000,
+            high_watermark: 75,
+            low_watermark: 40,
+        };
+        let concurrency = config.to_concurrency_policy();
+        let monitor = BackpressureMonitor::with_config(config);
+
+        assert_eq!(monitor.meta().buffer_capacity, concurrency.buffer_size);
+        assert_eq!(monitor.high_watermark_bytes, concurrency.high_watermark_bytes());
+        assert_eq!(monitor.low_watermark_bytes, concurrency.low_watermark_bytes());
+    }
+
+    #[test]
+    fn test_backpressure_state_display() {
+        assert_eq!(format!("{}", BackpressureState::Normal), "normal");
+        assert_eq!(format!("{}", BackpressureState::HighWatermark), "high_watermark");
+        assert_eq!(format!("{}", BackpressureState::BackpressureApplied), "backpressure_applied");
+    }
+
+    #[test]
+    fn test_backpressure_monitor() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 1000,
+            high_watermark: 80,
+            low_watermark: 50,
+        };
+        let monitor = BackpressureMonitor::with_config(config);
+
+        // Initially normal
+        assert_eq!(monitor.state(), BackpressureState::Normal);
+        assert_eq!(monitor.meta().buffer_capacity, 1000);
+        assert_eq!(monitor.meta().usage_percent, 0.0);
+
+        // Write to reach high watermark
+        let state = monitor.on_write(850);
+        assert_eq!(state, BackpressureState::HighWatermark);
+        assert_eq!(monitor.meta().usage_percent, 85.0);
+
+        // Read to go below low watermark
+        let state = monitor.on_read(400);
+        assert_eq!(state, BackpressureState::Normal);
+        assert_eq!(monitor.meta().usage_percent, 45.0);
+    }
+
+    #[tokio::test]
+    async fn test_backpressure_pipe_creation() {
+        let pipe = BackpressurePipe::new();
+        assert_eq!(pipe.capacity(), 4 * 1024 * 1024);
+        assert_eq!(pipe.state(), BackpressureState::Normal);
+        assert_eq!(pipe.meta().buffer_capacity, 4 * 1024 * 1024);
+        assert!(pipe.meta().age <= pipe.age());
+    }
+
+    #[test]
+    fn test_backpressure_pipe_state_transitions() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 1000,
+            high_watermark: 80,
+            low_watermark: 50,
+        };
+        let pipe = BackpressurePipe::with_config(config);
+
+        assert_eq!(pipe.state(), BackpressureState::Normal);
+        assert_eq!(pipe.meta().state, BackpressureState::Normal);
+
+        pipe.record_write(850);
+        assert_eq!(pipe.state(), BackpressureState::BackpressureApplied);
+        assert_eq!(pipe.meta().state, BackpressureState::BackpressureApplied);
+
+        pipe.record_read(400);
+        assert_eq!(pipe.state(), BackpressureState::Normal);
+        assert_eq!(pipe.meta().state, BackpressureState::Normal);
+    }
+}
@@ -14,12 +14,21 @@

 //! I/O scheduling types for adaptive buffer sizing and load management.
 //!
-//! This is the live scheduling implementation. `rustfs_io_core` supplies the
-//! shared config shapes (`IoSchedulerConfig`, `IoPriorityQueueConfig`) that the
-//! types here project into through `to_core_config`, plus the `io_profile`
-//! storage-media model; bandwidth samples come from `rustfs_io_metrics`.
-//! Same-named io-core types are those config shapes, not a backing
-//! implementation this module delegates to.
+//! # Migration Note
+//!
+//! This module contains types that are also available in `rustfs_io_core`.
+//! For new code, prefer using types from `rustfs_io_core` directly:
+//!
+//! ```ignore
+//! // Recommended: Use io-core types
+//! use rustfs_io_core::{
+//!     IoLoadLevel, IoPriority, IoSchedulerConfig,
+//!     calculate_optimal_buffer_size, get_buffer_size_for_media,
+//! };
+//! ```
+//!
+//! This module remains for backward compatibility and provides additional
+//! runtime monitoring features (`IoPriorityMetrics`, `IoStrategyDebugInfo`).

 use rustfs_config::{KI_B, MI_B};
 use rustfs_io_core::io_profile::{AccessPattern, StorageMedia, StorageProfile};
@@ -1753,6 +1762,169 @@ impl<T> IoPriorityQueue<T> {
    }
 }

+// ============================================
+// I/O Priority Queue Metrics
+// ============================================
+
+/// Global metrics for I/O priority queue monitoring.
+///
+/// These metrics are emitted through the shared metrics pipeline and provide
+/// visibility into the priority queue behavior.
+#[allow(dead_code)]
+pub struct IoPriorityMetrics {
+    /// High priority queue depth.
+    pub high_queue_depth: AtomicU64,
+    /// Normal priority queue depth.
+    pub normal_queue_depth: AtomicU64,
+    /// Low priority queue depth.
+    pub low_queue_depth: AtomicU64,
+    /// High priority total wait time in nanoseconds.
+    pub high_wait_time_ns: AtomicU64,
+    /// Normal priority total wait time in nanoseconds.
+    pub normal_wait_time_ns: AtomicU64,
+    /// Low priority total wait time in nanoseconds.
+    pub low_wait_time_ns: AtomicU64,
+    /// Total starvation events count.
+    pub starvation_events: AtomicU64,
+    /// High priority requests processed.
+    pub high_processed: AtomicU64,
+    /// Normal priority requests processed.
+    pub normal_processed: AtomicU64,
+    /// Low priority requests processed.
+    pub low_processed: AtomicU64,
+}
+
+#[allow(dead_code)]
+impl Default for IoPriorityMetrics {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[allow(dead_code)]
+impl IoPriorityMetrics {
+    /// Create a new metrics instance.
+    pub const fn new() -> Self {
+        Self {
+            high_queue_depth: AtomicU64::new(0),
+            normal_queue_depth: AtomicU64::new(0),
+            low_queue_depth: AtomicU64::new(0),
+            high_wait_time_ns: AtomicU64::new(0),
+            normal_wait_time_ns: AtomicU64::new(0),
+            low_wait_time_ns: AtomicU64::new(0),
+            starvation_events: AtomicU64::new(0),
+            high_processed: AtomicU64::new(0),
+            normal_processed: AtomicU64::new(0),
+            low_processed: AtomicU64::new(0),
+        }
+    }
+
+    /// Update queue depths from status.
+    #[allow(dead_code)]
+    pub fn update_queue_depths(&self, status: &IoQueueStatus) {
+        self.high_queue_depth
+            .store(status.high_priority_waiting as u64, Ordering::Relaxed);
+        self.normal_queue_depth
+            .store(status.normal_priority_waiting as u64, Ordering::Relaxed);
+        self.low_queue_depth
+            .store(status.low_priority_waiting as u64, Ordering::Relaxed);
+    }
+
+    /// Record a starvation event.
+    #[allow(dead_code)]
+    pub fn record_starvation(&self) {
+        self.starvation_events.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record a processed request.
+    #[allow(dead_code)]
+    pub fn record_processed(&self, priority: IoPriority) {
+        match priority {
+            IoPriority::High => self.high_processed.fetch_add(1, Ordering::Relaxed),
+            IoPriority::Normal => self.normal_processed.fetch_add(1, Ordering::Relaxed),
+            IoPriority::Low => self.low_processed.fetch_add(1, Ordering::Relaxed),
+        };
+    }
+
+    /// Record wait time for a priority level.
+    pub fn record_wait_time(&self, priority: IoPriority, wait_ns: u64) {
+        match priority {
+            IoPriority::High => self.high_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
+            IoPriority::Normal => self.normal_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
+            IoPriority::Low => self.low_wait_time_ns.fetch_add(wait_ns, Ordering::Relaxed),
+        };
+    }
+
+    /// Get high priority queue depth.
+    pub fn get_high_queue_depth(&self) -> u64 {
+        self.high_queue_depth.load(Ordering::Relaxed)
+    }
+
+    /// Get normal priority queue depth.
+    pub fn get_normal_queue_depth(&self) -> u64 {
+        self.normal_queue_depth.load(Ordering::Relaxed)
+    }
+
+    /// Get low priority queue depth.
+    pub fn get_low_queue_depth(&self) -> u64 {
+        self.low_queue_depth.load(Ordering::Relaxed)
+    }
+
+    /// Get total starvation events.
+    pub fn get_starvation_events(&self) -> u64 {
+        self.starvation_events.load(Ordering::Relaxed)
+    }
+
+    /// Get metrics summary for logging/debugging.
+    pub fn summary(&self) -> String {
+        format!(
+            "high_queue={}, normal_queue={}, low_queue={}, starvation={}, high_proc={}, normal_proc={}, low_proc={}",
+            self.get_high_queue_depth(),
+            self.get_normal_queue_depth(),
+            self.get_low_queue_depth(),
+            self.get_starvation_events(),
+            self.high_processed.load(Ordering::Relaxed),
+            self.normal_processed.load(Ordering::Relaxed),
+            self.low_processed.load(Ordering::Relaxed)
+        )
+    }
+}
+
+/// Global I/O priority metrics instance.
+#[allow(dead_code)]
+pub static IO_PRIORITY_METRICS: IoPriorityMetrics = IoPriorityMetrics::new();
+
+/// Get optimized buffer size for I/O operations.
+///
+/// This function provides adaptive buffer sizing based on:
+/// - File size (small files get smaller buffers)
+/// - Concurrent request count (high concurrency gets smaller buffers)
+/// - Base buffer size from configuration
+///
+/// # Arguments
+///
+/// * `file_size` - Size of the file being read/written (-1 for unknown)
+///
+/// # Returns
+///
+/// Optimal buffer size in bytes
+///
+/// # Example
+///
+/// ```ignore
+/// let buffer_size = get_buffer_size_opt_in(1024 * 1024); // 1MB file
+/// assert!(buffer_size >= 64 * 1024); // At least 64KB
+/// ```
+#[allow(dead_code)]
+pub fn get_buffer_size_opt_in(file_size: i64) -> usize {
+    // Get base buffer size from configuration
+    let base_buffer_size =
+        rustfs_utils::get_env_usize(rustfs_config::ENV_OBJECT_IO_BUFFER_SIZE, rustfs_config::DEFAULT_OBJECT_IO_BUFFER_SIZE);
+
+    // Apply concurrency-aware adjustments
+    get_concurrency_aware_buffer_size(file_size, base_buffer_size)
+}
+
 // ============================================
 // Unit Tests
 // ============================================
@@ -1761,12 +1933,13 @@ impl<T> IoPriorityQueue<T> {
 #[allow(unused_imports)]
 mod tests {
    use super::{
-        IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig, IoSchedulingContext, IoStrategy,
-        get_advanced_buffer_size, get_concurrency_aware_buffer_size,
+        IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoSchedulerConfig,
+        IoSchedulingContext, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size,
    };
    use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
    use rustfs_io_metrics::bandwidth::{BandwidthSnapshot, BandwidthTier};
    use serial_test::serial;
+    use std::sync::atomic::Ordering;
    use std::time::Duration;

    #[tokio::test]
@@ -1953,6 +2126,29 @@ mod tests {
        assert_eq!(config.starvation_threshold_secs, 120);
    }

+    #[tokio::test]
+    #[serial]
+    async fn test_io_priority_metrics() {
+        let metrics = IoPriorityMetrics::new();
+
+        // Test initial state
+        assert_eq!(metrics.get_high_queue_depth(), 0);
+        assert_eq!(metrics.get_normal_queue_depth(), 0);
+        assert_eq!(metrics.get_low_queue_depth(), 0);
+        assert_eq!(metrics.get_starvation_events(), 0);
+
+        // Test recording
+        metrics.record_starvation();
+        assert_eq!(metrics.get_starvation_events(), 1);
+
+        metrics.record_processed(IoPriority::High);
+        metrics.record_processed(IoPriority::High);
+        metrics.record_processed(IoPriority::Normal);
+
+        assert_eq!(metrics.high_processed.load(Ordering::Relaxed), 2);
+        assert_eq!(metrics.normal_processed.load(Ordering::Relaxed), 1);
+    }
+
    // ============================================
    // Multi-Factor Strategy Tests
    // ============================================
@@ -65,11 +65,6 @@ pub struct ConcurrencyManager {
    bandwidth_monitor: Arc<Mutex<BandwidthMonitor>>,
    /// Metrics collector for I/O latency tracking (P50, P95, P99)
    metrics_collector: Arc<MetricsCollector>,
-    /// Experimental fixed-count foreground PutObject admission gate.
-    put_admission_semaphore: Arc<Semaphore>,
-    put_admission_enabled: bool,
-    put_admission_limit: usize,
-    put_admission_wait_timeout: Duration,
 }

 impl std::fmt::Debug for ConcurrencyManager {
@@ -119,18 +114,6 @@ pub enum DiskReadAdmission {
    Rejected,
 }

-/// Outcome of foreground PutObject request admission.
-#[derive(Debug)]
-pub enum PutObjectAdmission {
-    /// Foreground PUT admission is disabled; proceed on the legacy path.
-    Disabled,
-    /// Request is admitted and must hold the permit until the store write
-    /// returns or the request fails before mutation.
-    Admitted(tokio::sync::OwnedSemaphorePermit),
-    /// The fixed-count gate stayed full until the configured wait timeout.
-    Rejected,
-}
-
 #[allow(dead_code)]
 impl ConcurrencyManager {
    /// Create a new concurrency manager with default settings
@@ -178,18 +161,6 @@ impl ConcurrencyManager {
        // Initialize metrics collector for I/O latency tracking
        // Keep 1000 samples for P95/P99 calculation
        let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000));
-        let put_admission_enabled = rustfs_utils::get_env_bool(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE,
-        );
-        let put_admission_limit = rustfs_utils::get_env_usize(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT,
-        );
-        let put_admission_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-        ));

        // Build queue config directly from scheduler config.
        let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config);
@@ -205,10 +176,6 @@ impl ConcurrencyManager {
            pattern_detector,
            bandwidth_monitor,
            metrics_collector,
-            put_admission_semaphore: Arc::new(Semaphore::new(if put_admission_enabled { put_admission_limit } else { 0 })),
-            put_admission_enabled,
-            put_admission_limit,
-            put_admission_wait_timeout,
        }
    }

@@ -232,16 +199,6 @@ impl ConcurrencyManager {
        self.degraded_read_semaphore.close();
    }

-    #[cfg(test)]
-    pub(crate) fn with_put_admission_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self {
-        let mut manager = Self::new();
-        manager.put_admission_semaphore = Arc::new(Semaphore::new(if enabled { limit } else { 0 }));
-        manager.put_admission_enabled = enabled;
-        manager.put_admission_limit = limit;
-        manager.put_admission_wait_timeout = wait_timeout;
-        manager
-    }
-
    /// Track a GetObject request
    pub fn track_request() -> GetObjectGuard {
        GetObjectGuard::new()
@@ -327,32 +284,6 @@ impl ConcurrencyManager {
        }
    }

-    /// Admit a foreground PutObject request under the experimental fixed-count gate.
-    ///
-    /// The default-off path returns [`PutObjectAdmission::Disabled`] without
-    /// touching the semaphore, preserving legacy behavior. When enabled, the
-    /// permit must be acquired before body ingest and held until the store write
-    /// returns, so saturated foreground writes can fail with `SlowDown` before
-    /// creating visible side effects.
-    pub async fn admit_put_object(&self) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
-        if !self.put_admission_enabled || self.put_admission_limit == 0 {
-            return Ok(PutObjectAdmission::Disabled);
-        }
-
-        if self.put_admission_wait_timeout.is_zero() {
-            return Ok(match self.put_admission_semaphore.clone().try_acquire_owned() {
-                Ok(permit) => PutObjectAdmission::Admitted(permit),
-                Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected,
-                Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected,
-            });
-        }
-
-        match tokio::time::timeout(self.put_admission_wait_timeout, self.put_admission_semaphore.clone().acquire_owned()).await {
-            Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)),
-            Err(_) => Ok(PutObjectAdmission::Rejected),
-        }
-    }
-
    // ============================================
    // Adaptive I/O Strategy Methods
    // ============================================
@@ -761,16 +692,8 @@ impl ConcurrencyManager {

    /// Get a read-only workload admission snapshot for foreground writes.
    pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot {
-        let (active, limit, hard_gate_enabled) = if self.put_admission_enabled && self.put_admission_limit > 0 {
-            (
-                self.put_admission_limit
-                    .saturating_sub(self.put_admission_semaphore.available_permits()),
-                self.put_admission_limit,
-                true,
-            )
-        } else {
-            (PutObjectGuard::concurrent_count(), self.scheduler_config.max_concurrent_reads, false)
-        };
+        let active = PutObjectGuard::concurrent_count();
+        let limit = self.scheduler_config.max_concurrent_reads;
        let state = if limit == 0 {
            AdmissionState::Disabled
        } else if active >= limit {
@@ -783,10 +706,7 @@ impl ConcurrencyManager {
            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));

        match state {
-            AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
-            AdmissionState::Saturated if hard_gate_enabled => {
-                admission.with_reason("foreground write admission permits exhausted")
-            }
+            AdmissionState::Disabled => admission.with_reason("foreground write pressure tracking disabled"),
            AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"),
            _ => admission,
        }
@@ -863,7 +783,7 @@ impl Default for ConcurrencyManager {
 mod integration_tests {
    use super::super::io_schedule::{IoLoadLevel, IoPriority};
    use super::super::request_guard::GetObjectGuard;
-    use super::{ConcurrencyManager, PutObjectAdmission};
+    use super::ConcurrencyManager;
    use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
    use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
    use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
@@ -960,71 +880,6 @@ mod integration_tests {
        crate::storage::concurrency::reset_active_put_requests();
    }

-    #[tokio::test]
-    #[serial]
-    async fn test_concurrency_manager_put_admission_disabled_does_not_touch_gate() {
-        let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO);
-
-        let admission = manager
-            .admit_put_object()
-            .await
-            .expect("disabled put admission must not close");
-
-        assert!(matches!(admission, PutObjectAdmission::Disabled));
-        assert_eq!(manager.put_admission_semaphore.available_permits(), 0);
-        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Open);
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn test_concurrency_manager_put_admission_rejects_when_limit_full() {
-        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
-
-        let first = manager.admit_put_object().await.expect("first put admission should acquire");
-        assert!(matches!(first, PutObjectAdmission::Admitted(_)));
-        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated);
-
-        let second = manager
-            .admit_put_object()
-            .await
-            .expect("full put admission gate should reject, not close");
-        assert!(matches!(second, PutObjectAdmission::Rejected));
-    }
-
-    #[tokio::test]
-    #[serial]
-    async fn test_concurrency_manager_put_admission_reuses_released_permit() {
-        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
-
-        let first = manager.admit_put_object().await.expect("first put admission should acquire");
-        drop(first);
-
-        let second = manager
-            .admit_put_object()
-            .await
-            .expect("released put admission permit should be reusable");
-        assert!(matches!(second, PutObjectAdmission::Admitted(_)));
-    }
-
-    #[tokio::test(start_paused = true)]
-    #[serial]
-    async fn test_concurrency_manager_put_admission_wait_timeout_rejects() {
-        let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(5));
-        let held = manager.admit_put_object().await.expect("first put admission should acquire");
-        let waiter_manager = manager.clone();
-
-        let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await });
-        tokio::task::yield_now().await;
-        tokio::time::advance(Duration::from_secs(5)).await;
-
-        let admission = waiter
-            .await
-            .expect("put admission waiter task must not panic")
-            .expect("put admission gate must stay open");
-        assert!(matches!(admission, PutObjectAdmission::Rejected));
-        drop(held);
-    }
-
    #[tokio::test]
    #[serial]
    async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {
@@ -24,14 +24,16 @@
 //! - **Concurrency Management**: Coordination of concurrent GetObject requests
 //! - **Request Tracking**: RAII guards for request lifecycle management
 //!
-//! # Relationship to the shared crates
+//! # Migration Note
 //!
-//! The scheduling algorithm lives in [`io_schedule`], not in `rustfs-io-core`:
-//! this module does not delegate to it. `rustfs-io-core` owns the shared
-//! config shapes and the `io_profile` storage-media model that [`io_schedule`]
-//! consumes, and `rustfs-io-metrics` owns bandwidth sampling and metric
-//! recording.
+//! Core algorithms have been migrated to `rustfs-io-core` and metrics to
+//! `rustfs-io-metrics`. This module maintains API compatibility while
+//! delegating to the new implementations.

+// Sub-modules
+// pub mod bandwidth_monitor; // Migrated to rustfs-io-metrics
+// pub mod global_metrics; // Migrated to rustfs-io-metrics
+// pub mod io_profile; // Migrated to rustfs-io-core
 pub mod io_schedule;
 pub mod manager;
 pub mod request_guard;
@@ -43,15 +45,34 @@ pub mod request_guard;
 // I/O scheduling types (from io_schedule.rs for backward compatibility)
 #[allow(unused_imports)]
 pub use io_schedule::{
-    IoLoadLevel, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy,
-    get_advanced_buffer_size, get_concurrency_aware_buffer_size, get_put_concurrency_aware_buffer_size,
+    IO_PRIORITY_METRICS, IoLoadLevel, IoPriority, IoPriorityMetrics, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus,
+    IoSchedulerConfig, IoStrategy, get_advanced_buffer_size, get_buffer_size_opt_in, get_concurrency_aware_buffer_size,
+    get_put_concurrency_aware_buffer_size,
 };

 // Request tracking
 pub use request_guard::{GetObjectGuard, PutObjectGuard};

 // Concurrency manager
-pub use manager::{ConcurrencyManager, DiskReadAdmission, PutObjectAdmission};
+pub use manager::{ConcurrencyManager, DiskReadAdmission};
+
+// ============================================
+// New Module Re-exports (for gradual migration)
+// ============================================
+
+// Re-export types from rustfs-io-core for convenience
+pub use rustfs_io_core::{
+    // Backpressure types
+    BackpressureMonitor,
+    // Deadlock detection types
+    DeadlockDetector,
+    // Scheduler types
+    IoScheduler,
+    // Lock optimization types
+    LockOptimizer,
+};
+
+// Re-export types from rustfs-io-metrics for convenience

 // ============================================
 // Helper Functions
@@ -62,8 +83,37 @@ pub fn get_concurrency_manager() -> &'static ConcurrencyManager {
    ConcurrencyManager::global()
 }

-/// Reset the active put requests counter (for testing).
-#[cfg(test)]
+/// Reset the active get requests counter (for testing).
+#[allow(dead_code)]
+pub fn reset_active_get_requests() {
+    io_schedule::ACTIVE_GET_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed);
+}
+
+#[allow(dead_code)]
 pub fn reset_active_put_requests() {
    io_schedule::ACTIVE_PUT_REQUESTS.store(0, std::sync::atomic::Ordering::Relaxed);
 }
+
+/// Create a new I/O scheduler with default configuration.
+#[allow(dead_code)]
+pub fn create_io_scheduler() -> IoScheduler {
+    IoScheduler::with_defaults()
+}
+
+/// Create a new backpressure monitor with default configuration.
+#[allow(dead_code)]
+pub fn create_backpressure_monitor() -> BackpressureMonitor {
+    BackpressureMonitor::with_defaults()
+}
+
+/// Create a new deadlock detector with default configuration.
+#[allow(dead_code)]
+pub fn create_deadlock_detector() -> DeadlockDetector {
+    DeadlockDetector::with_defaults()
+}
+
+/// Create a new lock optimizer with default configuration.
+#[allow(dead_code)]
+pub fn create_lock_optimizer() -> LockOptimizer {
+    LockOptimizer::with_defaults()
+}
@@ -14,15 +14,17 @@

 //! Integration tests for concurrent request fix.
 //!
-//! These tests verify that the timeout and deadlock detection mechanisms work
-//! correctly under high concurrency scenarios.
+//! These tests verify that the timeout, backpressure, and deadlock detection
+//! mechanisms work correctly under high concurrency scenarios.

 #[cfg(test)]
 mod tests {
+    use crate::storage::backpressure::{BackpressureMonitor, BackpressureState, ObjectPipeBackpressurePolicy};
    use crate::storage::concurrency::{IoLoadLevel, IoPriority};
    use crate::storage::deadlock_detector::{
        DeadlockDetector, LockInfo, LockType, RequestHangDetectionPolicy, RequestResourceTracker,
    };
+    use crate::storage::lock_optimizer::{LockOptimizeConfig, LockOptimizer, LockStats};
    use crate::storage::timeout_wrapper::{GetObjectTimeoutPolicy, RequestTimeoutWrapper, TimedGetObjectResult};
    use std::time::Duration;

@@ -112,6 +114,82 @@ mod tests {
        }
    }

+    // ============================================
+    // Backpressure Tests
+    // ============================================
+
+    #[test]
+    fn test_backpressure_config_defaults() {
+        let config = ObjectPipeBackpressurePolicy::default();
+        assert_eq!(config.buffer_size, 4 * 1024 * 1024); // 4MB
+        assert_eq!(config.high_watermark, 80);
+        assert_eq!(config.low_watermark, 50);
+    }
+
+    #[test]
+    fn test_backpressure_monitor_state_transitions() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 1000,
+            high_watermark: 80,
+            low_watermark: 50,
+        };
+        let monitor = BackpressureMonitor::with_config(config);
+
+        // Initially normal
+        assert_eq!(monitor.state(), BackpressureState::Normal);
+
+        // Write to reach high watermark
+        let state = monitor.on_write(850);
+        assert_eq!(state, BackpressureState::HighWatermark);
+
+        // Read to go below low watermark
+        let state = monitor.on_read(400);
+        assert_eq!(state, BackpressureState::Normal);
+    }
+
+    #[test]
+    fn test_backpressure_usage_percent() {
+        let config = ObjectPipeBackpressurePolicy {
+            buffer_size: 1000,
+            high_watermark: 80,
+            low_watermark: 50,
+        };
+        let monitor = BackpressureMonitor::with_config(config);
+
+        monitor.on_write(500);
+        assert!((monitor.usage_percent() - 50.0).abs() < 1.0);
+    }
+
+    // ============================================
+    // Lock Optimizer Tests
+    // ============================================
+
+    #[test]
+    fn test_lock_optimize_config_defaults() {
+        let config = LockOptimizeConfig::default();
+        assert!(config.enabled);
+        assert_eq!(config.acquire_timeout, Duration::from_secs(5));
+    }
+
+    #[test]
+    fn test_lock_stats_tracking() {
+        let stats = LockStats::new();
+
+        stats.record_acquire();
+        stats.record_early_release(Duration::from_millis(100));
+        stats.record_early_release(Duration::from_millis(200));
+
+        assert_eq!(stats.locks_acquired.load(std::sync::atomic::Ordering::Relaxed), 1);
+        assert_eq!(stats.locks_released_early.load(std::sync::atomic::Ordering::Relaxed), 2);
+        assert_eq!(stats.max_hold_time(), Duration::from_millis(200));
+    }
+
+    #[test]
+    fn test_lock_optimizer_creation() {
+        let optimizer = LockOptimizer::new();
+        assert!(optimizer.is_enabled());
+    }
+
    // ============================================
    // I/O Priority Tests
    // ============================================
@@ -0,0 +1,458 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Lock Optimization for GetObject Operations.
+//!
+//! This module provides optimized lock management for read operations,
+//! reducing lock contention by releasing locks early (after metadata read)
+//! rather than holding them for the entire data transfer duration.
+//!
+//! # Migration Note
+//!
+//! For new code, consider using `rustfs_io_core::LockOptimizer` which provides
+//! the same core functionality with better separation of concerns. This module
+//! remains for backward compatibility and storage-specific configuration.
+//!
+//! ```ignore
+//! // Recommended: Use io-core directly
+//! use rustfs_io_core::LockOptimizer;
+//! let optimizer = LockOptimizer::with_defaults();
+//! ```
+
+// Allow dead_code for public API that may be used by external modules or future features
+//! # Key Features
+//!
+//! - Early lock release after metadata read
+//! - Lock hold time monitoring
+//! - Configurable optimization (can be disabled for debugging)
+//! - Lock contention metrics emitted through the shared metrics pipeline
+//!
+//! # Architecture
+//!
+//! ```text
+//! Traditional:  [Acquire Lock] --> [Read Metadata] --> [Transfer Data] --> [Release Lock]
+//!                           |<------------------ Lock Held ------------------>|
+//!
+//! Optimized:    [Acquire Lock] --> [Read Metadata] --> [Release Lock] --> [Transfer Data]
+//!                           |<- Lock Held ->|
+//! ```
+
+use std::sync::Arc;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::time::{Duration, Instant};
+use tracing::debug;
+
+use metrics::histogram;
+
+/// Lock optimization configuration.
+#[derive(Debug, Clone)]
+pub struct LockOptimizeConfig {
+    /// Whether to enable lock optimization.
+    /// When enabled, read locks are released after metadata read.
+    /// When disabled, locks are held for the entire operation (traditional behavior).
+    pub enabled: bool,
+    /// Lock acquisition timeout.
+    pub acquire_timeout: Duration,
+}
+
+impl Default for LockOptimizeConfig {
+    fn default() -> Self {
+        Self {
+            enabled: rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE,
+            acquire_timeout: Duration::from_secs(rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT),
+        }
+    }
+}
+
+impl LockOptimizeConfig {
+    /// Load configuration from environment variables.
+    pub fn from_env() -> Self {
+        let enabled = rustfs_utils::get_env_bool(
+            rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE,
+            rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE,
+        );
+        let acquire_timeout = Duration::from_secs(rustfs_utils::get_env_u64(
+            rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+            rustfs_config::DEFAULT_OBJECT_LOCK_ACQUIRE_TIMEOUT,
+        ));
+
+        Self {
+            enabled,
+            acquire_timeout,
+        }
+    }
+}
+
+/// Statistics for lock optimization monitoring.
+#[derive(Debug, Default)]
+pub struct LockStats {
+    /// Total locks acquired.
+    pub locks_acquired: AtomicU64,
+    /// Total locks released early.
+    pub locks_released_early: AtomicU64,
+    /// Total lock hold time in microseconds.
+    pub total_hold_time_us: AtomicU64,
+    /// Maximum lock hold time in microseconds.
+    pub max_hold_time_us: AtomicU64,
+}
+
+impl LockStats {
+    /// Create new lock statistics.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    /// Record a lock acquisition.
+    pub fn record_acquire(&self) {
+        self.locks_acquired.fetch_add(1, Ordering::Relaxed);
+    }
+
+    /// Record an early lock release.
+    pub fn record_early_release(&self, hold_time: Duration) {
+        self.locks_released_early.fetch_add(1, Ordering::Relaxed);
+        self.record_hold_time(hold_time);
+    }
+
+    /// Record lock hold time.
+    fn record_hold_time(&self, hold_time: Duration) {
+        let hold_time_us = hold_time.as_micros() as u64;
+        self.total_hold_time_us.fetch_add(hold_time_us, Ordering::Relaxed);
+
+        // Update max hold time
+        let mut current_max = self.max_hold_time_us.load(Ordering::Relaxed);
+        while hold_time_us > current_max {
+            match self
+                .max_hold_time_us
+                .compare_exchange_weak(current_max, hold_time_us, Ordering::Relaxed, Ordering::Relaxed)
+            {
+                Ok(_) => break,
+                Err(actual) => current_max = actual,
+            }
+        }
+    }
+
+    /// Get average hold time.
+    pub fn avg_hold_time(&self) -> Duration {
+        let total = self.total_hold_time_us.load(Ordering::Relaxed);
+        let count = self.locks_released_early.load(Ordering::Relaxed);
+        total.checked_div(count).map(Duration::from_micros).unwrap_or(Duration::ZERO)
+    }
+
+    /// Get maximum hold time.
+    pub fn max_hold_time(&self) -> Duration {
+        Duration::from_micros(self.max_hold_time_us.load(Ordering::Relaxed))
+    }
+}
+
+/// Global lock statistics.
+static LOCK_STATS: std::sync::OnceLock<Arc<LockStats>> = std::sync::OnceLock::new();
+
+/// Get global lock statistics.
+pub fn get_lock_stats() -> Arc<LockStats> {
+    LOCK_STATS.get_or_init(|| Arc::new(LockStats::new())).clone()
+}
+
+/// An optimized lock guard that supports early release.
+///
+/// This wraps the actual lock guard and provides:
+/// - Early release capability (before drop)
+/// - Hold time tracking
+/// - Metrics reporting
+pub struct OptimizedLockGuard<G> {
+    /// The underlying lock guard.
+    guard: Option<G>,
+    /// When the lock was acquired.
+    acquire_time: Instant,
+    /// Whether the lock has been released.
+    released: bool,
+    /// Lock resource name (for logging).
+    resource: String,
+    /// Statistics reference.
+    stats: Arc<LockStats>,
+}
+
+impl<G> OptimizedLockGuard<G> {
+    /// Create a new optimized lock guard.
+    pub fn new(guard: G, resource: impl Into<String>) -> Self {
+        let stats = get_lock_stats();
+        stats.record_acquire();
+
+        Self {
+            guard: Some(guard),
+            acquire_time: Instant::now(),
+            released: false,
+            resource: resource.into(),
+            stats,
+        }
+    }
+
+    /// Get the lock hold time so far.
+    pub fn hold_time(&self) -> Duration {
+        self.acquire_time.elapsed()
+    }
+
+    /// Check if the lock has been released.
+    pub fn is_released(&self) -> bool {
+        self.released
+    }
+
+    /// Release the lock early (before drop).
+    ///
+    /// This is the key optimization: releasing the lock after
+    /// metadata read rather than waiting for the entire operation.
+    pub fn early_release(&mut self) {
+        if self.released {
+            return;
+        }
+
+        let hold_time = self.hold_time();
+        self.guard.take();
+        self.released = true;
+
+        self.stats.record_early_release(hold_time);
+
+        histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64());
+
+        debug!(
+            resource = %self.resource,
+            hold_time_ms = hold_time.as_millis(),
+            "Lock released early (optimization active)"
+        );
+    }
+
+    /// Get a reference to the underlying guard.
+    pub fn as_ref(&self) -> Option<&G> {
+        if self.released { None } else { self.guard.as_ref() }
+    }
+}
+
+impl<G> Drop for OptimizedLockGuard<G> {
+    fn drop(&mut self) {
+        if !self.released {
+            let hold_time = self.hold_time();
+            self.guard.take();
+            self.released = true;
+
+            self.stats.record_early_release(hold_time);
+
+            histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64());
+
+            debug!(
+                resource = %self.resource,
+                hold_time_ms = hold_time.as_millis(),
+                "Lock released on drop (normal release)"
+            );
+        }
+    }
+}
+
+/// A scope guard that releases a lock when it goes out of scope.
+///
+/// This is a simpler version of OptimizedLockGuard for cases
+/// where we just need RAII semantics without tracking.
+pub struct LockScopeGuard<G> {
+    guard: Option<G>,
+}
+
+impl<G> LockScopeGuard<G> {
+    /// Create a new scope guard.
+    pub fn new(guard: G) -> Self {
+        Self { guard: Some(guard) }
+    }
+
+    /// Release the lock early.
+    pub fn release(&mut self) {
+        self.guard.take();
+    }
+}
+
+impl<G> Drop for LockScopeGuard<G> {
+    fn drop(&mut self) {
+        self.guard.take();
+    }
+}
+
+/// Helper for managing lock optimization in GetObject operations.
+///
+/// This provides a clean interface for the common pattern:
+/// 1. Acquire lock
+/// 2. Read metadata
+/// 3. Release lock (if optimization enabled)
+/// 4. Transfer data (without lock)
+pub struct LockOptimizer {
+    /// Configuration.
+    config: LockOptimizeConfig,
+}
+
+impl LockOptimizer {
+    /// Create a new lock optimizer with default configuration.
+    pub fn new() -> Self {
+        Self {
+            config: LockOptimizeConfig::from_env(),
+        }
+    }
+
+    /// Create a new lock optimizer with custom configuration.
+    pub fn with_config(config: LockOptimizeConfig) -> Self {
+        Self { config }
+    }
+
+    /// Check if lock optimization is enabled.
+    pub fn is_enabled(&self) -> bool {
+        self.config.enabled
+    }
+
+    /// Get the lock acquisition timeout.
+    pub fn acquire_timeout(&self) -> Duration {
+        self.config.acquire_timeout
+    }
+
+    /// Wrap a lock guard for optimization.
+    pub fn wrap_guard<G>(&self, guard: G, resource: impl Into<String>) -> OptimizedLockGuard<G> {
+        OptimizedLockGuard::new(guard, resource)
+    }
+
+    /// Execute a metadata read operation with lock optimization.
+    ///
+    /// This is the main entry point for optimized lock usage:
+    /// - If optimization is enabled: lock is released after metadata_fn completes
+    /// - If optimization is disabled: lock is held until the returned guard is dropped
+    ///
+    /// # Arguments
+    ///
+    /// * `guard` - The lock guard to optimize
+    /// * `resource` - Resource name for logging
+    /// * `metadata_fn` - Function to read metadata while holding lock
+    ///
+    /// # Returns
+    ///
+    /// A tuple of (metadata result, optional guard to hold for later release)
+    pub async fn with_optimized_lock<G, F, Fut, T>(
+        &self,
+        guard: G,
+        resource: impl Into<String>,
+        metadata_fn: F,
+    ) -> (T, Option<OptimizedLockGuard<G>>)
+    where
+        F: FnOnce() -> Fut,
+        Fut: std::future::Future<Output = T>,
+    {
+        let resource = resource.into();
+        let mut optimized = OptimizedLockGuard::new(guard, &resource);
+
+        // Execute metadata read while holding lock
+        let result = metadata_fn().await;
+
+        if self.config.enabled {
+            // Release lock early
+            optimized.early_release();
+            (result, None)
+        } else {
+            // Keep lock for caller to release
+            (result, Some(optimized))
+        }
+    }
+}
+
+impl Default for LockOptimizer {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+/// Check if lock optimization is enabled globally.
+pub fn is_lock_optimization_enabled() -> bool {
+    rustfs_utils::get_env_bool(
+        rustfs_config::ENV_OBJECT_LOCK_OPTIMIZATION_ENABLE,
+        rustfs_config::DEFAULT_OBJECT_LOCK_OPTIMIZATION_ENABLE,
+    )
+}
+
+#[cfg(test)]
+#[allow(unused_imports)]
+mod tests {
+    use super::{LockOptimizeConfig, LockOptimizer, LockStats, OptimizedLockGuard};
+    use std::sync::Mutex;
+    use std::sync::atomic::Ordering;
+    use std::time::Duration;
+
+    #[test]
+    fn test_lock_optimize_config_default() {
+        let config = LockOptimizeConfig::default();
+        assert!(config.enabled);
+        assert_eq!(config.acquire_timeout, Duration::from_secs(5));
+    }
+
+    #[test]
+    fn test_lock_stats() {
+        let stats = LockStats::new();
+
+        stats.record_acquire();
+        stats.record_early_release(Duration::from_millis(100));
+        stats.record_early_release(Duration::from_millis(200));
+
+        assert_eq!(stats.locks_acquired.load(Ordering::Relaxed), 1);
+        assert_eq!(stats.locks_released_early.load(Ordering::Relaxed), 2);
+        assert_eq!(stats.max_hold_time(), Duration::from_millis(200));
+    }
+
+    #[test]
+    fn test_optimized_lock_guard() {
+        let guard = Mutex::new(42);
+        let locked = guard.lock().unwrap();
+        let mut optimized = OptimizedLockGuard::new(locked, "test-resource");
+
+        assert!(!optimized.is_released());
+        assert!(optimized.hold_time() < Duration::from_secs(1));
+
+        optimized.early_release();
+        assert!(optimized.is_released());
+    }
+
+    #[test]
+    fn test_lock_optimizer() {
+        let optimizer = LockOptimizer::new();
+        assert!(optimizer.is_enabled());
+    }
+
+    #[tokio::test]
+    async fn test_with_optimized_lock_enabled() {
+        let optimizer = LockOptimizer::new();
+        let guard = Mutex::new(42);
+        let locked = guard.lock().unwrap();
+
+        let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await;
+
+        assert_eq!(result, 100);
+        // With optimization enabled, guard should be None (released early)
+        assert!(returned_guard.is_none());
+    }
+
+    #[tokio::test]
+    async fn test_with_optimized_lock_disabled() {
+        let config = LockOptimizeConfig {
+            enabled: false,
+            acquire_timeout: Duration::from_secs(5),
+        };
+        let optimizer = LockOptimizer::with_config(config);
+        let guard = Mutex::new(42);
+        let locked = guard.lock().unwrap();
+
+        let (result, returned_guard) = optimizer.with_optimized_lock(locked, "test-resource", || async { 100 }).await;
+
+        assert_eq!(result, 100);
+        // With optimization disabled, guard should be Some (held for later)
+        assert!(returned_guard.is_some());
+    }
+}
@@ -13,10 +13,12 @@
 // limitations under the License.

 pub mod access;
+pub mod backpressure;
 pub mod concurrency;
 pub mod deadlock_detector;
 pub mod ecfs;
 pub(crate) mod helper;
+pub mod lock_optimizer;
 pub mod options;
 pub mod request_context;
 pub mod rpc;
@@ -585,7 +585,6 @@ mod tests {
        let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode");
        assert_eq!(decoded.info().bitrot_start_cycle, 9);
        assert_eq!(decoded.operations.queue_length, 2);
-        assert_eq!(decoded.operations.queued_by_source.mrf, 0);
    }

    #[test]
@@ -117,7 +117,7 @@ pub(crate) mod access_consumer {

 pub(crate) mod concurrency_consumer {
    pub(crate) use super::super::concurrency::{
-        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectAdmission, PutObjectGuard,
+        ConcurrencyManager, DiskReadAdmission, GetObjectGuard, IoQueueStatus, IoStrategy, PutObjectGuard,
        get_concurrency_aware_buffer_size, get_concurrency_manager, get_put_concurrency_aware_buffer_size,
    };
 }