fix(ci): use English headers in the upgrade matrix table

ci(upgrade): render an upgrade matrix in the report; fix from default
The upgrade report only ever showed the requested deb URLs and a case table. The nightly chain runs died installing the OLD package (default from_version 1.0.0-rc.4-preview.1 has no .deb asset on its release, and release 1.0.0-rc.4 ships none either), leaving an empty Total: 0 report with no indication of what was upgraded. - Default from_version is now 1.0.0-rc.3 (ships rustfs_1.0.0.rc.3_amd64.deb). Matches the auto-testing default from PR #32. - The report generator also parses the [UPG-TOPO] lines the suite now emits and renders an 'Upgrade Matrix' section: per topology and KMS backend, the versions actually in place before/after (captured via 'rustfs --version' on the node) and the aggregated result. When the suite dies before any topology completes, the matrix says so instead of silently showing nothing.
2026-09-04 19:25:40 +00:00 · 2026-09-04 19:21:11 +08:00 · 2026-09-04 19:11:39 +08:00
36 changed files with 1071 additions and 5713 deletions
@@ -1,2 +1,2 @@
 sha256-darwin=a881fd7d3f5cb94654221ca85b8b30cce1b95e608824a55a15339cbc294e6d34
-sha256-linux=a2933d83dfe74ffa03410a0959333a1c48288b8469ca9f17273d449d7510c24b
+sha256-linux=e9a8d64e73f627c4d26c236dbbba690c9ee03a9e26d42a4244515b4439365535
@@ -69,7 +69,7 @@ filter = 'package(rustfs-ecstore) & test(/^(bucket::lifecycle::bucket_lifecycle_
 setup = 'ecstore-large-stack'

 [[profile.default.scripts]]
-filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
+filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

 [[profile.default.scripts]]
@@ -210,7 +210,7 @@ filter = 'package(rustfs-ecstore) & test(/^(bucket::lifecycle::bucket_lifecycle_
 setup = 'ecstore-large-stack'

 [[profile.ci.scripts]]
-filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
+filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'

 [[profile.ci.scripts]]
@@ -18,9 +18,9 @@ on:
  workflow_dispatch:
    inputs:
      from_version:
-        description: 'OLD RustFS release tag (e.g. 1.0.0-rc.4-preview.1)'
+        description: 'OLD RustFS release tag (must ship a .deb asset, e.g. 1.0.0-rc.3)'
        required: false
-        default: '1.0.0-rc.4-preview.1'
+        default: '1.0.0-rc.3'
      from_url:
        description: 'OLD .deb URL. Overrides from_version.'
        required: false
@@ -203,54 +203,75 @@ jobs:
            TO_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
          fi
          CASE_TABLE="/tmp/rustfs-upgrade-cases.md"
-          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          MATRIX_TABLE="/tmp/rustfs-upgrade-matrix.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" "${MATRIX_TABLE}" <<'PY'
          import re
          import sys

-          log_file, out_file = sys.argv[1], sys.argv[2]
+          log_file, out_file, matrix_file = sys.argv[1], sys.argv[2], sys.argv[3]
          ansi = re.compile(r'\x1b\[[0-9;]*m')
          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+          topo_re = re.compile(
+              r'^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$')

          rows = []
          index = {}
+          topo_rows = []
          try:
-            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
-              for raw in fh:
-                line = ansi.sub('', raw).strip()
-                m = start_re.match(line)
-                if m:
-                  case_id, name = m.group(1), m.group(2)
-                  if case_id not in index:
-                    index[case_id] = len(rows)
-                    rows.append([case_id, name, 'RUNNING'])
-                  continue
-                m = done_re.match(line)
-                if m:
-                  status, case_id = m.group(1), m.group(2)
-                  if case_id in index:
-                    rows[index[case_id]][2] = status
-                  else:
-                    rows.append([case_id, case_id, status])
-                    index[case_id] = len(rows) - 1
+              with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+                  for raw in fh:
+                      line = ansi.sub('', raw).strip()
+                      m = topo_re.match(line)
+                      if m:
+                          topo_rows.append(m.groups())
+                          continue
+                      m = start_re.match(line)
+                      if m:
+                          case_id, name = m.group(1), m.group(2)
+                          if case_id not in index:
+                              index[case_id] = len(rows)
+                              rows.append([case_id, name, 'RUNNING'])
+                          continue
+                      m = done_re.match(line)
+                      if m:
+                          status, case_id = m.group(1), m.group(2)
+                          if case_id in index:
+                              rows[index[case_id]][2] = status
+                          else:
+                              rows.append([case_id, case_id, status])
+                              index[case_id] = len(rows) - 1
          except FileNotFoundError:
-            rows = []
+              rows = []

          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
          for _, _, status in rows:
-            counts[status] = counts.get(status, 0) + 1
+              counts[status] = counts.get(status, 0) + 1

          with open(out_file, 'w', encoding='utf-8') as out:
-            out.write('## Case Summary\n\n')
-            out.write(f"- Total: {len(rows)}\\n")
-            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
-            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
-            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
-            out.write('\\n')
-            out.write('| Case | Name | Status |\\n')
-            out.write('| --- | --- | --- |\\n')
-            for case_id, name, status in rows:
-              out.write(f'| {case_id} | {name} | {status} |\\n')
+              out.write('## Case Summary\n\n')
+              out.write(f"- Total: {len(rows)}\\n")
+              out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+              out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+              out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+              out.write('\\n')
+              out.write('| Case | Name | Status |\\n')
+              out.write('| --- | --- | --- |\\n')
+              for case_id, name, status in rows:
+                  out.write(f'| {case_id} | {name} | {status} |\\n')
+
+          # Upgrade matrix: one row per topology/backend with the versions
+          # captured on the nodes (rustfs --version) and the aggregated
+          # result. The dashboard renders this table directly.
+          with open(matrix_file, 'w', encoding='utf-8') as out:
+              out.write('## Upgrade Matrix\n\n')
+              out.write('| Topology | KMS Backend | From Version | To Version | Result |\n')
+              out.write('| --- | --- | --- | --- | --- |\n')
+              for topo, backend, old_v, new_v, npass, nfail in topo_rows:
+                  result = 'PASS' if nfail == '0' else 'FAIL'
+                  out.write(f'| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) |\n')
+              if not topo_rows:
+                  out.write('| - | - | - | - | NOT RUN (suite failed before upgrade) |\n')
          PY
          {
            echo "# RustFS upgrade compatibility report"
@@ -261,6 +282,8 @@ jobs:
            echo "- To: ${TO_SOURCE}"
            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
            echo ""
+            cat "${MATRIX_TABLE}" || true
+            echo ""
            cat "${CASE_TABLE}" || true
            echo ""
            echo "## Log tail"
@@ -1328,9 +1328,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.16.0"
+version = "1.15.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9c054752dd9e4dc73d0b75748c99ac2d0feafbf2f25c7b0516f03a3534161223"
+checksum = "954c563ce84507722d2679f07a35d21b9c6466b3872d513020d0281fc8112ac9"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-runtime-api-macros",
@@ -1368,9 +1368,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-types"
-version = "1.6.3"
+version = "1.6.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8f94d16e797ec62cd999fc9d5942b48fa7050c3093ddadff48e4d7528d16fcb9"
+checksum = "fce83ce9abbb198d25bc7131e468d0f9fe1257125e58c39f3f9fc9f5098c9647"
 dependencies = [
 "base64-simd",
 "bytes",
@@ -5103,9 +5103,9 @@ dependencies = [

 [[package]]
 name = "hickory-net"
-version = "0.26.2"
+version = "0.26.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "084e7bd6a377435d568f652153e571b50970d7ccc1d1eeec0519f834632287e1"
+checksum = "e2295ed2f9c31e471e1428a8f88a3f0e1f4b27c15049592138d1eebe9c35b183"
 dependencies = [
 "async-trait",
 "cfg-if",
@@ -5127,9 +5127,9 @@ dependencies = [

 [[package]]
 name = "hickory-proto"
-version = "0.26.2"
+version = "0.26.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7e2da0694c15b44c6f68a6b05e0233617008c54080e31d6eb848d858a9c5b38d"
+checksum = "0bab31817bfb44672a252e97fe81cd0c18d1b2cf892108922f6818820df8c643"
 dependencies = [
 "data-encoding",
 "idna",
@@ -5147,9 +5147,9 @@ dependencies = [

 [[package]]
 name = "hickory-resolver"
-version = "0.26.2"
+version = "0.26.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0e4f9f4603319422d482738f3f6fe5aac03157fdbfed1cd85a3ff45adb09072f"
+checksum = "f0d58d28879ceecde6607729660c2667a081ccdc082e082675042793960f178c"
 dependencies = [
 "cfg-if",
 "futures-util",
@@ -5226,9 +5226,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.25.0"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2ec7782e005cabd5eaf350febde384cd799faa3a0e624587aa8759c240e0b592"
+checksum = "e2645642a23d4061ec15a4a6e74f851a3145c3125356846cfc7772ff9c6f2737"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5240,6 +5240,7 @@ dependencies = [
 "futures-util",
 "hdrhistogram",
 "hotpath-macros",
+ "hotpath-meta",
 "http 1.5.0",
 "libc",
 "object 0.36.7",
@@ -5259,15 +5260,30 @@ dependencies = [

 [[package]]
 name = "hotpath-macros"
-version = "0.25.0"
+version = "0.24.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "929b2285d2cd21b2733a7fb6ebc843bb4f83dbd1db0122f5f9ebb9567b1e2613"
+checksum = "89a3d3cdf9b0d4d3d4f6d4a29798f3b9170401ba500eaa58dd8f890f926af0f1"
 dependencies = [
 "proc-macro2",
 "quote",
 "syn 2.0.119",
 ]

+[[package]]
+name = "hotpath-macros-meta"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e84cd2417fa60938241cf1cd6c03e09953f5c821122dc5da9b8f27975d136c5b"
+
+[[package]]
+name = "hotpath-meta"
+version = "0.24.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4d2c145b67b1a4e7bcefa918995e212c30a49a85f05cc5962fe1f717878d560b"
+dependencies = [
+ "hotpath-macros-meta",
+]
+
 [[package]]
 name = "htmlescape"
 version = "0.3.1"
@@ -5626,6 +5642,18 @@ dependencies = [
 "tempfile",
 ]

+[[package]]
+name = "internal-russh-num-bigint"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ae8e22120c32fb4d19ec55fba35015f57095cd95a2e3b732e44457f5915b2ee8"
+dependencies = [
+ "num-integer",
+ "num-traits",
+ "rand 0.10.2",
+ "rand_core 0.10.1",
+]
+
 [[package]]
 name = "io-uring"
 version = "0.7.14"
@@ -5876,20 +5904,17 @@ dependencies = [

 [[package]]
 name = "kafka-protocol"
-version = "0.18.0"
+version = "0.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "099d5c2f1b40cd830cbf18ca4d2a0805f2875b811ca18f0372b2433e68fe2dda"
+checksum = "66292444a1cd4d430d450d472c30cba839d0724229aba2d79affffcf901516e2"
 dependencies = [
 "anyhow",
 "bytes",
 "crc",
 "crc32c",
- "flate2",
 "indexmap 2.14.1",
- "lz4",
- "snap",
+ "paste",
 "uuid",
- "zstd",
 ]

 [[package]]
@@ -6898,8 +6923,6 @@ checksum = "93e7820bc0a80a0238e650327316f929ba18d5be054b647490a3a6a339f3e7c0"
 dependencies = [
 "num-integer",
 "num-traits",
- "rand 0.10.2",
- "rand_core 0.10.1",
 ]

 [[package]]
@@ -7492,9 +7515,9 @@ dependencies = [

 [[package]]
 name = "pageant"
-version = "0.2.3"
+version = "0.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d8eab09a361a4afe0b1668be978cd97e4f052e927a92b0b608cf902965d49ce"
+checksum = "3adadc44070da6f464b0918655a12f5792c156e088d8c4082d13e27d94c3e791"
 dependencies = [
 "base16ct 1.0.0",
 "byteorder",
@@ -7583,6 +7606,12 @@ dependencies = [
 "phc",
 ]

+[[package]]
+name = "paste"
+version = "1.0.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57c0d7b74b563b49d38dae00a0c37d4d6de9b432382b2892f0574ddcae73fd0a"
+
 [[package]]
 name = "path-absolutize"
 version = "4.0.1"
@@ -9221,9 +9250,9 @@ dependencies = [

 [[package]]
 name = "russh"
-version = "0.63.2"
+version = "0.63.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8e134e2480f4e86f83e4aa45b4c0a9723f84beaffa694c54bdf056e74efdd7dd"
+checksum = "35bab1b87d915817d5d9cc352637cd40d5f0b298a48c6309af9156a4addc3031"
 dependencies = [
 "aes 0.9.3",
 "aws-lc-rs",
@@ -9252,12 +9281,13 @@ dependencies = [
 "hex-literal",
 "hmac 0.13.0",
 "inout 0.2.2",
+ "internal-russh-num-bigint",
 "keccak",
 "log",
 "md5",
 "ml-kem",
 "module-lattice",
- "num-bigint 0.5.1",
+ "num-bigint 0.4.8",
 "p256 0.14.0",
 "p384 0.14.0",
 "p521",
@@ -9948,11 +9978,11 @@ dependencies = [

 [[package]]
 name = "rustfs-kafka"
-version = "1.3.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "759f8ddf709b497006e0f422a89890f11ac86c84c2c39df282891477c22f1a13"
+checksum = "4eee0644a99743fb2f51db7fbae1a6ca2d85f064daf7595784eaa52834a68c96"
 dependencies = [
- "base64 0.23.1",
+ "base64 0.22.1",
 "bytes",
 "fnv",
 "hmac 0.13.0",
@@ -9962,22 +9992,22 @@ dependencies = [
 "pbkdf2 0.13.0",
 "rand 0.10.2",
 "rustls",
+ "rustls-native-certs",
 "sha2 0.11.0",
 "socket2",
 "thiserror 2.0.20",
 "tracing",
 "twox-hash",
- "uuid",
 "webpki-roots 1.0.9",
 ]

 [[package]]
 name = "rustfs-kafka-async"
-version = "1.3.1"
+version = "1.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5aa574a13d91ef529e3a08f059a5c3ee8d4c3b16cf62f62c72595a6dc4f79f8f"
+checksum = "7cd1997c3116cb94ede80d9a0b828f46dd27386cc93825fce141a92eb3aa9630"
 dependencies = [
- "base64 0.23.1",
+ "base64 0.22.1",
 "bytes",
 "hmac 0.13.0",
 "kafka-protocol",
@@ -9986,11 +10016,11 @@ dependencies = [
 "rand 0.10.2",
 "rustfs-kafka",
 "rustls",
+ "rustls-native-certs",
 "sha2 0.11.0",
 "tokio",
 "tokio-rustls",
 "tracing",
- "uuid",
 "webpki-roots 1.0.9",
 ]

@@ -10164,18 +10194,18 @@ dependencies = [

 [[package]]
 name = "rustfs-mimalloc"
-version = "0.5.3"
+version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "46a7b69356718defa4060de3059e609c9da6619c35924fefef55d902ccd94958"
+checksum = "85d1a75bd188260754c4fa8ace1bb1f6a9956ee36c1d330f98fed5d25e39fa15"
 dependencies = [
 "rustfs-mimalloc-sys",
 ]

 [[package]]
 name = "rustfs-mimalloc-sys"
-version = "0.5.3"
+version = "0.5.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1dd3ec9b7e7b9fed453acd8b4e32713ddc0e01e1aee31a3deec5b8025c0880c3"
+checksum = "5d3adbf24cbe37f040f856c04f8d4baf75626eacd4e8b17eb6b3d79de7fde692"
 dependencies = [
 "cc",
 ]
@@ -12271,7 +12301,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",
@@ -12455,9 +12485,9 @@ dependencies = [

 [[package]]
 name = "tinyvec"
-version = "1.13.2"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4cf0ded5c4e56918d8f8a339e1bb67d038d3bc6d144ac407904015ba2e4cde9b"
+checksum = "bb4ebadaa0af04fab11ae01eb5f9fdb5f9c5b875506e210e71c07873528baa7f"
 dependencies = [
 "tinyvec_macros",
 ]
@@ -13622,14 +13652,13 @@ checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"

 [[package]]
 name = "wnaf"
-version = "0.14.1"
+version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "795ca18b3fdb5e62bf982199278341ddcf7ebf7d32e25e212ad05d496e95f6fa"
+checksum = "ab12e7090f27e2ffd9322651492942d50c2926094af30601e1964337db39daf1"
 dependencies = [
 "ff 0.14.0",
 "group 0.14.0",
 "hybrid-array",
- "primefield",
 ]

 [[package]]
@@ -165,7 +165,7 @@ http-body = "1.1.0"
 http-body-util = "0.1.5"
 minlz = "1.2.3"
 reqwest = "0.13.4"
-rustfs-kafka-async = { version = "1.3.1" }
+rustfs-kafka-async = { version = "1.2.0" }
 socket2 = { version = "0.6.5" }
 tokio = { version = "1.53.1" }
 tokio-rustls = { default-features = false, version = "0.26.4" }
@@ -244,8 +244,8 @@ aws-sdk-kms = { default-features = false, version = "1.117.0" }
 aws-sdk-s3 = { default-features = false, version = "1.144.0" }
 aws-sdk-sts = { default-features = false, version = "1.113.0" }
 aws-smithy-http-client = { default-features = false, version = "1.4.0" }
-aws-smithy-runtime-api = { version = "1.16.0" }
-aws-smithy-types = { version = "1.6.3" }
+aws-smithy-runtime-api = { version = "1.15.0" }
+aws-smithy-types = { version = "1.6.2" }
 base64-simd = "0.8.0"
 brotli = "9.0.0"
 clap = { version = "4.6.6" }
@@ -359,15 +359,15 @@ libunftp = { version = "0.23.0" }
 unftp-core = "0.1.0"
 suppaftp = { version = "11.0.0" }
 rcgen = { version = "0.14.10", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
-russh = { version = "0.63.2" }
+russh = { version = "0.63.1" }
 russh-sftp = "2.4.0"

 # WebDAV
 dav-server = "0.11.0"

 # Performance Analysis and Memory Profiling
-rustfs-mimalloc = { version = "0.5.3" }
-hotpath = { version = "0.25.0", default-features = false }
+rustfs-mimalloc = { version = "0.5.2" }
+hotpath = { version = "0.24.0", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

@@ -252,13 +252,6 @@ pub mod bucket {
        };
    }

-    pub mod sealed_credentials {
-        pub use crate::bucket::sealed_credentials::{
-            CredentialSealer, SEALED_CREDENTIAL_VERSION, SealScope, SealedCredential, SealedCredentialError,
-            SealedCredentialStore, credential_sealer, install_credential_sealer, seal_secret, unseal_secret,
-        };
-    }
-
    pub mod replication {
        pub use crate::bucket::replication::replication_pool::{
            DurableMrfBacklogSummary, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBacklogObservabilitySummary,
@@ -87,12 +87,16 @@ use rustfs_filemeta::{
 use rustfs_scanner_metrics::metrics::{
    IlmAction, Metrics, ScannerLifecycleExpiryStateUpdate, ScannerLifecycleTransitionStateUpdate, global_metrics,
 };
-use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::parse_bool};
+use rustfs_utils::{
+    get_env_i64, get_env_usize,
+    path::encode_dir_object,
+    string::{parse_bool, strings_has_prefix_fold},
+};
 use s3s::dto::{
    BucketLifecycleConfiguration, ExpirationStatus, ObjectLockConfiguration, RestoreRequest, RestoreRequestType, RestoreStatus,
    Timestamp,
 };
-use s3s::header::X_AMZ_RESTORE;
+use s3s::header::{X_AMZ_RESTORE, X_AMZ_SERVER_SIDE_ENCRYPTION};
 use sha2::{Digest, Sha256};
 use std::any::Any;
 use std::collections::{BTreeMap, HashMap, HashSet};
@@ -161,6 +165,7 @@ pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
 )]
 pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
+pub const AMZ_ENCRYPTION_AES: &str = "AES256";
 #[allow(
    dead_code,
    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
@@ -4797,24 +4802,24 @@ fn attach_tier_operation_lease(mut reader: GetObjectReader, lease: TierOperation
    reader
 }

-/// Resolve the RestoreObject request options.
-///
-/// Returns the typed [`StorageError`]: flattening these into an opaque
-/// `io::Error` string erased the identity the S3 layer needs to answer
-/// InvalidArgument instead of a generic 500 (backlog#2205).
-pub async fn post_restore_opts(version_id: &str, bucket: &str, object: &str) -> Result<ObjectOptions, Error> {
+pub async fn post_restore_opts(version_id: &str, bucket: &str, object: &str) -> Result<ObjectOptions, std::io::Error> {
    let versioned = BucketVersioningSys::prefix_enabled(bucket, object).await;
    let version_suspended = BucketVersioningSys::prefix_suspended(bucket, object).await;
    let vid = version_id.trim();
    if !vid.is_empty() && vid != NULL_VERSION_ID {
        if let Err(_err) = Uuid::parse_str(vid) {
-            return Err(StorageError::InvalidVersionID(bucket.to_string(), object.to_string(), vid.to_string()));
+            return Err(std::io::Error::other(
+                StorageError::InvalidVersionID(bucket.to_string(), object.to_string(), vid.to_string()).to_string(),
+            ));
        }
        if !versioned && !version_suspended {
-            return Err(StorageError::InvalidArgument(
-                bucket.to_string(),
-                object.to_string(),
-                format!("version-id specified {vid} but versioning is not enabled on {bucket}"),
+            return Err(std::io::Error::other(
+                StorageError::InvalidArgument(
+                    bucket.to_string(),
+                    object.to_string(),
+                    format!("version-id specified {} but versioning is not enabled on {}", vid, bucket),
+                )
+                .to_string(),
            ));
        }
    }
@@ -4867,18 +4872,43 @@ pub async fn put_restore_opts(
    }
    meta.insert(X_AMZ_STORAGE_CLASS.as_str().to_lowercase(), sc);*/

-    // A SELECT restore must never reach the restore writer: the caller writes
-    // the retrieved bytes back to the source bucket/object, so building
-    // SELECT output options here produced a source overwrite carrying only
-    // the OutputLocation metadata instead of a write to `OutputLocation.S3`
-    // (backlog#1341). RestoreObject rejects SELECT at the API boundary; this
-    // is the fail-closed backstop for any other caller.
-    if rreq
-        .type_
-        .as_ref()
-        .is_some_and(|type_| type_.as_str() == RestoreRequestType::SELECT)
+    if let Some(type_) = &rreq.type_
+        && type_.as_str() == RestoreRequestType::SELECT
    {
-        return Err(std::io::Error::other("SELECT restore requests are not supported"));
+        let Some(s3) = select_restore_s3_location(rreq)? else {
+            return Err(std::io::Error::other("OutputLocation.S3 required for SELECT requests"));
+        };
+        if let Some(user_metadata) = s3.user_metadata.as_ref() {
+            for metadata in user_metadata {
+                let name = metadata
+                    .name
+                    .as_deref()
+                    .ok_or_else(|| std::io::Error::other("SELECT restore metadata name is required"))?;
+                let value = metadata.value.clone().unwrap_or_default();
+                if strings_has_prefix_fold(name, "x-amz-meta") {
+                    meta.insert(name.to_string(), value);
+                } else {
+                    meta.insert(format!("x-amz-meta-{name}"), value);
+                }
+            }
+        }
+        if let Some(tags) = &s3.tagging {
+            meta.insert(
+                AMZ_OBJECT_TAGGING.to_string(),
+                serde_urlencoded::to_string(tags.tag_set.clone()).unwrap_or_else(|_| "".to_string()),
+            );
+        }
+        if let Some(encryption) = &s3.encryption
+            && encryption.encryption_type.as_str() != ""
+        {
+            meta.insert(X_AMZ_SERVER_SIDE_ENCRYPTION.as_str().to_string(), AMZ_ENCRYPTION_AES.to_string());
+        }
+        return Ok(ObjectOptions {
+            versioned: BucketVersioningSys::prefix_enabled(bucket, object).await,
+            version_suspended: BucketVersioningSys::prefix_suspended(bucket, object).await,
+            user_defined: meta,
+            ..Default::default()
+        });
    }
    for (k, v) in oi.user_defined.iter() {
        meta.insert(k.to_string(), v.clone());
@@ -31,7 +31,6 @@ pub mod policy_sys;
 pub mod quota;
 pub mod remote_s3_client;
 pub mod replication;
-pub mod sealed_credentials;
 pub mod tagging;
 pub mod target;
 pub mod utils;
@@ -14,9 +14,9 @@

 //! Outbound client for an on-demand migration source bucket.
 //!
-//! `SourceClient` maps local keys onto a read-only `SourceBackend`. The
-//! S3 backend uses the shared remote builder and exposes the surface the
-//! migration path needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
+//! `SourceClient` wraps an `aws_sdk_s3::Client` built through the shared
+//! remote builder and exposes the read-only surface the migration path
+//! needs (HEAD, ranged streaming GET, ListObjectsV2, GetObjectTagging, a
 //! probe for admin validation). Every request carries the
 //! `source-proxy-request` anti-loop marker in both the `x-rustfs-` and
 //! `x-minio-` prefixes so a RustFS/MinIO source answers locally instead of
@@ -511,7 +511,7 @@ pub struct SourceObject {
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourcePage {
    pub objects: Vec<SourceObject>,
-    /// Rolled-up prefixes, in the same namespace as `objects`; always empty when the
+    /// Rolled-up prefixes, in the local namespace; always empty when the
    /// request carried no delimiter.
    pub common_prefixes: Vec<String>,
    pub is_truncated: bool,
@@ -519,8 +519,7 @@ pub struct SourcePage {
 }

 /// One `ListObjectsV2` page request against the source. Keys are given in the
-/// local namespace at `SourceClient`, and in the source namespace at
-/// `SourceBackend`; `SourceClient` maps them through `source_prefix`.
+/// local namespace; `SourceClient` maps them through `source_prefix`.
 #[derive(Clone, Debug, Default, PartialEq, Eq)]
 pub struct SourceListRequest<'a> {
    pub prefix: Option<&'a str>,
@@ -576,30 +575,8 @@ impl Intercept for SourceProxyMarkerInterceptor {
    }
 }

-/// Read-only provider operations in the source bucket namespace.
-///
-/// Implementations must preserve streaming, honor the requested range and
-/// pagination cursor, and classify failures without including credentials.
-/// `SourceClient` owns prefix mapping so every provider shares the same local
-/// namespace. Continuation tokens are opaque and must never be prefix-mapped.
-#[async_trait::async_trait]
-pub trait SourceBackend: Send + Sync {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError>;
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError>;
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError>;
-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError>;
-    /// Verify bucket access; `SourceClient` separately probes a filtered listing.
-    async fn probe(&self) -> Result<(), SourceError>;
-}
-
-/// S3-compatible implementation, including request signing and anti-loop headers.
-pub struct S3SourceBackend {
-    client: S3Client,
-    bucket: String,
-}
-
 pub struct SourceClient {
-    backend: Box<dyn SourceBackend>,
+    client: S3Client,
    endpoint: String,
    bucket: String,
    source_prefix: Option<String>,
@@ -632,10 +609,7 @@ impl SourceClient {
    fn from_config_builder(config: aws_sdk_s3::config::Builder, endpoint: String, spec: &SourceClientSpec) -> Self {
        let client = S3Client::from_conf(config.interceptor(SourceProxyMarkerInterceptor::new()).build());
        Self {
-            backend: Box::new(S3SourceBackend {
-                client,
-                bucket: spec.bucket.clone(),
-            }),
+            client,
            endpoint,
            bucket: spec.bucket.clone(),
            source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
@@ -678,15 +652,35 @@ impl SourceClient {
    }

    pub async fn head_object(&self, key: &str) -> Result<SourceHead, SourceError> {
-        self.backend.head(&self.source_key(key)).await
+        let output = self
+            .client
+            .head_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_head_from_head_output(output)
    }

-    /// Streams the object, preserving an optional HTTP byte range.
+    /// Streams the object; `range` is passed through as an HTTP `Range`
+    /// header and omitted entirely when `None`.
    pub async fn get_object(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        self.backend.get(&self.source_key(key), range).await
+        let range = range.map(range_header_value).transpose()?;
+        let output = self
+            .client
+            .get_object()
+            .bucket(&self.bucket)
+            .key(self.source_key(key))
+            .set_range(range)
+            .send()
+            .await
+            .map_err(classify_sdk_error)?;
+        source_get_from_output(output)
    }

-    /// Lists one page under the local prefix.
+    /// Lists one page under the local `prefix`. Keys are returned in the
+    /// local namespace; entries outside `source_prefix` are skipped.
    pub async fn list_objects_v2(
        &self,
        prefix: Option<&str>,
@@ -702,81 +696,9 @@ impl SourceClient {
        .await
    }

-    /// Maps keys and common prefixes while leaving opaque cursors untouched.
+    /// [`Self::list_objects_v2`] with the delimiter and start-after the
+    /// list-through merge needs (rustfs/backlog#2164).
    pub async fn list_page(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
-        let prefix = self.source_key(request.prefix.unwrap_or_default());
-        let start_after = request.start_after.map(|key| self.source_key(key));
-        let mut page = self
-            .backend
-            .list(&SourceListRequest {
-                prefix: Some(&prefix),
-                start_after: start_after.as_deref(),
-                ..*request
-            })
-            .await?;
-        page.objects = page
-            .objects
-            .into_iter()
-            .filter_map(|object| self.local_object(object))
-            .collect();
-        page.common_prefixes = page
-            .common_prefixes
-            .into_iter()
-            .filter_map(|prefix| self.local_key(&prefix).map(str::to_string))
-            .collect();
-        Ok(page)
-    }
-
-    fn local_object(&self, mut object: SourceObject) -> Option<SourceObject> {
-        object.key = self.local_key(&object.key)?.to_string();
-        Some(object)
-    }
-
-    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
-        self.backend.tagging(&self.source_key(key)).await
-    }
-
-    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
-        self.backend.probe().await?;
-        let page = self.list_objects_v2(None, None, 1).await?;
-        Ok(SourceProbe {
-            sample_object: page.objects.into_iter().next(),
-            has_more_objects: page.is_truncated,
-        })
-    }
-}
-
-#[async_trait::async_trait]
-impl SourceBackend for S3SourceBackend {
-    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
-        let output = self
-            .client
-            .head_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_head_from_head_output(output)
-    }
-
-    /// Streams the object; `range` is passed through as an HTTP `Range`
-    /// header and omitted entirely when `None`.
-    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
-        let range = range.map(range_header_value).transpose()?;
-        let output = self
-            .client
-            .get_object()
-            .bucket(&self.bucket)
-            .key(key)
-            .set_range(range)
-            .send()
-            .await
-            .map_err(classify_sdk_error)?;
-        source_get_from_output(output)
-    }
-
-    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
        // `start_after` is silently ignored by S3 once a continuation token is
        // present; refuse the ambiguous pair rather than list from the wrong
        // position.
@@ -789,9 +711,9 @@ impl SourceBackend for S3SourceBackend {
            .client
            .list_objects_v2()
            .bucket(&self.bucket)
-            .prefix(request.prefix.unwrap_or_default())
+            .prefix(self.source_key(request.prefix.unwrap_or_default()))
            .set_delimiter(request.delimiter.map(str::to_string))
-            .set_start_after(request.start_after.map(str::to_string))
+            .set_start_after(request.start_after.map(|after| self.source_key(after)))
            .set_continuation_token(request.continuation_token.map(str::to_string))
            .max_keys(request.max_keys)
            .send()
@@ -809,13 +731,13 @@ impl SourceBackend for S3SourceBackend {
            .contents
            .unwrap_or_default()
            .into_iter()
-            .filter_map(s3_source_object)
+            .filter_map(|object| self.source_object(object))
            .collect();
        let common_prefixes = output
            .common_prefixes
            .unwrap_or_default()
            .into_iter()
-            .filter_map(|prefix| prefix.prefix)
+            .filter_map(|prefix| Some(self.local_key(prefix.prefix.as_deref()?)?.to_string()))
            .collect();

        Ok(SourcePage {
@@ -826,43 +748,48 @@ impl SourceBackend for S3SourceBackend {
        })
    }

-    async fn tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
+    fn source_object(&self, object: SdkObject) -> Option<SourceObject> {
+        let key = self.local_key(object.key.as_deref()?)?.to_string();
+        let etag = normalize_etag(object.e_tag);
+        let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
+        Some(SourceObject {
+            key,
+            etag,
+            size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
+            last_modified: system_time(object.last_modified),
+            storage_class: object.storage_class.map(|class| class.as_str().to_string()),
+            is_multipart_etag,
+        })
+    }
+
+    pub async fn get_object_tagging(&self, key: &str) -> Result<HashMap<String, String>, SourceError> {
        let output = self
            .client
            .get_object_tagging()
            .bucket(&self.bucket)
-            .key(key)
+            .key(self.source_key(key))
            .send()
            .await
            .map_err(classify_sdk_error)?;
        Ok(output.tag_set.into_iter().map(|tag| (tag.key, tag.value)).collect())
    }

-    async fn probe(&self) -> Result<(), SourceError> {
+    /// Admin validation: HeadBucket plus a one-key listing under the prefix.
+    pub async fn probe(&self) -> Result<SourceProbe, SourceError> {
        self.client
            .head_bucket()
            .bucket(&self.bucket)
            .send()
            .await
            .map_err(classify_sdk_error)?;
-        Ok(())
+        let page = self.list_objects_v2(None, None, 1).await?;
+        Ok(SourceProbe {
+            sample_object: page.objects.into_iter().next(),
+            has_more_objects: page.is_truncated,
+        })
    }
 }

-fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
-    let key = object.key?;
-    let etag = normalize_etag(object.e_tag);
-    let is_multipart_etag = etag.as_deref().is_some_and(is_multipart_etag);
-    Some(SourceObject {
-        key,
-        etag,
-        size: object.size.and_then(|size| u64::try_from(size).ok()).unwrap_or(0),
-        last_modified: system_time(object.last_modified),
-        storage_class: object.storage_class.map(|class| class.as_str().to_string()),
-        is_multipart_etag,
-    })
-}
-
 #[cfg(test)]
 mod tests {
    use super::*;
@@ -1267,68 +1194,6 @@ mod tests {
        assert!(requests[1].uri.contains("continuation-token=token-1"), "{}", requests[1].uri);
    }

-    #[tokio::test]
-    async fn list_page_maps_delimiter_prefixes_and_start_after_but_not_cursors() {
-        let body = r#"<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
-<IsTruncated>true</IsTruncated><NextContinuationToken>data/opaque</NextContinuationToken>
-<CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
-<CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
-</ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
-        let first = client
-            .list_page(&SourceListRequest {
-                prefix: Some("photos/"),
-                delimiter: Some("/"),
-                start_after: Some("photos/a"),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("delimiter listing should succeed");
-        assert_eq!(first.common_prefixes, vec!["photos/"]);
-        assert_eq!(first.next_continuation_token.as_deref(), Some("data/opaque"));
-        let second = client
-            .list_page(&SourceListRequest {
-                continuation_token: first.next_continuation_token.as_deref(),
-                max_keys: 2,
-                ..Default::default()
-            })
-            .await
-            .expect("opaque continuation should succeed");
-        assert_eq!(second.common_prefixes, first.common_prefixes);
-        let requests = recorded(&requests);
-        let query = |request: &RecordedRequest| {
-            Url::parse(&request.uri)
-                .expect("request URI")
-                .query_pairs()
-                .into_owned()
-                .collect::<HashMap<_, _>>()
-        };
-        let first_query = query(&requests[0]);
-        assert_eq!(first_query.get("prefix").map(String::as_str), Some("data/photos/"));
-        assert_eq!(first_query.get("start-after").map(String::as_str), Some("data/photos/a"));
-        assert_eq!(first_query.get("delimiter").map(String::as_str), Some("/"));
-        let second_query = query(&requests[1]);
-        assert_eq!(second_query.get("continuation-token").map(String::as_str), Some("data/opaque"));
-        assert!(!second_query.contains_key("start-after"));
-    }
-
-    #[tokio::test]
-    async fn list_page_rejects_ambiguous_cursor_before_sending() {
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![]).await;
-        let err = client
-            .list_page(&SourceListRequest {
-                start_after: Some("a"),
-                continuation_token: Some("opaque"),
-                max_keys: 1,
-                ..Default::default()
-            })
-            .await
-            .expect_err("ambiguous list position must fail");
-        assert!(matches!(err, SourceError::Other(_)));
-        assert!(recorded(&requests).is_empty(), "invalid request must never reach the source");
-    }
-
    #[tokio::test]
    async fn list_objects_v2_rejects_truncated_page_without_token() {
        let (client, _) = scripted_client(&spec(None), vec![ok(Vec::new(), LIST_TRUNCATED_WITHOUT_TOKEN)]).await;
@@ -1492,14 +1357,11 @@ mod tests {

    fn prefix_client(prefix: Option<String>) -> SourceClient {
        SourceClient {
-            backend: Box::new(S3SourceBackend {
-                client: S3Client::from_conf(
-                    aws_sdk_s3::Config::builder()
-                        .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-                        .build(),
-                ),
-                bucket: "bucket".to_string(),
-            }),
+            client: S3Client::from_conf(
+                aws_sdk_s3::Config::builder()
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
            endpoint: "https://source.example.com".to_string(),
            bucket: "bucket".to_string(),
            source_prefix: prefix.filter(|prefix| !prefix.is_empty()),
@@ -1,351 +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.
-
-//! Sealed remote credentials shared by the three stores that persist one
-//! (rustfs/backlog#2168): replication targets (`bucket-targets.json`), remote
-//! tiers (`tier-config.bin`) and on-demand migration sources
-//! (`on-demand-migration.json`).
-//!
-//! The design record is `docs/architecture/remote-credential-sealing-adr.md`.
-//! What this module owns: the versioned envelope, the encryption context that
-//! binds a ciphertext to the record owning it, the sealer registration point,
-//! and the fail-closed error type. What it deliberately does not own: any KMS
-//! call (ECStore does not depend on `rustfs-kms`; the binary installs a
-//! sealer, exactly like `ON_DEMAND_MIGRATION_CONFIG_HOOK` and the event
-//! dispatch hook in `crates/ecstore/src/services/event_notification.rs`), and
-//! any decision about which stored field a consumer writes.
-
-use async_trait::async_trait;
-use serde::{Deserialize, Serialize};
-use std::collections::HashMap;
-use std::fmt;
-use std::sync::{Arc, OnceLock};
-
-/// Envelope format this build writes. A reader accepts only versions it
-/// knows; an unknown version is a typed error, never a fallback.
-pub const SEALED_CREDENTIAL_VERSION: u8 = 1;
-
-/// Which store a sealed value belongs to. Part of the encryption context, so
-/// a ciphertext cannot be replayed into a different store.
-#[derive(Clone, Copy, Debug, PartialEq, Eq)]
-pub enum SealedCredentialStore {
-    /// `bucket-targets.json` (replication and other bucket targets).
-    BucketTargets,
-    /// `tier-config.bin` (remote tiers).
-    TierConfig,
-    /// `on-demand-migration.json` (migration sources).
-    OnDemandMigration,
-}
-
-impl SealedCredentialStore {
-    pub fn as_str(self) -> &'static str {
-        match self {
-            SealedCredentialStore::BucketTargets => "bucket-targets",
-            SealedCredentialStore::TierConfig => "tier-config",
-            SealedCredentialStore::OnDemandMigration => "on-demand-migration",
-        }
-    }
-}
-
-/// Identity of the record a secret belongs to: the store, its owner (bucket
-/// name, tier name, or target ARN) and the field name. Rendered into the KMS
-/// encryption context so a ciphertext moved between buckets, tiers or fields
-/// fails to decrypt instead of silently authorizing a different remote.
-#[derive(Clone, Debug, PartialEq, Eq)]
-pub struct SealScope {
-    pub store: SealedCredentialStore,
-    pub owner: String,
-    pub field: &'static str,
-}
-
-impl SealScope {
-    pub fn new(store: SealedCredentialStore, owner: impl Into<String>, field: &'static str) -> Self {
-        Self {
-            store,
-            owner: owner.into(),
-            field,
-        }
-    }
-
-    /// The encryption context handed to the sealer. Keys are stable: they are
-    /// part of the on-disk contract, because a ciphertext only decrypts under
-    /// the same context.
-    pub fn encryption_context(&self) -> HashMap<String, String> {
-        HashMap::from([
-            ("rustfs:store".to_string(), self.store.as_str().to_string()),
-            ("rustfs:owner".to_string(), self.owner.clone()),
-            ("rustfs:field".to_string(), self.field.to_string()),
-        ])
-    }
-}
-
-/// A sealed secret as persisted. `Debug` prints no ciphertext: a sealed value
-/// is not a secret, but it is noise in a log line and an operator reading one
-/// should see the key it is wrapped under, not the bytes.
-#[derive(Clone, Serialize, Deserialize, PartialEq, Eq)]
-pub struct SealedCredential {
-    /// Envelope version; see [`SEALED_CREDENTIAL_VERSION`].
-    pub v: u8,
-    /// KMS master key id the data key is wrapped under.
-    pub key_id: String,
-    /// Master key version, when the backend reports one. Carried so the KMS
-    /// re-wrap job (`docs/architecture/kms-bulk-rekey-contract.md`) can tell
-    /// stale envelopes apart; nothing here rotates on its own.
-    #[serde(default, skip_serializing_if = "Option::is_none")]
-    pub key_version: Option<String>,
-    /// Algorithm label reported by the sealer, for forensics and migration.
-    pub alg: String,
-    /// Ciphertext blob as produced by the sealer, base64 (standard, padded)
-    /// in the JSON stores and raw inside the tier msgpack payload.
-    pub ct: String,
-}
-
-impl fmt::Debug for SealedCredential {
-    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
-        f.debug_struct("SealedCredential")
-            .field("v", &self.v)
-            .field("key_id", &self.key_id)
-            .field("key_version", &self.key_version)
-            .field("alg", &self.alg)
-            .field("ct", &format_args!("<{} bytes sealed>", self.ct.len()))
-            .finish()
-    }
-}
-
-impl SealedCredential {
-    /// Rejects an envelope this build cannot read. Called before every
-    /// unseal so an unknown version fails here rather than inside a backend.
-    pub fn check_version(&self) -> Result<(), SealedCredentialError> {
-        if self.v == SEALED_CREDENTIAL_VERSION {
-            Ok(())
-        } else {
-            Err(SealedCredentialError::UnsupportedVersion(self.v))
-        }
-    }
-}
-
-/// Why a seal or unseal did not produce a usable value. Every variant is
-/// terminal for the record that carried it: a caller reports the remote as
-/// unusable, and never substitutes a default or empty credential.
-#[derive(Clone, Debug, PartialEq, Eq, thiserror::Error)]
-pub enum SealedCredentialError {
-    /// No sealer is installed: KMS is not configured, or the process has not
-    /// finished startup. Reading a sealed record is impossible here.
-    #[error("no credential sealer is installed")]
-    NoSealer,
-    /// The stored envelope is from a newer (or otherwise unknown) format.
-    #[error("unsupported sealed credential envelope version {0}")]
-    UnsupportedVersion(u8),
-    /// The stored bytes are not a well-formed envelope.
-    #[error("malformed sealed credential: {0}")]
-    Malformed(String),
-    /// The sealer refused: wrong encryption context, missing key, revoked
-    /// access, or a failed authentication tag.
-    #[error("sealed credential could not be unsealed: {0}")]
-    Kms(String),
-}
-
-/// The KMS-backed half, installed by the binary at startup.
-#[async_trait]
-pub trait CredentialSealer: Send + Sync + 'static {
-    /// Wraps `plaintext` under the scope's encryption context.
-    async fn seal(&self, plaintext: &str, scope: &SealScope) -> Result<SealedCredential, SealedCredentialError>;
-
-    /// Unwraps a stored envelope. Must fail when the envelope was sealed
-    /// under a different scope.
-    async fn unseal(&self, sealed: &SealedCredential, scope: &SealScope) -> Result<String, SealedCredentialError>;
-}
-
-static CREDENTIAL_SEALER: OnceLock<Arc<dyn CredentialSealer>> = OnceLock::new();
-
-/// Installs the process-wide sealer. Returns `false` when one is already
-/// installed, matching the other ECStore hooks.
-pub fn install_credential_sealer(sealer: Arc<dyn CredentialSealer>) -> bool {
-    CREDENTIAL_SEALER.set(sealer).is_ok()
-}
-
-/// The installed sealer, or `None` when KMS is not wired. Callers that only
-/// need to know whether sealing is possible use this; callers that must have
-/// it use [`seal_secret`] / [`unseal_secret`] and get the typed error.
-pub fn credential_sealer() -> Option<Arc<dyn CredentialSealer>> {
-    CREDENTIAL_SEALER.get().cloned()
-}
-
-/// Seals one secret field. Fails closed: without a sealer the caller must
-/// reject the write rather than persist the secret in clear text after the
-/// operator asked for sealing.
-pub async fn seal_secret(plaintext: &str, scope: &SealScope) -> Result<SealedCredential, SealedCredentialError> {
-    let sealer = credential_sealer().ok_or(SealedCredentialError::NoSealer)?;
-    sealer.seal(plaintext, scope).await
-}
-
-/// Unseals one secret field, rejecting an unknown envelope version first.
-pub async fn unseal_secret(sealed: &SealedCredential, scope: &SealScope) -> Result<String, SealedCredentialError> {
-    sealed.check_version()?;
-    let sealer = credential_sealer().ok_or(SealedCredentialError::NoSealer)?;
-    sealer.unseal(sealed, scope).await
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use parking_lot::Mutex;
-
-    /// Stands in for the KMS-backed sealer: records the context it was called
-    /// with, and refuses a ciphertext presented under a different one.
-    #[derive(Default)]
-    struct FakeSealer {
-        sealed_contexts: Mutex<Vec<HashMap<String, String>>>,
-    }
-
-    #[async_trait]
-    impl CredentialSealer for FakeSealer {
-        async fn seal(&self, plaintext: &str, scope: &SealScope) -> Result<SealedCredential, SealedCredentialError> {
-            let context = scope.encryption_context();
-            self.sealed_contexts.lock().push(context.clone());
-            let mut bound = serde_json::to_string(&context).expect("context serializes");
-            bound.push('|');
-            bound.push_str(plaintext);
-            Ok(SealedCredential {
-                v: SEALED_CREDENTIAL_VERSION,
-                key_id: "key-1".to_string(),
-                key_version: Some("3".to_string()),
-                alg: "AES-256-GCM".to_string(),
-                ct: base64_simd::STANDARD.encode_to_string(bound.as_bytes()),
-            })
-        }
-
-        async fn unseal(&self, sealed: &SealedCredential, scope: &SealScope) -> Result<String, SealedCredentialError> {
-            let raw = base64_simd::STANDARD
-                .decode_to_vec(sealed.ct.as_bytes())
-                .map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
-            let bound = String::from_utf8(raw).map_err(|err| SealedCredentialError::Malformed(err.to_string()))?;
-            let expected = serde_json::to_string(&scope.encryption_context()).expect("context serializes");
-            bound
-                .strip_prefix(&expected)
-                .and_then(|rest| rest.strip_prefix('|'))
-                .map(str::to_string)
-                .ok_or_else(|| SealedCredentialError::Kms("encryption context mismatch".to_string()))
-        }
-    }
-
-    fn scope(owner: &str) -> SealScope {
-        SealScope::new(SealedCredentialStore::OnDemandMigration, owner, "secret_key")
-    }
-
-    #[tokio::test]
-    async fn seal_round_trips_and_binds_the_scope() {
-        let sealer = Arc::new(FakeSealer::default());
-        let sealed = sealer.seal("super-secret", &scope("photos")).await.expect("seal");
-        assert_eq!(sealed.v, SEALED_CREDENTIAL_VERSION);
-        assert_eq!(sealed.key_version.as_deref(), Some("3"));
-        assert_eq!(sealer.unseal(&sealed, &scope("photos")).await.expect("unseal"), "super-secret");
-
-        // The same ciphertext under another bucket must not unseal.
-        let err = sealer
-            .unseal(&sealed, &scope("other-bucket"))
-            .await
-            .expect_err("a ciphertext must not move between owners");
-        assert!(matches!(err, SealedCredentialError::Kms(_)), "{err}");
-
-        // Nor under another field of the same record.
-        let other_field = SealScope::new(SealedCredentialStore::OnDemandMigration, "photos", "session_token");
-        let err = sealer
-            .unseal(&sealed, &other_field)
-            .await
-            .expect_err("a ciphertext must not move between fields");
-        assert!(matches!(err, SealedCredentialError::Kms(_)), "{err}");
-
-        let contexts = sealer.sealed_contexts.lock();
-        assert_eq!(contexts.len(), 1);
-        assert_eq!(contexts[0]["rustfs:store"], "on-demand-migration");
-        assert_eq!(contexts[0]["rustfs:owner"], "photos");
-        assert_eq!(contexts[0]["rustfs:field"], "secret_key");
-    }
-
-    #[tokio::test]
-    async fn an_unknown_envelope_version_is_rejected_before_the_sealer_is_asked() {
-        let sealed = SealedCredential {
-            v: SEALED_CREDENTIAL_VERSION + 1,
-            key_id: "key-1".to_string(),
-            key_version: None,
-            alg: "AES-256-GCM".to_string(),
-            ct: "Zm9v".to_string(),
-        };
-        assert_eq!(
-            sealed.check_version().expect_err("a newer envelope must not be read"),
-            SealedCredentialError::UnsupportedVersion(SEALED_CREDENTIAL_VERSION + 1)
-        );
-        // The global helper reports the version, not "no sealer", even in a
-        // process where none is installed.
-        assert_eq!(
-            unseal_secret(&sealed, &scope("photos")).await.expect_err("version first"),
-            SealedCredentialError::UnsupportedVersion(SEALED_CREDENTIAL_VERSION + 1)
-        );
-    }
-
-    #[tokio::test]
-    async fn without_a_sealer_both_directions_fail_closed() {
-        // This test binary installs no sealer, so the global helpers must
-        // report NoSealer rather than fall back to clear text.
-        assert!(credential_sealer().is_none(), "no sealer is installed in unit tests");
-        assert_eq!(
-            seal_secret("super-secret", &scope("photos")).await.expect_err("seal"),
-            SealedCredentialError::NoSealer
-        );
-        let sealed = SealedCredential {
-            v: SEALED_CREDENTIAL_VERSION,
-            key_id: "key-1".to_string(),
-            key_version: None,
-            alg: "AES-256-GCM".to_string(),
-            ct: "Zm9v".to_string(),
-        };
-        assert_eq!(
-            unseal_secret(&sealed, &scope("photos")).await.expect_err("unseal"),
-            SealedCredentialError::NoSealer
-        );
-    }
-
-    #[test]
-    fn debug_and_serde_keep_the_on_disk_shape_stable() {
-        let sealed = SealedCredential {
-            v: 1,
-            key_id: "key-1".to_string(),
-            key_version: None,
-            alg: "AES-256-GCM".to_string(),
-            ct: "Zm9v".to_string(),
-        };
-        // key_version is omitted when absent, so an envelope from a backend
-        // without version history stays compact.
-        assert_eq!(
-            serde_json::to_string(&sealed).expect("serialize"),
-            r#"{"v":1,"key_id":"key-1","alg":"AES-256-GCM","ct":"Zm9v"}"#
-        );
-        let parsed: SealedCredential = serde_json::from_str(r#"{"v":1,"key_id":"key-1","alg":"AES-256-GCM","ct":"Zm9v"}"#)
-            .expect("an envelope without key_version parses");
-        assert_eq!(parsed, sealed);
-
-        let rendered = format!("{sealed:?}");
-        assert!(rendered.contains("key-1"), "{rendered}");
-        assert!(!rendered.contains("Zm9v"), "Debug must not print the ciphertext: {rendered}");
-    }
-
-    #[test]
-    fn a_malformed_envelope_is_a_typed_error() {
-        let err = serde_json::from_str::<SealedCredential>(r#"{"v":1,"key_id":"key-1"}"#)
-            .map_err(|err| SealedCredentialError::Malformed(err.to_string()))
-            .expect_err("a truncated envelope must not parse");
-        assert!(matches!(err, SealedCredentialError::Malformed(_)), "{err}");
-    }
-}
@@ -18,203 +18,25 @@
 #![allow(unused_must_use)]
 #![allow(clippy::all)]

-use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
+use serde::{Deserialize, Serialize};

-#[derive(Serialize, Deserialize, Default, Clone)]
-#[serde(default)]
-pub struct TierServicePrincipalAuth {
-    #[serde(rename = "TenantID", alias = "tenantID", alias = "tenant_id")]
-    pub tenant_id: String,
-    #[serde(rename = "ClientID", alias = "clientID", alias = "client_id")]
-    pub client_id: String,
-    #[serde(rename = "ClientSecret", alias = "clientSecret", alias = "client_secret")]
-    pub client_secret: String,
-}
-
-impl TierServicePrincipalAuth {
-    pub(crate) fn is_empty(&self) -> bool {
-        self.tenant_id.is_empty() && self.client_id.is_empty() && self.client_secret.is_empty()
-    }
-}
-
-impl std::fmt::Debug for TierServicePrincipalAuth {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("TierServicePrincipalAuth")
-            .field("tenant_id", &self.tenant_id)
-            .field("client_id", &self.client_id)
-            .field("client_secret", &"REDACTED")
-            .finish()
-    }
-}
-
-#[derive(Serialize, Deserialize, Default, Clone)]
+#[derive(Serialize, Deserialize, Default, Debug, Clone)]
 #[serde(default)]
 pub struct TierCreds {
-    #[serde(rename = "access", alias = "accessKey")]
+    #[serde(rename = "accessKey")]
    pub access_key: String,
-    #[serde(rename = "secret", alias = "secretKey")]
+    #[serde(rename = "secretKey")]
    pub secret_key: String,

-    #[serde(rename = "awsrole", alias = "awsRole")]
+    #[serde(rename = "awsRole")]
    pub aws_role: bool,
-    #[serde(rename = "awsroleWebIdentity", alias = "awsRoleWebIdentityTokenFile")]
+    #[serde(rename = "awsRoleWebIdentityTokenFile")]
    pub aws_role_web_identity_token_file: String,
-    #[serde(rename = "awsroleARN", alias = "awsRoleArn", alias = "awsRoleARN")]
+    #[serde(rename = "awsRoleArn")]
    pub aws_role_arn: String,

-    #[serde(rename = "azSP", alias = "azsp", skip_serializing_if = "TierServicePrincipalAuth::is_empty")]
-    pub azure_service_principal: TierServicePrincipalAuth,
+    //azsp: ServicePrincipalAuth,

-    #[serde(
-        rename = "creds",
-        alias = "credsJson",
-        alias = "credsJSON",
-        alias = "creds_json",
-        default,
-        skip_serializing_if = "Vec::is_empty",
-        with = "base64_bytes"
-    )]
+    //#[serde(rename = "credsJson")]
    pub creds_json: Vec<u8>,
 }
-
-impl std::fmt::Debug for TierCreds {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.debug_struct("TierCreds")
-            .field("access_key", &self.access_key)
-            .field("secret_key", &"REDACTED")
-            .field("aws_role", &self.aws_role)
-            .field(
-                "aws_role_web_identity_token_file",
-                &(!self.aws_role_web_identity_token_file.is_empty()).then_some("REDACTED"),
-            )
-            .field("aws_role_arn", &self.aws_role_arn)
-            .field("azure_service_principal", &self.azure_service_principal)
-            .field("creds_json", &(!self.creds_json.is_empty()).then_some("REDACTED"))
-            .finish()
-    }
-}
-
-mod base64_bytes {
-    use super::*;
-
-    pub(super) fn serialize<S>(value: &[u8], serializer: S) -> Result<S::Ok, S::Error>
-    where
-        S: Serializer,
-    {
-        serializer.serialize_str(&base64_simd::STANDARD.encode_to_string(value))
-    }
-
-    pub(super) fn deserialize<'de, D>(deserializer: D) -> Result<Vec<u8>, D::Error>
-    where
-        D: Deserializer<'de>,
-    {
-        #[derive(Deserialize)]
-        #[serde(untagged)]
-        enum EncodedBytes {
-            Base64(String),
-            Legacy(Vec<u8>),
-        }
-
-        match EncodedBytes::deserialize(deserializer)? {
-            EncodedBytes::Base64(value) => base64_simd::STANDARD
-                .decode_to_vec(value.as_bytes())
-                .or_else(|_| base64_simd::STANDARD_NO_PAD.decode_to_vec(value.as_bytes()))
-                .or_else(|_| base64_simd::URL_SAFE.decode_to_vec(value.as_bytes()))
-                .or_else(|_| base64_simd::URL_SAFE_NO_PAD.decode_to_vec(value.as_bytes()))
-                .map_err(de::Error::custom),
-            EncodedBytes::Legacy(value) => Ok(value),
-        }
-    }
-}
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-
-    #[test]
-    fn tier_creds_accepts_madmin_wire_names_and_base64_gcs_json() {
-        let service_account = r#"{"type":"service_account","project_id":"tier-🚀x"}"#.as_bytes();
-        let encoded = "eyJ0eXBlIjoic2VydmljZV9hY2NvdW50IiwicHJvamVjdF9pZCI6InRpZXIt8J+agHgifQ==";
-        let creds: TierCreds = serde_json::from_value(serde_json::json!({
-            "access": "access",
-            "secret": "secret",
-            "awsrole": false,
-            "creds": encoded,
-        }))
-        .expect("madmin tier credentials should decode");
-
-        assert_eq!(creds.access_key, "access");
-        assert_eq!(creds.secret_key, "secret");
-        assert_eq!(creds.creds_json.as_slice(), &service_account[..]);
-
-        let wire = serde_json::to_value(&creds).expect("madmin tier credentials should encode");
-        assert_eq!(wire["access"], "access");
-        assert_eq!(wire["secret"], "secret");
-        assert_eq!(wire["creds"], encoded);
-        assert!(wire.get("accessKey").is_none());
-        assert!(wire.get("secretKey").is_none());
-
-        let legacy: TierCreds = serde_json::from_value(serde_json::json!({
-            "accessKey": "legacy-access",
-            "secretKey": "legacy-secret",
-            "credsJson": service_account,
-        }))
-        .expect("the former RustFS field names and byte-array encoding should remain readable");
-        assert_eq!(legacy.access_key, "legacy-access");
-        assert_eq!(legacy.secret_key, "legacy-secret");
-        assert_eq!(legacy.creds_json.as_slice(), &service_account[..]);
-    }
-
-    #[test]
-    fn tier_creds_accepts_all_supported_base64_alphabets_and_padding_modes() {
-        let service_account = r#"{"type":"service_account","project_id":"tier-🚀"}"#.as_bytes();
-        for encoder in [
-            base64_simd::STANDARD,
-            base64_simd::STANDARD_NO_PAD,
-            base64_simd::URL_SAFE,
-            base64_simd::URL_SAFE_NO_PAD,
-        ] {
-            let encoded = encoder.encode_to_string(service_account);
-            let creds: TierCreds = serde_json::from_value(serde_json::json!({ "creds": encoded }))
-                .expect("all supported madmin base64 forms should decode");
-            assert_eq!(creds.creds_json, service_account);
-        }
-    }
-
-    #[test]
-    fn tier_creds_debug_redacts_secret_payloads() {
-        let creds = TierCreds {
-            access_key: "access".to_string(),
-            secret_key: "tier-secret-value".to_string(),
-            aws_role_web_identity_token_file: "/var/run/private-token".to_string(),
-            creds_json: br#"{"private_key":"gcs-private-key-value"}"#.to_vec(),
-            ..Default::default()
-        };
-
-        let rendered = format!("{creds:?}");
-        assert!(!rendered.contains("tier-secret-value"));
-        assert!(!rendered.contains("/var/run/private-token"));
-        assert!(!rendered.contains("gcs-private-key-value"));
-    }
-
-    #[test]
-    fn tier_creds_accepts_canonical_madmin_azure_service_principal_wire_shape() {
-        let creds: TierCreds = serde_json::from_value(serde_json::json!({
-            "azSP": {
-                "TenantID": "tenant",
-                "ClientID": "client",
-                "ClientSecret": "service-principal-secret"
-            }
-        }))
-        .expect("canonical madmin azure service principal credentials should decode");
-
-        assert_eq!(creds.azure_service_principal.tenant_id, "tenant");
-        assert_eq!(creds.azure_service_principal.client_id, "client");
-        assert_eq!(creds.azure_service_principal.client_secret, "service-principal-secret");
-        let wire = serde_json::to_value(&creds).expect("canonical madmin credentials should encode");
-        assert_eq!(wire["azSP"]["TenantID"], "tenant");
-        assert_eq!(wire["azSP"]["ClientID"], "client");
-        assert_eq!(wire["azSP"]["ClientSecret"], "service-principal-secret");
-        assert!(!format!("{creds:?}").contains("service-principal-secret"));
-    }
-}
@@ -42,7 +42,7 @@ const WASABI_ALTERNATIVE_ENDPOINTS: &[(&str, &str)] = &[
 pub enum TierType {
    #[default]
    Unsupported,
-    #[serde(rename = "s3", alias = "S3")]
+    #[serde(rename = "s3")]
    S3,
    #[serde(rename = "wasabi")]
    Wasabi,
@@ -58,7 +58,7 @@ pub enum TierType {
    Huaweicloud,
    #[serde(rename = "azure")]
    Azure,
-    #[serde(rename = "gcs", alias = "GCS")]
+    #[serde(rename = "gcs")]
    GCS,
    #[serde(rename = "r2")]
    R2,
@@ -138,18 +138,16 @@ impl TierType {
    }
 }

-pub(crate) const TIER_CREDENTIAL_REDACTED: &str = "REDACTED";
-
-#[derive(Default, Serialize, Deserialize)]
+#[derive(Default, Debug, Serialize, Deserialize)]
 #[serde(default)]
 pub struct TierConfig {
    #[serde(skip)]
    pub version: String,
-    #[serde(rename = "type", alias = "Type")]
+    #[serde(rename = "type")]
    pub tier_type: TierType,
-    #[serde(rename = "Name", alias = "name", skip_serializing)]
+    #[serde(skip)]
    pub name: String,
-    #[serde(rename = "s3", alias = "S3", skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "s3", skip_serializing_if = "Option::is_none")]
    pub s3: Option<TierS3>,
    #[serde(rename = "wasabi", skip_serializing_if = "Option::is_none")]
    pub wasabi: Option<TierWasabi>,
@@ -161,7 +159,7 @@ pub struct TierConfig {
    pub huaweicloud: Option<TierHuaweicloud>,
    #[serde(rename = "azure", skip_serializing_if = "Option::is_none")]
    pub azure: Option<TierAzure>,
-    #[serde(rename = "gcs", alias = "GCS", skip_serializing_if = "Option::is_none")]
+    #[serde(rename = "gcs", skip_serializing_if = "Option::is_none")]
    pub gcs: Option<TierGCS>,
    #[serde(rename = "r2", skip_serializing_if = "Option::is_none")]
    pub r2: Option<TierR2>,
@@ -172,91 +170,109 @@ pub struct TierConfig {
 }

 impl Clone for TierConfig {
-    fn clone(&self) -> Self {
-        self.redacted()
+    fn clone(&self) -> TierConfig {
+        let mut s3 = None;
+        let mut wasabi = None;
+        let mut r = None;
+        let mut compatible_backend = None;
+        let mut aliyun = None;
+        let mut tencent = None;
+        let mut huaweicloud = None;
+        let mut azure = None;
+        let mut gcs = None;
+        let mut r2 = None;
+        match self.tier_type {
+            TierType::S3 => {
+                if let Some(s3_) = self.s3.as_ref() {
+                    let mut s3_clone = s3_.clone();
+                    s3_clone.secret_key = "REDACTED".to_string();
+                    s3 = Some(s3_clone);
+                }
+            }
+            TierType::Wasabi => {
+                if let Some(wasabi_) = self.wasabi.as_ref() {
+                    let mut wasabi_clone = wasabi_.clone();
+                    wasabi_clone.secret_key = "REDACTED".to_string();
+                    wasabi = Some(wasabi_clone);
+                }
+            }
+            TierType::RustFS => {
+                if let Some(r_) = self.rustfs.as_ref() {
+                    let mut r_clone = r_.clone();
+                    r_clone.secret_key = "REDACTED".to_string();
+                    r = Some(r_clone);
+                }
+            }
+            TierType::MinIO => {
+                if let Some(compatible_backend_) = self.minio.as_ref() {
+                    let mut compatible_backend_clone = compatible_backend_.clone();
+                    compatible_backend_clone.secret_key = "REDACTED".to_string();
+                    compatible_backend = Some(compatible_backend_clone);
+                }
+            }
+            TierType::Aliyun => {
+                if let Some(aliyun_) = self.aliyun.as_ref() {
+                    let mut aliyun_clone = aliyun_.clone();
+                    aliyun_clone.secret_key = "REDACTED".to_string();
+                    aliyun = Some(aliyun_clone);
+                }
+            }
+            TierType::Tencent => {
+                if let Some(tencent_) = self.tencent.as_ref() {
+                    let mut tencent_clone = tencent_.clone();
+                    tencent_clone.secret_key = "REDACTED".to_string();
+                    tencent = Some(tencent_clone);
+                }
+            }
+            TierType::Huaweicloud => {
+                if let Some(huaweicloud_) = self.huaweicloud.as_ref() {
+                    let mut huaweicloud_clone = huaweicloud_.clone();
+                    huaweicloud_clone.secret_key = "REDACTED".to_string();
+                    huaweicloud = Some(huaweicloud_clone);
+                }
+            }
+            TierType::Azure => {
+                if let Some(azure_) = self.azure.as_ref() {
+                    let mut azure_clone = azure_.clone();
+                    azure_clone.secret_key = "REDACTED".to_string();
+                    azure = Some(azure_clone);
+                }
+            }
+            TierType::GCS => {
+                if let Some(gcs_) = self.gcs.as_ref() {
+                    let mut gcs_clone = gcs_.clone();
+                    gcs_clone.creds = "REDACTED".to_string();
+                    gcs = Some(gcs_clone);
+                }
+            }
+            TierType::R2 => {
+                if let Some(r2_) = self.r2.as_ref() {
+                    let mut r2_clone = r2_.clone();
+                    r2_clone.secret_key = "REDACTED".to_string();
+                    r2 = Some(r2_clone);
+                }
+            }
+            _ => (),
+        }
+        TierConfig {
+            version: self.version.clone(),
+            tier_type: self.tier_type.clone(),
+            name: self.name.clone(),
+            s3,
+            wasabi,
+            rustfs: r,
+            minio: compatible_backend,
+            aliyun,
+            tencent,
+            huaweicloud,
+            azure,
+            gcs,
+            r2,
+        }
    }
 }

 impl TierConfig {
-    pub(crate) fn redacted(&self) -> Self {
-        let mut redacted = Self {
-            version: self.version.clone(),
-            tier_type: self.tier_type.clone(),
-            name: self.name.clone(),
-            ..Default::default()
-        };
-        match self.tier_type {
-            TierType::S3 => {
-                redacted.s3 = self.s3.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    if !backend.aws_role_web_identity_token_file.is_empty() {
-                        backend.aws_role_web_identity_token_file = TIER_CREDENTIAL_REDACTED.to_string();
-                    }
-                    backend
-                });
-            }
-            TierType::Wasabi => {
-                redacted.wasabi = self.wasabi.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::RustFS => {
-                redacted.rustfs = self.rustfs.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::MinIO => {
-                redacted.minio = self.minio.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::Aliyun => {
-                redacted.aliyun = self.aliyun.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::Tencent => {
-                redacted.tencent = self.tencent.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::Huaweicloud => {
-                redacted.huaweicloud = self.huaweicloud.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::Azure => {
-                redacted.azure = self.azure.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    if !backend.sp_auth.client_secret.is_empty() {
-                        backend.sp_auth.client_secret = TIER_CREDENTIAL_REDACTED.to_string();
-                    }
-                    backend
-                });
-            }
-            TierType::GCS => {
-                redacted.gcs = self.gcs.clone().map(|mut backend| {
-                    backend.creds = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::R2 => {
-                redacted.r2 = self.r2.clone().map(|mut backend| {
-                    backend.secret_key = TIER_CREDENTIAL_REDACTED.to_string();
-                    backend
-                });
-            }
-            TierType::Unsupported => {}
-        }
-        redacted
-    }
-
    pub(crate) fn clone_with_credentials(&self) -> Self {
        Self {
            version: self.version.clone(),
@@ -356,61 +372,31 @@ impl TierConfig {
    }
 }

-impl std::fmt::Debug for TierConfig {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let redacted = self.redacted();
-        f.debug_struct("TierConfig")
-            .field("version", &redacted.version)
-            .field("tier_type", &redacted.tier_type)
-            .field("name", &redacted.name)
-            .field("s3", &redacted.s3)
-            .field("wasabi", &redacted.wasabi)
-            .field("aliyun", &redacted.aliyun)
-            .field("tencent", &redacted.tencent)
-            .field("huaweicloud", &redacted.huaweicloud)
-            .field("azure", &redacted.azure)
-            .field("gcs", &redacted.gcs)
-            .field("r2", &redacted.r2)
-            .field("rustfs", &redacted.rustfs)
-            .field("minio", &redacted.minio)
-            .finish()
-    }
-}
-
 //type S3Options = impl Fn(TierS3) -> Pin<Box<Result<()>>> + Send + Sync + 'static;

 #[derive(Serialize, Deserialize, Default, Debug, Clone)]
 #[serde(default)]
 pub struct TierS3 {
-    #[serde(alias = "Name")]
    pub name: String,
-    #[serde(alias = "Endpoint")]
    pub endpoint: String,
-    #[serde(rename = "accessKey", alias = "AccessKey")]
+    #[serde(rename = "accessKey")]
    pub access_key: String,
-    #[serde(rename = "secretKey", alias = "SecretKey")]
+    #[serde(rename = "secretKey")]
    pub secret_key: String,
-    #[serde(alias = "Bucket")]
    pub bucket: String,
-    #[serde(alias = "Prefix")]
    pub prefix: String,
-    #[serde(alias = "Region")]
    pub region: String,
-    #[serde(rename = "storageClass", alias = "StorageClass")]
+    #[serde(rename = "storageClass")]
    pub storage_class: String,
-    #[serde(rename = "AWSRole", alias = "awsRole", skip_serializing)]
+    #[serde(skip)]
    pub aws_role: bool,
-    #[serde(
-        rename = "AWSRoleWebIdentityTokenFile",
-        alias = "awsRoleWebIdentityTokenFile",
-        skip_serializing
-    )]
+    #[serde(skip)]
    pub aws_role_web_identity_token_file: String,
-    #[serde(rename = "AWSRoleARN", alias = "awsRoleARN", alias = "awsRoleArn", skip_serializing)]
+    #[serde(skip)]
    pub aws_role_arn: String,
-    #[serde(rename = "AWSRoleSessionName", alias = "awsRoleSessionName", skip_serializing)]
+    #[serde(skip)]
    pub aws_role_session_name: String,
-    #[serde(rename = "AWSRoleDurationSeconds", alias = "awsRoleDurationSeconds", skip_serializing)]
+    #[serde(skip)]
    pub aws_role_duration_seconds: i32,
 }

@@ -637,11 +623,8 @@ pub struct TierHuaweicloud {
 #[derive(Serialize, Deserialize, Default, Debug, Clone)]
 #[serde(default)]
 pub struct ServicePrincipalAuth {
-    #[serde(alias = "TenantID")]
    pub tenant_id: String,
-    #[serde(alias = "ClientID")]
    pub client_id: String,
-    #[serde(alias = "ClientSecret")]
    pub client_secret: String,
 }

@@ -657,9 +640,9 @@ pub struct TierAzure {
    pub bucket: String,
    pub prefix: String,
    pub region: String,
-    #[serde(rename = "storageClass", alias = "StorageClass")]
+    #[serde(rename = "storageClass")]
    pub storage_class: String,
-    #[serde(rename = "spAuth", alias = "SPAuth")]
+    #[serde(rename = "spAuth")]
    pub sp_auth: ServicePrincipalAuth,
 }

@@ -713,19 +696,14 @@ fn AzureStorageClass(sc string) func(az *TierAzure) error {
 #[derive(Serialize, Deserialize, Default, Debug, Clone)]
 #[serde(default)]
 pub struct TierGCS {
-    #[serde(alias = "Name")]
    pub name: String,
-    #[serde(alias = "Endpoint")]
    pub endpoint: String,
-    #[serde(rename = "creds", alias = "Creds")]
+    #[serde(rename = "creds")]
    pub creds: String,
-    #[serde(alias = "Bucket")]
    pub bucket: String,
-    #[serde(alias = "Prefix")]
    pub prefix: String,
-    #[serde(alias = "Region")]
    pub region: String,
-    #[serde(rename = "storageClass", alias = "StorageClass")]
+    #[serde(rename = "storageClass")]
    pub storage_class: String,
 }

@@ -747,43 +725,6 @@ pub struct TierR2 {
 mod tests {
    use super::*;

-    #[test]
-    fn s3_gcs_type_uppercase_aliases_preserve_lowercase_output() {
-        let s3: TierType = serde_json::from_str(r#""S3""#).expect("uppercase S3 wire value should decode");
-        let gcs: TierType = serde_json::from_str(r#""GCS""#).expect("uppercase GCS wire value should decode");
-        assert!(matches!(s3, TierType::S3));
-        assert!(matches!(gcs, TierType::GCS));
-        assert_eq!(serde_json::to_string(&s3).expect("S3 type should encode"), r#""s3""#);
-        assert_eq!(serde_json::to_string(&gcs).expect("GCS type should encode"), r#""gcs""#);
-    }
-
-    #[test]
-    fn azure_service_principal_accepts_canonical_madmin_field_names() {
-        for field in ["TenantID", "ClientID", "ClientSecret"] {
-            let mut sp_auth = serde_json::Map::new();
-            sp_auth.insert(field.to_string(), serde_json::Value::String("present".to_string()));
-            let config: TierConfig = serde_json::from_value(serde_json::json!({
-                "type": "azure",
-                "Name": "COLD-AZURE",
-                "azure": {
-                    "name": "COLD-AZURE",
-                    "endpoint": "https://azure.example.invalid",
-                    "accessKey": "account",
-                    "secretKey": "key",
-                    "bucket": "archive",
-                    "SPAuth": sp_auth
-                }
-            }))
-            .expect("mixed RustFS/madmin Azure payload should decode");
-            let sp_auth = &config.azure.expect("Azure payload should exist").sp_auth;
-
-            assert!(
-                !sp_auth.tenant_id.is_empty() || !sp_auth.client_id.is_empty() || !sp_auth.client_secret.is_empty(),
-                "canonical {field} must not be silently discarded"
-            );
-        }
-    }
-
    fn wasabi_config() -> TierWasabi {
        TierWasabi {
            name: "COLD-WASABI".to_string(),
@@ -897,14 +838,9 @@ mod tests {
        let config = TierConfig {
            tier_type: TierType::Wasabi,
            wasabi: Some(wasabi_config()),
-            rustfs: Some(TierRustFS {
-                access_key: "inactive-access".to_string(),
-                secret_key: "inactive-secret".to_string(),
-                ..Default::default()
-            }),
            ..Default::default()
        };
-        let redacted = config.redacted();
+        let redacted = config.clone();
        assert_eq!(
            redacted
                .wasabi
@@ -913,46 +849,21 @@ mod tests {
                .secret_key,
            "REDACTED"
        );
-        assert!(redacted.rustfs.is_none(), "the external view should retain only the active provider");
-        let cloned = config.clone();
-        assert_eq!(cloned.wasabi.expect("redacted Wasabi clone should remain").secret_key, "REDACTED");
-        assert!(cloned.rustfs.is_none(), "ordinary Clone must retain its redacted API semantics");
-        let preserved = config.clone_with_credentials();
        assert_eq!(
-            preserved
+            config
+                .clone_with_credentials()
                .wasabi
                .as_ref()
-                .expect("credential-bearing Wasabi snapshot should remain")
+                .expect("credential-bearing Wasabi payload should remain")
                .secret_key,
            "secret"
        );
-        assert_eq!(
-            preserved
-                .rustfs
-                .expect("credential-bearing snapshots should preserve inactive provider data")
-                .secret_key,
-            "inactive-secret"
-        );

        let mut debug_config = wasabi_config();
        debug_config.secret_key = "wasabi-debug-secret-value".to_string();
        let debug = format!("{debug_config:?}");
        assert!(debug.contains("REDACTED"));
        assert!(!debug.contains("wasabi-debug-secret-value"));
-
-        let debug = format!(
-            "{:?}",
-            TierConfig {
-                tier_type: TierType::RustFS,
-                rustfs: Some(TierRustFS {
-                    secret_key: "rustfs-debug-secret-value".to_string(),
-                    ..Default::default()
-                }),
-                ..Default::default()
-            }
-        );
-        assert!(debug.contains("REDACTED"));
-        assert!(!debug.contains("rustfs-debug-secret-value"));
    }

    #[test]
@@ -983,7 +894,7 @@ mod tests {
        assert_eq!(encoded, expected);
        let decoded: TierConfig = serde_json::from_value(encoded).expect("Wasabi Admin JSON should decode");
        assert!(matches!(decoded.tier_type, TierType::Wasabi));
-        let redacted = config.redacted();
+        let redacted = config.clone();
        assert_eq!(
            config
                .wasabi
@@ -1005,87 +916,4 @@ mod tests {
            "REDACTED"
        );
    }
-
-    #[test]
-    fn api_serialization_and_debug_redact_s3_gcs_and_azure_credentials() {
-        let cases = [
-            (
-                "s3",
-                TierConfig {
-                    tier_type: TierType::S3,
-                    s3: Some(TierS3 {
-                        secret_key: "s3-secret-bytes".to_string(),
-                        aws_role_web_identity_token_file: "/var/run/s3-private-token".to_string(),
-                        ..Default::default()
-                    }),
-                    ..Default::default()
-                },
-                vec!["s3-secret-bytes", "/var/run/s3-private-token"],
-            ),
-            (
-                "gcs",
-                TierConfig {
-                    tier_type: TierType::GCS,
-                    gcs: Some(TierGCS {
-                        creds: r#"{"type":"service_account","private_key":"gcs-private-key-bytes"}"#.to_string(),
-                        ..Default::default()
-                    }),
-                    ..Default::default()
-                },
-                vec!["gcs-private-key-bytes"],
-            ),
-            (
-                "azure",
-                TierConfig {
-                    tier_type: TierType::Azure,
-                    azure: Some(TierAzure {
-                        secret_key: "azure-account-secret-bytes".to_string(),
-                        sp_auth: ServicePrincipalAuth {
-                            client_secret: "azure-client-secret-bytes".to_string(),
-                            ..Default::default()
-                        },
-                        ..Default::default()
-                    }),
-                    ..Default::default()
-                },
-                vec!["azure-account-secret-bytes", "azure-client-secret-bytes"],
-            ),
-        ];
-
-        for (provider, config, secrets) in cases {
-            let api = serde_json::to_string(&config.redacted()).expect("redacted API config should serialize");
-            let debug = format!("{config:?}");
-            assert!(api.contains(TIER_CREDENTIAL_REDACTED), "{provider} API output should be visibly redacted");
-            assert!(
-                debug.contains(TIER_CREDENTIAL_REDACTED),
-                "{provider} Debug output should be visibly redacted"
-            );
-            for secret in secrets {
-                assert!(!api.contains(secret), "{provider} API output exposed credential bytes");
-                assert!(!debug.contains(secret), "{provider} Debug output exposed credential bytes");
-            }
-        }
-    }
-
-    #[test]
-    fn azure_static_account_redaction_preserves_an_empty_service_principal_secret() {
-        let config = TierConfig {
-            tier_type: TierType::Azure,
-            azure: Some(TierAzure {
-                secret_key: "azure-account-secret-bytes".to_string(),
-                sp_auth: ServicePrincipalAuth::default(),
-                ..Default::default()
-            }),
-            ..Default::default()
-        };
-
-        let api = serde_json::to_value(config.redacted()).expect("redacted Azure API config should serialize");
-        let debug = format!("{config:?}");
-
-        assert_eq!(api["azure"]["secretKey"], TIER_CREDENTIAL_REDACTED);
-        assert_eq!(api["azure"]["spAuth"]["client_secret"], "");
-        assert!(debug.contains("client_secret: \"\""));
-        assert!(!debug.contains("client_secret: \"REDACTED\""));
-        assert!(!debug.contains("azure-account-secret-bytes"));
-    }
 }
@@ -20,7 +20,7 @@

 use crate::error::is_err_bucket_not_found;
 use crate::services::tier::{
-    tier::{ERR_TIER_BACKEND_IN_USE, ERR_TIER_INVALID_CONFIG, ERR_TIER_TYPE_UNSUPPORTED},
+    tier::{ERR_TIER_INVALID_CONFIG, ERR_TIER_TYPE_UNSUPPORTED},
    tier_config::{TierConfig, TierType},
    tier_handlers::{ERR_TIER_BUCKET_NOT_FOUND, ERR_TIER_NOT_FOUND, ERR_TIER_PERM_ERR},
    warm_backend_aliyun::WarmBackendAliyun,
@@ -55,21 +55,18 @@ use s3s::header::{
 };
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::Duration;
 use time::OffsetDateTime;
 use time::format_description::well_known::{Rfc2822, Rfc3339};
-use tokio::io::AsyncReadExt;
 use tracing::{info, warn};

 pub type WarmBackendImpl = Box<dyn WarmBackend + Send + Sync + 'static>;

+const PROBE_OBJECT: &str = "probeobject";
+
 /// Largest object the S3-compatible warm backends accept for a multipart put.
 pub(crate) const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
 /// Part-count ceiling S3-compatible services impose on a multipart upload.
 pub(crate) const MAX_PARTS_COUNT: i64 = 10000;
-pub(crate) const WARM_BACKEND_PROBE_TIMEOUT: Duration = Duration::from_secs(30);
-const WARM_BACKEND_PROBE_RECONCILE_INTERVAL: Duration = Duration::from_secs(1);
-const WARM_BACKEND_PROBE_FINAL_RECONCILE_TIMEOUT: Duration = Duration::from_secs(1);

 #[derive(Default)]
 pub struct WarmBackendGetOpts {
@@ -263,23 +260,6 @@ pub(crate) struct S3CompatibleWarmBackendParams<'a> {
    pub validate_endpoint: fn(&url::Url) -> Result<(), rustfs_utils::egress::OutboundUrlError>,
 }

-/// Return the authority format accepted by `TransitionClient::new` while
-/// retaining an explicitly configured port. `url::Url::host_str()` omits the
-/// brackets needed when an IPv6 literal is combined with a port.
-pub(crate) fn endpoint_authority(url: &url::Url) -> Result<String, std::io::Error> {
-    let host = url
-        .host_str()
-        .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?;
-    let port = url.port().unwrap_or(if url.scheme() == "https" { 443 } else { 80 });
-    if host.starts_with('[') && host.ends_with(']') {
-        Ok(format!("{host}:{port}"))
-    } else if host.contains(':') {
-        Ok(format!("[{host}]:{port}"))
-    } else {
-        Ok(format!("{host}:{port}"))
-    }
-}
-
 /// Build the [`WarmBackendS3`] shared by the S3-compatible warm backend providers.
 ///
 /// Credential, bucket, and endpoint validation run in this order because the
@@ -318,11 +298,17 @@ pub(crate) async fn new_s3_compatible_warm_backend(
        bucket_lookup: params.bucket_lookup,
        ..Default::default()
    };
-    let endpoint = endpoint_authority(&u)?;
-    // Run the SSRF guard after the host-presence check so a host-less endpoint
-    // keeps this constructor's stable error text.
+    let scheme = u.scheme();
+    let default_port = if scheme == "https" { 443 } else { 80 };
+    let host = u
+        .host_str()
+        .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?;
+    // Runs after the host-presence check above (not immediately after Url::parse) so a
+    // host-less endpoint still reports this constructor's own "missing host" text instead of
+    // validate_endpoint's differently-worded rejection for the same input.
    (params.validate_endpoint)(&u).map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?;
-    let client = TransitionClient::new(&endpoint, opts, params.provider_tag).await?;
+    let client =
+        TransitionClient::new(&format!("{}:{}", host, u.port().unwrap_or(default_port)), opts, params.provider_tag).await?;

    let client = Arc::new(client);
    let core = TransitionCore(Arc::clone(&client));
@@ -465,187 +451,25 @@ impl TransitionCandidateReconciler for MeteredTransitionCandidateReconciler {
    }
 }

-async fn remove_discovered_probe_candidate(
-    w: &WarmBackendImpl,
-    probe_object: &str,
-    candidate: TransitionCandidateProbe,
-) -> Result<bool, std::io::Error> {
-    match candidate {
-        TransitionCandidateProbe::Missing => Ok(false),
-        TransitionCandidateProbe::VersionedPresent(remote_version_id) => {
-            w.remove_exact(probe_object, &remote_version_id).await?;
-            Ok(true)
-        }
-        TransitionCandidateProbe::UnversionedPresent => {
-            w.remove(probe_object, "").await?;
-            Ok(true)
-        }
-        TransitionCandidateProbe::Ambiguous => {
-            Err(std::io::Error::other("remote tier probe PUT produced multiple possible versions"))
-        }
-        TransitionCandidateProbe::Unsupported => Err(std::io::Error::new(
-            std::io::ErrorKind::Unsupported,
-            "remote tier cannot discover the outcome of a probe PUT",
-        )),
-    }
-}
-
-async fn compensate_uncertain_probe_put(
-    w: &WarmBackendImpl,
-    probe_object: &str,
-    settle_deadline: tokio::time::Instant,
-) -> Result<(), std::io::Error> {
-    let final_deadline = settle_deadline + WARM_BACKEND_PROBE_FINAL_RECONCILE_TIMEOUT;
-    let mut removed_any = false;
-    while tokio::time::Instant::now() < settle_deadline {
-        let candidate = match tokio::time::timeout_at(settle_deadline, w.probe_transition_candidate(probe_object)).await {
-            Ok(candidate) => candidate?,
-            Err(_) => break,
-        };
-        if matches!(candidate, TransitionCandidateProbe::Missing) && removed_any {
-            break;
-        }
-        removed_any |= tokio::time::timeout_at(settle_deadline, remove_discovered_probe_candidate(w, probe_object, candidate))
-            .await
-            .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out reconciling a remote tier probe PUT"))??;
-
-        let now = tokio::time::Instant::now();
-        if now >= settle_deadline {
-            break;
-        }
-        tokio::time::sleep_until(std::cmp::min(settle_deadline, now + WARM_BACKEND_PROBE_RECONCILE_INTERVAL)).await;
-    }
-
-    let candidate = tokio::time::timeout_at(final_deadline, w.probe_transition_candidate(probe_object))
-        .await
-        .map_err(|_| {
-            std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out confirming the final remote tier probe state")
-        })??;
-    if !tokio::time::timeout_at(final_deadline, remove_discovered_probe_candidate(w, probe_object, candidate))
-        .await
-        .map_err(|_| {
-            std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out removing the final remote tier probe candidate")
-        })??
-    {
-        return Ok(());
-    }
-
-    let final_candidate = tokio::time::timeout_at(final_deadline, w.probe_transition_candidate(probe_object))
-        .await
-        .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "timed out confirming remote tier probe cleanup"))??;
-    match final_candidate {
-        TransitionCandidateProbe::Missing => Ok(()),
-        _ => Err(std::io::Error::other("remote tier probe cleanup could not be confirmed")),
-    }
-}
-
-fn probe_cleanup_incomplete_error() -> AdminError {
-    let mut err = ERR_TIER_PERM_ERR.clone();
-    err.message = "Remote tier probe outcome is uncertain; cleanup is incomplete".to_string();
-    err
-}
-
-async fn check_warm_backend_with_deadlines(
-    w: Option<&WarmBackendImpl>,
-    deadline: tokio::time::Instant,
-    cleanup_deadline: tokio::time::Instant,
-) -> Result<(), AdminError> {
+pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), AdminError> {
    let w = w.ok_or_else(|| ERR_TIER_NOT_FOUND.clone())?;
-    let probe_object = format!("rustfs-tier-probe-{}", uuid::Uuid::new_v4());
-    let timeout_error = || {
-        let mut err = ERR_TIER_BACKEND_IN_USE.clone();
-        err.message = "Timed out validating the remote tier mutation".to_string();
-        err
-    };
-    tokio::time::timeout_at(deadline, w.validate())
+    w.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?;
+    let remote_version_id = w
+        .put(PROBE_OBJECT, ReaderImpl::Body(Bytes::from("RustFS".as_bytes().to_vec())), 5)
        .await
-        .map_err(|_| timeout_error())?
-        .map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?;
-    let put_result =
-        tokio::time::timeout_at(deadline, w.put(&probe_object, ReaderImpl::Body(Bytes::from_static(b"RustFS")), 6)).await;
-    let remote_version_id = match put_result {
-        Ok(Ok(remote_version_id)) => remote_version_id,
-        Ok(Err(_)) => {
-            return Err(match compensate_uncertain_probe_put(w, &probe_object, cleanup_deadline).await {
-                Ok(()) => ERR_TIER_PERM_ERR.clone(),
-                Err(_) => probe_cleanup_incomplete_error(),
-            });
-        }
-        Err(_) => {
-            let err = timeout_error();
-            return Err(match compensate_uncertain_probe_put(w, &probe_object, cleanup_deadline).await {
-                Ok(()) => err,
-                Err(_) => probe_cleanup_incomplete_error(),
-            });
-        }
-    };
+        .map_err(|_| ERR_TIER_PERM_ERR.clone())?;

-    // S3-family backends do not replay a failed request before returning `Ok`,
-    // while GCS discovers every matching generation. The authoritative probe
-    // below therefore closes the acknowledged-PUT path; only an error or
-    // timeout needs the longer visibility reconciliation above.
-    let authoritative_candidate = match tokio::time::timeout_at(deadline, w.probe_transition_candidate(&probe_object)).await {
-        Ok(Ok(candidate)) => candidate,
-        Ok(Err(_)) | Err(_) => {
-            return Err(match compensate_uncertain_probe_put(w, &probe_object, cleanup_deadline).await {
-                Ok(()) => ERR_TIER_INVALID_CONFIG.clone(),
-                Err(_) => probe_cleanup_incomplete_error(),
-            });
-        }
-    };
-    let response_version_is_valid = w.validate_remote_version_id(&remote_version_id).is_ok();
-    let response_matches_candidate = match &authoritative_candidate {
-        TransitionCandidateProbe::UnversionedPresent => remote_version_id.is_empty(),
-        TransitionCandidateProbe::VersionedPresent(candidate_version) => candidate_version == &remote_version_id,
-        TransitionCandidateProbe::Missing | TransitionCandidateProbe::Ambiguous | TransitionCandidateProbe::Unsupported => false,
-    };
-    if !response_version_is_valid || !response_matches_candidate {
-        return Err(match compensate_uncertain_probe_put(w, &probe_object, cleanup_deadline).await {
-            Ok(()) => ERR_TIER_INVALID_CONFIG.clone(),
-            Err(_) => probe_cleanup_incomplete_error(),
-        });
-    }
-
-    let read_result = tokio::time::timeout_at(deadline, async {
-        let mut reader = w
-            .get(
-                &probe_object,
-                &remote_version_id,
-                WarmBackendGetOpts {
-                    start_offset: 0,
-                    length: 7,
-                },
-            )
+    if w.validate_remote_version_id(&remote_version_id).is_err() {
+        w.remove_exact(PROBE_OBJECT, &remote_version_id)
            .await
            .map_err(|_| ERR_TIER_PERM_ERR.clone())?;
-        let mut body = Vec::new();
-        reader
-            .take(7)
-            .read_to_end(&mut body)
-            .await
-            .map_err(|_| ERR_TIER_PERM_ERR.clone())?;
-        if body != b"RustFS" {
-            return Err(ERR_TIER_PERM_ERR.clone());
-        }
-        Ok(())
-    })
-    .await
-    .map_err(|_| timeout_error())
-    .and_then(|result| result);
-    let cleanup_result = tokio::time::timeout_at(cleanup_deadline, async {
-        if !remove_discovered_probe_candidate(w, &probe_object, authoritative_candidate).await? {
-            return Err(std::io::Error::other("remote tier probe disappeared before cleanup"));
-        }
-        match w.probe_transition_candidate(&probe_object).await? {
-            TransitionCandidateProbe::Missing => Ok(()),
-            _ => Err(std::io::Error::other("remote tier probe remained after cleanup")),
-        }
-    })
-    .await;
-    if !matches!(cleanup_result, Ok(Ok(()))) {
-        return Err(probe_cleanup_incomplete_error());
+        return Err(ERR_TIER_INVALID_CONFIG.clone());
    }
-    if let Err(err) = read_result {
+
+    let read_result = w.get(PROBE_OBJECT, &remote_version_id, WarmBackendGetOpts::default()).await;
+    let remove_result = w.remove(PROBE_OBJECT, &remote_version_id).await;
+    //xhttp.DrainBody(r);
+    if read_result.is_err() || remove_result.is_err() {
        //if is_err_bucket_not_found(&err) {
        //    return Err(ERR_TIER_BUCKET_NOT_FOUND);
        //}
@@ -653,28 +477,12 @@ async fn check_warm_backend_with_deadlines(
            return Err(ERR_TIER_MISSING_CREDENTIALS);
        }*/
        //else {
-        return Err(err);
+        return Err(ERR_TIER_PERM_ERR.clone());
        //}
    }
    Ok(())
 }

-/// Validate a backend using a caller-owned deadline while retaining a bounded
-/// reconciliation window for an uncertain probe PUT. The validation future is
-/// kept alive through cleanup so an outer timeout cannot abandon the remote
-/// probe object.
-pub(crate) async fn check_warm_backend_until(
-    w: Option<&WarmBackendImpl>,
-    deadline: tokio::time::Instant,
-) -> Result<(), AdminError> {
-    check_warm_backend_with_deadlines(w, deadline, deadline + WARM_BACKEND_PROBE_FINAL_RECONCILE_TIMEOUT).await
-}
-
-pub async fn check_warm_backend(w: Option<&WarmBackendImpl>) -> Result<(), AdminError> {
-    let deadline = tokio::time::Instant::now() + WARM_BACKEND_PROBE_TIMEOUT;
-    check_warm_backend_with_deadlines(w, deadline, deadline + WARM_BACKEND_PROBE_TIMEOUT).await
-}
-
 pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBackendImpl, AdminError> {
    let mut d: Option<WarmBackendImpl> = None;
    match tier.tier_type {
@@ -893,7 +701,7 @@ pub async fn new_warm_backend(tier: &TierConfig, probe: bool) -> Result<WarmBack
    let d: WarmBackendImpl = Box::new(MeteredWarmBackend { inner: d });

    if probe {
-        check_warm_backend(Some(&d)).await?;
+        d.validate().await.map_err(|_| ERR_TIER_INVALID_CONFIG.clone())?;
    }
    Ok(d)
 }
@@ -946,7 +754,6 @@ pub(crate) async fn new_transition_candidate_reconciler(
 #[cfg(test)]
 mod tests {
    use super::*;
-    use crate::services::tier::test_util::{MockWarmBackend, MockWarmOp};
    use crate::services::tier::tier_config::TierWasabi;
    use std::sync::{
        Arc,
@@ -1113,38 +920,13 @@ mod tests {

    struct RejectingProbeVersionBackend {
        gets: Arc<AtomicUsize>,
-        present: Arc<std::sync::atomic::AtomicBool>,
        removed_versions: Arc<tokio::sync::Mutex<Vec<String>>>,
-        returned_version: String,
    }

    struct RecordingProbeBackend {
        get_versions: Arc<tokio::sync::Mutex<Vec<String>>>,
-        present: Arc<std::sync::atomic::AtomicBool>,
        removed_versions: Arc<tokio::sync::Mutex<Vec<String>>>,
-        remove_clears_candidate: bool,
        fail_get: bool,
-        body: ProbeBody,
-    }
-
-    struct HangingProbePutBackend {
-        put_started: Arc<tokio::sync::Notify>,
-        present: Arc<std::sync::atomic::AtomicBool>,
-        probes: Arc<AtomicUsize>,
-        removed_versions: Arc<tokio::sync::Mutex<Vec<String>>>,
-    }
-
-    struct LateVisibleProbeBackend {
-        visible_at: tokio::time::Instant,
-        removed: Arc<std::sync::atomic::AtomicBool>,
-        probes: Arc<AtomicUsize>,
-        removed_versions: Arc<tokio::sync::Mutex<Vec<String>>>,
-    }
-
-    #[derive(Clone, Copy)]
-    enum ProbeBody {
-        Exact,
-        Mismatch,
    }

    #[async_trait::async_trait]
@@ -1194,7 +976,7 @@ mod tests {
        }

        async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
-            Ok(self.returned_version.clone())
+            Ok(uuid::Uuid::nil().to_string())
        }

        async fn put_with_meta(
@@ -1217,19 +999,10 @@ mod tests {
        }

        async fn remove_exact(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> {
-            self.present.store(false, Ordering::SeqCst);
            self.removed_versions.lock().await.push(rv.to_string());
            Ok(())
        }

-        async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-            if self.present.load(Ordering::SeqCst) {
-                Ok(TransitionCandidateProbe::VersionedPresent(PROBE_VERSION.to_string()))
-            } else {
-                Ok(TransitionCandidateProbe::Missing)
-            }
-        }
-
        async fn in_use(&self) -> Result<bool, std::io::Error> {
            Ok(false)
        }
@@ -1256,121 +1029,15 @@ mod tests {
            if self.fail_get {
                Err(std::io::Error::other("probe GET failed"))
            } else {
-                match self.body {
-                    ProbeBody::Exact => Ok(ReadCloser::new(std::io::Cursor::new(b"RustFS".to_vec()))),
-                    ProbeBody::Mismatch => Ok(ReadCloser::new(std::io::Cursor::new(b"RustFT".to_vec()))),
-                }
+                Ok(ReadCloser::new(std::io::Cursor::new(Vec::new())))
            }
        }

        async fn remove(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> {
-            if self.remove_clears_candidate {
-                self.present.store(false, Ordering::SeqCst);
-            }
            self.removed_versions.lock().await.push(rv.to_string());
            Ok(())
        }

-        async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-            if self.present.load(Ordering::SeqCst) {
-                Ok(TransitionCandidateProbe::VersionedPresent(PROBE_VERSION.to_string()))
-            } else {
-                Ok(TransitionCandidateProbe::Missing)
-            }
-        }
-
-        async fn in_use(&self) -> Result<bool, std::io::Error> {
-            Ok(false)
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl WarmBackend for HangingProbePutBackend {
-        async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
-            self.put_started.notify_one();
-            std::future::pending().await
-        }
-
-        async fn put_with_meta(
-            &self,
-            object: &str,
-            r: ReaderImpl,
-            length: i64,
-            _meta: HashMap<String, String>,
-        ) -> Result<String, std::io::Error> {
-            self.put(object, r, length).await
-        }
-
-        async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-            Err(std::io::Error::other("GET must not run after a timed out probe PUT"))
-        }
-
-        async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> {
-            Err(std::io::Error::other("generic remove must not replace exact probe cleanup"))
-        }
-
-        async fn remove_exact(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> {
-            self.present.store(false, Ordering::SeqCst);
-            self.removed_versions.lock().await.push(rv.to_string());
-            Ok(())
-        }
-
-        async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-            self.probes.fetch_add(1, Ordering::SeqCst);
-            if self.present.load(Ordering::SeqCst) {
-                Ok(TransitionCandidateProbe::VersionedPresent(PROBE_VERSION.to_string()))
-            } else {
-                Ok(TransitionCandidateProbe::Missing)
-            }
-        }
-
-        async fn in_use(&self) -> Result<bool, std::io::Error> {
-            Ok(false)
-        }
-    }
-
-    #[async_trait::async_trait]
-    impl WarmBackend for LateVisibleProbeBackend {
-        async fn put(&self, _object: &str, _r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
-            Err(std::io::Error::new(
-                std::io::ErrorKind::ConnectionReset,
-                "probe PUT response was lost before the object became visible",
-            ))
-        }
-
-        async fn put_with_meta(
-            &self,
-            object: &str,
-            r: ReaderImpl,
-            length: i64,
-            _meta: HashMap<String, String>,
-        ) -> Result<String, std::io::Error> {
-            self.put(object, r, length).await
-        }
-
-        async fn get(&self, _object: &str, _rv: &str, _opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-            Err(std::io::Error::other("GET must not run after a lost probe PUT response"))
-        }
-
-        async fn remove(&self, _object: &str, _rv: &str) -> Result<(), std::io::Error> {
-            Err(std::io::Error::other("generic remove must not replace exact probe cleanup"))
-        }
-
-        async fn remove_exact(&self, _object: &str, rv: &str) -> Result<(), std::io::Error> {
-            self.removed.store(true, Ordering::SeqCst);
-            self.removed_versions.lock().await.push(rv.to_string());
-            Ok(())
-        }
-
-        async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-            self.probes.fetch_add(1, Ordering::SeqCst);
-            if tokio::time::Instant::now() >= self.visible_at && !self.removed.load(Ordering::SeqCst) {
-                Ok(TransitionCandidateProbe::VersionedPresent(PROBE_VERSION.to_string()))
-            } else {
-                Ok(TransitionCandidateProbe::Missing)
-            }
-        }
-
        async fn in_use(&self) -> Result<bool, std::io::Error> {
            Ok(false)
        }
@@ -1431,15 +1098,13 @@ mod tests {
        assert_eq!(probe, TransitionCandidateProbe::Unsupported);
    }

-    #[tokio::test(start_paused = true)]
+    #[tokio::test]
    async fn check_warm_backend_removes_exact_probe_when_versioning_drifts() {
        let gets = Arc::new(AtomicUsize::new(0));
        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let backend: WarmBackendImpl = Box::new(RejectingProbeVersionBackend {
            gets: gets.clone(),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            removed_versions: removed_versions.clone(),
-            returned_version: uuid::Uuid::nil().to_string(),
        });

        let err = check_warm_backend(Some(&backend))
@@ -1448,27 +1113,7 @@ mod tests {

        assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code);
        assert_eq!(gets.load(Ordering::SeqCst), 0);
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn check_warm_backend_rejects_empty_put_version_for_a_versioned_candidate() {
-        let gets = Arc::new(AtomicUsize::new(0));
-        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let backend: WarmBackendImpl = Box::new(RejectingProbeVersionBackend {
-            gets: gets.clone(),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
-            removed_versions: removed_versions.clone(),
-            returned_version: String::new(),
-        });
-
-        let err = check_warm_backend(Some(&backend))
-            .await
-            .expect_err("an empty PUT version must not read or generically delete a versioned object");
-
-        assert_eq!(err.code, ERR_TIER_INVALID_CONFIG.code);
-        assert_eq!(gets.load(Ordering::SeqCst), 0);
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
+        assert_eq!(removed_versions.lock().await.as_slice(), [uuid::Uuid::nil().to_string()]);
    }

    #[tokio::test]
@@ -1477,11 +1122,8 @@ mod tests {
        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let backend: WarmBackendImpl = Box::new(RecordingProbeBackend {
            get_versions: get_versions.clone(),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            removed_versions: removed_versions.clone(),
-            remove_clears_candidate: true,
            fail_get: false,
-            body: ProbeBody::Exact,
        });

        check_warm_backend(Some(&backend))
@@ -1498,11 +1140,8 @@ mod tests {
        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
        let backend: WarmBackendImpl = Box::new(RecordingProbeBackend {
            get_versions: get_versions.clone(),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
            removed_versions: removed_versions.clone(),
-            remove_clears_candidate: true,
            fail_get: true,
-            body: ProbeBody::Exact,
        });

        let err = check_warm_backend(Some(&backend))
@@ -1514,169 +1153,6 @@ mod tests {
        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
    }

-    #[tokio::test]
-    async fn check_warm_backend_removes_probe_after_body_mismatch() {
-        let get_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let backend: WarmBackendImpl = Box::new(RecordingProbeBackend {
-            get_versions,
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
-            removed_versions: removed_versions.clone(),
-            remove_clears_candidate: true,
-            fail_get: false,
-            body: ProbeBody::Mismatch,
-        });
-
-        let err = check_warm_backend(Some(&backend))
-            .await
-            .expect_err("a mismatched body should fail after cleanup");
-
-        assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
-    }
-
-    #[tokio::test]
-    async fn check_warm_backend_rejects_a_stale_candidate_after_successful_delete() {
-        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let backend: WarmBackendImpl = Box::new(RecordingProbeBackend {
-            get_versions: Arc::new(tokio::sync::Mutex::new(Vec::new())),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
-            removed_versions: removed_versions.clone(),
-            remove_clears_candidate: false,
-            fail_get: false,
-            body: ProbeBody::Exact,
-        });
-
-        let err = check_warm_backend(Some(&backend))
-            .await
-            .expect_err("cleanup must not succeed while the deleted candidate remains visible");
-
-        assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-        assert!(err.message.contains("cleanup is incomplete"));
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn check_warm_backend_reconciles_a_lost_put_response() {
-        let backend = MockWarmBackend::new();
-        backend.lose_next_put_response();
-        let driver: WarmBackendImpl = Box::new(backend.clone());
-
-        let err = check_warm_backend(Some(&driver))
-            .await
-            .expect_err("a lost probe PUT response must fail after compensation");
-
-        assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-        assert_eq!(backend.object_count().await, 0);
-        assert_eq!(backend.exact_remove_count(), 1);
-        let operations = backend.op_log().await;
-        let put = operations.iter().find_map(|operation| match operation {
-            MockWarmOp::Put { object } => Some(object),
-            _ => None,
-        });
-        let probe = operations.iter().find_map(|operation| match operation {
-            MockWarmOp::Probe { object } => Some(object),
-            _ => None,
-        });
-        let remove = operations.iter().find_map(|operation| match operation {
-            MockWarmOp::Remove { object } => Some(object),
-            _ => None,
-        });
-        let (Some(put), Some(probe), Some(remove)) = (put, probe, remove) else {
-            panic!("lost-response compensation should PUT, probe, and remove");
-        };
-        assert_eq!(put, probe);
-        assert_eq!(probe, remove);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn check_warm_backend_retries_until_a_late_put_becomes_visible() {
-        let probes = Arc::new(AtomicUsize::new(0));
-        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let driver: WarmBackendImpl = Box::new(LateVisibleProbeBackend {
-            visible_at: tokio::time::Instant::now() + Duration::from_secs(5),
-            removed: Arc::new(std::sync::atomic::AtomicBool::new(false)),
-            probes: probes.clone(),
-            removed_versions: removed_versions.clone(),
-        });
-
-        let err = check_warm_backend(Some(&driver))
-            .await
-            .expect_err("a late-visible probe PUT must still report the lost response");
-
-        assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-        assert!(
-            probes.load(Ordering::SeqCst) > 5,
-            "reconciliation must not stop at the first Missing result"
-        );
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
-    }
-
-    #[tokio::test]
-    async fn check_warm_backend_reports_incomplete_cleanup_without_guessing() {
-        for candidate in [TransitionCandidateProbe::Unsupported, TransitionCandidateProbe::Ambiguous] {
-            let backend = MockWarmBackend::new();
-            backend.set_transition_candidate_probe_override(Some(candidate)).await;
-            backend.lose_next_put_response();
-            let driver: WarmBackendImpl = Box::new(backend.clone());
-
-            let err = check_warm_backend(Some(&driver))
-                .await
-                .expect_err("an uncertain candidate must fail without a guessed delete");
-
-            assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-            assert!(err.message.contains("cleanup is incomplete"));
-            assert_eq!(backend.remove_count().await, 0);
-            assert_eq!(backend.object_count().await, 1);
-        }
-    }
-
-    #[tokio::test]
-    async fn check_warm_backend_reports_an_exact_cleanup_failure() {
-        let backend = MockWarmBackend::new();
-        backend.set_remove_failure(true);
-        backend.lose_next_put_response();
-        let driver: WarmBackendImpl = Box::new(backend.clone());
-
-        let err = check_warm_backend(Some(&driver))
-            .await
-            .expect_err("an exact cleanup failure must replace the ambiguous PUT error");
-
-        assert_eq!(err.code, ERR_TIER_PERM_ERR.code);
-        assert!(err.message.contains("cleanup is incomplete"));
-        assert_eq!(backend.exact_remove_count(), 1);
-        assert_eq!(backend.object_count().await, 1);
-    }
-
-    #[tokio::test(start_paused = true)]
-    async fn check_warm_backend_reconciles_a_timed_out_put() {
-        let put_started = Arc::new(tokio::sync::Notify::new());
-        let probes = Arc::new(AtomicUsize::new(0));
-        let removed_versions = Arc::new(tokio::sync::Mutex::new(Vec::new()));
-        let driver: WarmBackendImpl = Box::new(HangingProbePutBackend {
-            put_started: put_started.clone(),
-            present: Arc::new(std::sync::atomic::AtomicBool::new(true)),
-            probes: probes.clone(),
-            removed_versions: removed_versions.clone(),
-        });
-        let check = check_warm_backend(Some(&driver));
-        tokio::pin!(check);
-        tokio::select! {
-            _ = put_started.notified() => {}
-            result = &mut check => panic!("probe completed before the PUT timeout: {result:?}"),
-        }
-
-        tokio::time::advance(WARM_BACKEND_PROBE_TIMEOUT + Duration::from_millis(1)).await;
-        let err = check.await.expect_err("a timed out probe PUT must fail after compensation");
-
-        assert_eq!(err.code, ERR_TIER_BACKEND_IN_USE.code);
-        assert!(
-            probes.load(Ordering::SeqCst) > 1,
-            "timed-out PUT reconciliation must keep checking through the visibility window"
-        );
-        assert_eq!(removed_versions.lock().await.as_slice(), [PROBE_VERSION]);
-    }
-
    #[tokio::test]
    async fn new_wasabi_backend_honors_probe_flag() {
        let tier = TierConfig {
@@ -1822,15 +1298,6 @@ mod tests {
        assert_eq!(insecure.client.endpoint_url.port_or_known_default(), Some(80));
    }

-    #[test]
-    fn endpoint_authority_preserves_ipv6_brackets_and_explicit_port() {
-        let url = url::Url::parse("https://[2001:db8::1]:9443").expect("the IPv6 endpoint should parse");
-        assert_eq!(
-            endpoint_authority(&url).expect("the endpoint should have an authority"),
-            "[2001:db8::1]:9443"
-        );
-    }
-
    #[tokio::test]
    async fn s3_compatible_backend_strips_only_a_trailing_prefix_separator() {
        let mut params = s3_compatible_params("http://tier.example.com:9000");
@@ -23,7 +23,7 @@ use std::collections::HashMap;
 use crate::services::tier::{
    tier_config::TierAliyun,
    warm_backend::{
-        S3CompatibleWarmBackendParams, TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
+        S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
        new_s3_compatible_warm_backend, optimal_part_size,
    },
    warm_backend_s3::WarmBackendS3,
@@ -89,10 +89,6 @@ impl WarmBackend for WarmBackendAliyun {
        self.0.remove(object, rv).await
    }

-    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        self.0.probe_transition_candidate(object).await
-    }
-
    async fn in_use(&self) -> Result<bool, std::io::Error> {
        self.0.in_use().await
    }
@@ -23,7 +23,7 @@ use std::collections::HashMap;
 use crate::services::tier::{
    tier_config::TierAzure,
    warm_backend::{
-        S3CompatibleWarmBackendParams, TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
+        S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
        new_s3_compatible_warm_backend, optimal_part_size,
    },
    warm_backend_s3::WarmBackendS3,
@@ -89,16 +89,6 @@ impl WarmBackend for WarmBackendAzure {
        self.0.remove(object, rv).await
    }

-    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        // Azure currently uses the shared S3/SigV4 transport, but its normal
-        // object path cannot persist exact remote versions across mixed
-        // RustFS releases. The mutation probe may still detect and precisely
-        // remove a versioned test object before rejecting that configuration.
-        self.0
-            .probe_transition_candidate_with_raw_version_header(object, "x-amz-version-id")
-            .await
-    }
-
    async fn in_use(&self) -> Result<bool, std::io::Error> {
        self.0.in_use().await
    }
@@ -108,23 +98,6 @@ impl WarmBackend for WarmBackendAzure {
 mod tests {
    use super::*;
    use crate::services::tier::tier_config::TierAzure;
-    use rustfs_s3_client::{
-        credentials::{Credentials, SignatureType, Static, Value},
-        transition_api::{Options, TransitionClient, TransitionCore},
-    };
-    use std::sync::Arc;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-
-    async fn read_request_head(stream: &mut tokio::net::TcpStream) -> String {
-        let mut request = Vec::new();
-        let mut buffer = [0_u8; 1024];
-        while !request.windows(4).any(|window| window == b"\r\n\r\n") {
-            let read = stream.read(&mut buffer).await.expect("fixture request should be readable");
-            assert_ne!(read, 0, "connection closed before request headers were received");
-            request.extend_from_slice(&buffer[..read]);
-        }
-        String::from_utf8_lossy(&request).into_owned()
-    }

    /// The SSRF guard itself is exercised once, generically, in
    /// `warm_backend::tests` (see backlog#2040/backlog#2041 and
@@ -146,85 +119,4 @@ mod tests {
            Err(err) => assert!(err.to_string().contains("not allowed")),
        }
    }
-
-    #[tokio::test]
-    async fn versioned_candidate_cleanup_uses_the_exact_s3_version_without_enabling_data_versions() {
-        let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
-            Err(err) => panic!("test listener should bind: {err}"),
-        };
-        let endpoint = listener
-            .local_addr()
-            .expect("listener local address should be available")
-            .to_string();
-        let fixture = tokio::spawn(async move {
-            let (mut get_stream, _) = listener.accept().await.expect("fixture should accept candidate GET");
-            let get_request = read_request_head(&mut get_stream).await;
-            get_stream
-                .write_all(
-                    b"HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nx-amz-version-id: azure-version\r\nConnection: close\r\n\r\nx",
-                )
-                .await
-                .expect("fixture should write candidate GET response");
-
-            let (mut delete_stream, _) = listener.accept().await.expect("fixture should accept exact DELETE");
-            let delete_request = read_request_head(&mut delete_stream).await;
-            delete_stream
-                .write_all(b"HTTP/1.1 204 No Content\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
-                .await
-                .expect("fixture should write exact DELETE response");
-            (get_request, delete_request)
-        });
-        let client = Arc::new(
-            TransitionClient::new(
-                &endpoint,
-                Options {
-                    creds: Credentials::new(Static(Value {
-                        access_key_id: "access-key".to_string(),
-                        secret_access_key: "secret-key".to_string(),
-                        signer_type: SignatureType::SignatureV4,
-                        ..Default::default()
-                    })),
-                    region: "us-east-1".to_string(),
-                    bucket_lookup: BucketLookupType::BucketLookupPath,
-                    max_retries: 1,
-                    ..Default::default()
-                },
-                "azure",
-            )
-            .await
-            .expect("fixture client should build"),
-        );
-        let backend = WarmBackendAzure(WarmBackendS3 {
-            core: TransitionCore(Arc::clone(&client)),
-            client,
-            bucket: "bucket".to_string(),
-            prefix: String::new(),
-            storage_class: String::new(),
-        });
-        assert!(
-            !backend.0.client.provider_version_capabilities().exact_get_delete,
-            "probe-only version discovery must not change Azure's persisted data-path contract"
-        );
-
-        let candidate = backend
-            .probe_transition_candidate("probe")
-            .await
-            .expect("Azure candidate should be discovered");
-        assert_eq!(candidate, TransitionCandidateProbe::VersionedPresent("azure-version".to_string()));
-        backend
-            .remove_exact("probe", "azure-version")
-            .await
-            .expect("Azure candidate should be deleted by exact version");
-
-        let (get_request, delete_request) = fixture.await.expect("fixture should join");
-        assert!(get_request.to_ascii_lowercase().contains("\r\nrange: bytes=0-0\r\n"));
-        assert!(
-            delete_request
-                .lines()
-                .next()
-                .is_some_and(|line| line.contains("DELETE /bucket/probe?versionId=azure-version "))
-        );
-    }
 }
@@ -18,13 +18,13 @@
 #![allow(unused_must_use)]
 #![allow(clippy::all)]

-use std::collections::{HashMap, HashSet};
-use std::future::Future;
+use std::collections::HashMap;
 use std::io::{Error, ErrorKind};
 use std::sync::Arc;

 use bytes::Bytes;
-use google_cloud_auth::credentials::service_account::Builder;
+use google_cloud_auth::credentials::Credentials;
+use google_cloud_auth::credentials::user_account::Builder;
 use google_cloud_storage as gcs;
 use google_cloud_storage::client::Storage;
 use google_cloud_storage::client::StorageControl;
@@ -32,7 +32,7 @@ use std::convert::TryFrom;

 use crate::services::tier::{
    tier_config::TierGCS,
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts},
+    warm_backend::{WarmBackend, WarmBackendGetOpts},
 };
 use rustfs_s3_client::{
    admin_handler_utils::AdminError,
@@ -43,7 +43,6 @@ use rustfs_utils::egress::validate_outbound_url;
 use tracing::warn;

 const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5;
-const MAX_GCS_CANDIDATE_PAGES: usize = 64;

 fn parse_generation(remote_version: &str) -> Result<Option<i64>, Error> {
    if remote_version.is_empty() {
@@ -58,85 +57,6 @@ fn parse_generation(remote_version: &str) -> Result<Option<i64>, Error> {
    Ok(Some(generation))
 }

-fn append_gcs_chunk<E: std::fmt::Display>(
-    contents: &mut Vec<u8>,
-    chunk: Result<Bytes, E>,
-    max_response_bytes: Option<usize>,
-) -> std::io::Result<()> {
-    let chunk = chunk.map_err(|err| std::io::Error::other(err.to_string()))?;
-    if max_response_bytes.is_some_and(|limit| contents.len().saturating_add(chunk.len()) > limit) {
-        return Err(std::io::Error::new(
-            ErrorKind::InvalidData,
-            "GCS object response exceeded the configured byte limit",
-        ));
-    }
-    contents.extend_from_slice(&chunk);
-    Ok(())
-}
-
-fn gcs_bucket_resource_name(bucket: &str) -> String {
-    format!("projects/_/buckets/{bucket}")
-}
-
-struct GcsCandidateObject {
-    name: String,
-    generation: i64,
-}
-
-struct GcsCandidatePage {
-    objects: Vec<GcsCandidateObject>,
-    next_page_token: String,
-}
-
-async fn probe_exact_gcs_candidate<F, Fut>(
-    remote_object: &str,
-    mut fetch_page: F,
-) -> Result<TransitionCandidateProbe, std::io::Error>
-where
-    F: FnMut(String) -> Fut,
-    Fut: Future<Output = Result<GcsCandidatePage, std::io::Error>>,
-{
-    let mut page_token = String::new();
-    let mut seen_page_tokens = HashSet::new();
-    let mut generation = None;
-    let mut pages_seen = 0_usize;
-
-    loop {
-        pages_seen += 1;
-        if pages_seen > MAX_GCS_CANDIDATE_PAGES {
-            return Err(std::io::Error::new(
-                ErrorKind::InvalidData,
-                "GCS candidate listing exceeded the page limit",
-            ));
-        }
-        let response = fetch_page(page_token.clone()).await?;
-        for candidate in response.objects.iter().filter(|candidate| candidate.name == remote_object) {
-            if candidate.generation <= 0 {
-                return Err(std::io::Error::new(
-                    ErrorKind::InvalidData,
-                    "GCS candidate listing returned a non-positive generation",
-                ));
-            }
-            if generation.replace(candidate.generation).is_some() {
-                return Ok(TransitionCandidateProbe::Ambiguous);
-            }
-        }
-
-        if response.next_page_token.is_empty() {
-            break;
-        }
-        if !seen_page_tokens.insert(response.next_page_token.clone()) {
-            return Err(std::io::Error::new(ErrorKind::InvalidData, "GCS candidate listing repeated a page token"));
-        }
-        page_token = response.next_page_token;
-    }
-
-    Ok(match generation {
-        Some(generation) => TransitionCandidateProbe::VersionedPresent(generation.to_string()),
-        None => TransitionCandidateProbe::Missing,
-    })
-}
-
 pub struct WarmBackendGCS {
    pub client: Arc<Storage>,
    pub control: Arc<StorageControl>,
@@ -160,8 +80,8 @@ impl WarmBackendGCS {
                .map_err(|err| std::io::Error::other(format!("tier endpoint is not allowed: {err}")))?;
        }

-        let service_account = serde_json::from_str(&conf.creds)?;
-        let credentials = Builder::new(service_account)
+        let authorized_user = serde_json::from_str(&conf.creds)?;
+        let credentials = Builder::new(authorized_user)
            //.with_retry_policy(AlwaysRetry.with_attempt_limit(3))
            //.with_backoff_policy(backoff)
            .build()
@@ -178,11 +98,7 @@ impl WarmBackendGCS {
        let client = Arc::new(client);
        // Control-plane client: the data-plane `Storage` client cannot delete or list objects;
        // delete_object/list_objects live on StorageControl.
-        let mut control_builder = StorageControl::builder().with_credentials(credentials);
-        if !conf.endpoint.is_empty() {
-            control_builder = control_builder.with_endpoint(conf.endpoint.clone());
-        }
-        let Ok(control) = control_builder.build().await else {
+        let Ok(control) = StorageControl::builder().with_credentials(credentials).build().await else {
            return Err(std::io::Error::other("StorageControl::builder error"));
        };
        let control = Arc::new(control);
@@ -220,10 +136,9 @@ impl WarmBackend for WarmBackendGCS {
            ReaderImpl::Body(content_body) => content_body.to_vec(),
            ReaderImpl::ObjectBody(mut content_body) => content_body.read_all().await?,
        };
-        let bucket = gcs_bucket_resource_name(&self.bucket);
        let Ok(res) = Box::pin(
            self.client
-                .write_object(&bucket, &self.get_dest(object), Bytes::from(d))
+                .write_object(&self.bucket, &self.get_dest(object), Bytes::from(d))
                .send_buffered(),
        )
        .await
@@ -239,9 +154,7 @@ impl WarmBackend for WarmBackendGCS {
    }

    async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-        let bucket = gcs_bucket_resource_name(&self.bucket);
-        let mut req = self.client.read_object(&bucket, &self.get_dest(object));
-        let mut max_response_bytes = None;
+        let mut req = self.client.read_object(&self.bucket, &self.get_dest(object));
        if let Some(generation) = parse_generation(rv)? {
            req = req.set_generation(generation);
        }
@@ -257,11 +170,6 @@ impl WarmBackend for WarmBackendGCS {
                .length
                .try_into()
                .map_err(|_| std::io::Error::other("invalid range: negative length"))?;
-            max_response_bytes = Some(
-                opts.length
-                    .try_into()
-                    .map_err(|_| std::io::Error::other("invalid range: length does not fit in memory"))?,
-            );
            req = req.set_read_range(google_cloud_storage::model_ext::ReadRange::segment(offset, count));
        }

@@ -269,8 +177,8 @@ impl WarmBackend for WarmBackendGCS {
            return Err(std::io::Error::other("read_object error"));
        };
        let mut contents = Vec::new();
-        while let Some(chunk) = reader.next().await {
-            append_gcs_chunk(&mut contents, chunk, max_response_bytes)?;
+        while let Ok(Some(chunk)) = reader.next().await.transpose() {
+            contents.extend_from_slice(&chunk);
        }
        Ok(ReadCloser::new(std::io::Cursor::new(contents)))
    }
@@ -282,7 +190,7 @@ impl WarmBackend for WarmBackendGCS {
        let mut req = self
            .control
            .delete_object()
-            .set_bucket(gcs_bucket_resource_name(&self.bucket))
+            .set_bucket(format!("projects/_/buckets/{}", self.bucket))
            .set_object(self.get_dest(object));
        if let Some(generation) = parse_generation(rv)? {
            req = req.set_generation(generation);
@@ -291,47 +199,13 @@ impl WarmBackend for WarmBackendGCS {
        Ok(())
    }

-    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        let remote_object = self.get_dest(object);
-        let parent = gcs_bucket_resource_name(&self.bucket);
-        probe_exact_gcs_candidate(&remote_object, |page_token| {
-            let control = self.control.clone();
-            let parent = parent.clone();
-            let prefix = remote_object.clone();
-            async move {
-                let response = control
-                    .list_objects()
-                    .set_parent(parent)
-                    .set_prefix(prefix)
-                    .set_versions(true)
-                    .set_page_size(2)
-                    .set_page_token(page_token)
-                    .send()
-                    .await
-                    .map_err(|err| std::io::Error::other(err.to_string()))?;
-                Ok(GcsCandidatePage {
-                    objects: response
-                        .objects
-                        .into_iter()
-                        .map(|candidate| GcsCandidateObject {
-                            name: candidate.name,
-                            generation: candidate.generation,
-                        })
-                        .collect(),
-                    next_page_token: response.next_page_token,
-                })
-            }
-        })
-        .await
-    }
-
    async fn in_use(&self) -> Result<bool, std::io::Error> {
        // Scope the listing to this tier's prefix (matching the other warm backends) and only
        // need to know whether a single object exists.
        let resp = self
            .control
            .list_objects()
-            .set_parent(gcs_bucket_resource_name(&self.bucket))
+            .set_parent(format!("projects/_/buckets/{}", self.bucket))
            .set_prefix(self.prefix.clone())
            .set_page_size(1)
            .send()
@@ -344,126 +218,10 @@ impl WarmBackend for WarmBackendGCS {

 #[cfg(test)]
 mod tests {
-    use super::GcsCandidateObject;
-    use super::GcsCandidatePage;
-    use super::MAX_GCS_CANDIDATE_PAGES;
    use super::WarmBackendGCS;
-    use super::append_gcs_chunk;
-    use super::gcs_bucket_resource_name;
    use super::parse_generation;
-    use super::probe_exact_gcs_candidate;
    use crate::services::tier::tier_config::TierGCS;
-    use crate::services::tier::warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts};
-    use google_cloud_auth::credentials::anonymous::Builder as Anonymous;
-    use google_cloud_storage::client::{Storage, StorageControl};
    use std::io::ErrorKind;
-    use std::sync::Arc;
-    use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};
-    use tokio::net::TcpListener;
-
-    async fn serve_data_plane_fixture(listener: TcpListener) -> Vec<String> {
-        let upload_body = r#"{"name":"probe","bucket":"tier-bucket","generation":"123"}"#;
-        let responses = [
-            format!(
-                "HTTP/1.1 200 OK\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{upload_body}",
-                upload_body.len()
-            ),
-            "HTTP/1.1 206 Partial Content\r\ncontent-type: application/octet-stream\r\ncontent-range: bytes 0-6/7\r\nx-goog-generation: 123\r\ncontent-length: 7\r\nconnection: close\r\n\r\nRustFS!"
-                .to_string(),
-            "HTTP/1.1 206 Partial Content\r\ncontent-type: application/octet-stream\r\ncontent-range: bytes 0-7/8\r\nx-goog-generation: 123\r\ncontent-length: 8\r\nconnection: close\r\n\r\nRustFS!!"
-                .to_string(),
-        ];
-        let mut requests = Vec::new();
-
-        for response in responses {
-            let (mut stream, _) = listener.accept().await.expect("the GCS fixture should accept a request");
-            let mut request = Vec::new();
-            loop {
-                let mut chunk = [0_u8; 1024];
-                let count = stream
-                    .read(&mut chunk)
-                    .await
-                    .expect("the GCS fixture should read request headers");
-                if count == 0 {
-                    break;
-                }
-                request.extend_from_slice(&chunk[..count]);
-                if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                    break;
-                }
-            }
-            let header_end = request
-                .windows(4)
-                .position(|window| window == b"\r\n\r\n")
-                .map(|position| position + 4)
-                .expect("the GCS fixture should receive complete request headers");
-            let headers = String::from_utf8_lossy(&request[..header_end]);
-            if headers.lines().any(|line| line.eq_ignore_ascii_case("expect: 100-continue")) {
-                stream
-                    .write_all(b"HTTP/1.1 100 Continue\r\n\r\n")
-                    .await
-                    .expect("the GCS fixture should acknowledge 100-continue");
-            }
-            let content_length = headers
-                .lines()
-                .find_map(|line| {
-                    let (name, value) = line.split_once(':')?;
-                    name.eq_ignore_ascii_case("content-length")
-                        .then(|| value.trim().parse::<usize>().expect("content-length should be numeric"))
-                })
-                .unwrap_or_default();
-            while request.len() < header_end.saturating_add(content_length) {
-                let mut chunk = [0_u8; 1024];
-                let count = stream
-                    .read(&mut chunk)
-                    .await
-                    .expect("the GCS fixture should read the request body");
-                if count == 0 {
-                    break;
-                }
-                request.extend_from_slice(&chunk[..count]);
-            }
-            requests.push(String::from_utf8_lossy(&request).into_owned());
-            stream
-                .write_all(response.as_bytes())
-                .await
-                .expect("the GCS fixture should write its response");
-        }
-
-        requests
-    }
-
-    fn candidate_page(objects: &[(&str, i64)], next_page_token: &str) -> GcsCandidatePage {
-        GcsCandidatePage {
-            objects: objects
-                .iter()
-                .map(|(name, generation)| GcsCandidateObject {
-                    name: (*name).to_string(),
-                    generation: *generation,
-                })
-                .collect(),
-            next_page_token: next_page_token.to_string(),
-        }
-    }
-
-    async fn probe_candidate_pages(
-        remote_object: &str,
-        pages: Vec<GcsCandidatePage>,
-    ) -> (Result<TransitionCandidateProbe, std::io::Error>, Vec<String>) {
-        let mut pages = pages.into_iter();
-        let mut requested_tokens = Vec::new();
-        let result = probe_exact_gcs_candidate(remote_object, |page_token| {
-            requested_tokens.push(page_token);
-            std::future::ready(
-                pages
-                    .next()
-                    .ok_or_else(|| std::io::Error::new(ErrorKind::UnexpectedEof, "test fixture ran out of GCS pages")),
-            )
-        })
-        .await;
-        (result, requested_tokens)
-    }

    #[test]
    fn generation_parser_preserves_exact_numeric_versions() {
@@ -483,243 +241,6 @@ mod tests {
        }
    }

-    #[test]
-    fn body_collection_propagates_an_error_after_a_complete_prefix() {
-        let mut contents = Vec::new();
-        append_gcs_chunk::<std::io::Error>(&mut contents, Ok(bytes::Bytes::from_static(b"RustFS")), Some(7))
-            .expect("the prefix chunk should be collected");
-        let err = append_gcs_chunk(&mut contents, Err(std::io::Error::other("trailing stream failure")), Some(7))
-            .expect_err("a trailing stream error must not be mistaken for EOF");
-
-        assert_eq!(contents, b"RustFS");
-        assert!(err.to_string().contains("trailing stream failure"));
-    }
-
-    #[test]
-    fn body_collection_rejects_a_chunk_that_exceeds_the_probe_limit() {
-        let mut contents = Vec::new();
-        let err = append_gcs_chunk::<std::io::Error>(&mut contents, Ok(bytes::Bytes::from_static(b"RustFSxx")), Some(7))
-            .expect_err("the GCS collection layer must reject an oversized probe response");
-
-        assert!(contents.is_empty());
-        assert_eq!(err.kind(), ErrorKind::InvalidData);
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_finds_exact_name_on_first_or_later_page() {
-        let (first, first_tokens) =
-            probe_candidate_pages("prefix/object", vec![candidate_page(&[("prefix/object", 7)], "")]).await;
-        assert_eq!(
-            first.expect("an exact first-page object should be discovered"),
-            TransitionCandidateProbe::VersionedPresent("7".to_string())
-        );
-        assert_eq!(first_tokens, [""]);
-
-        let (later, later_tokens) = probe_candidate_pages(
-            "prefix/object",
-            vec![
-                candidate_page(&[("prefix/object-shadow", 8)], "next"),
-                candidate_page(&[("prefix/object", 9)], ""),
-            ],
-        )
-        .await;
-        assert_eq!(
-            later.expect("an exact later-page object should be discovered"),
-            TransitionCandidateProbe::VersionedPresent("9".to_string())
-        );
-        assert_eq!(later_tokens, ["", "next"]);
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_ignores_non_exact_prefix_matches() {
-        let (probe, _) = probe_candidate_pages(
-            "prefix/object",
-            vec![candidate_page(
-                &[("prefix/object-shadow", 8), ("prefix/object/child", 9), ("prefix/object", 7)],
-                "",
-            )],
-        )
-        .await;
-
-        assert_eq!(
-            probe.expect("prefix-only matches should not hide the exact object"),
-            TransitionCandidateProbe::VersionedPresent("7".to_string())
-        );
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_reports_duplicate_exact_names_as_ambiguous() {
-        let (probe, _) = probe_candidate_pages(
-            "prefix/object",
-            vec![
-                candidate_page(&[("prefix/object", 7)], "next"),
-                candidate_page(&[("prefix/object", 8)], ""),
-            ],
-        )
-        .await;
-
-        assert_eq!(
-            probe.expect("multiple exact generations should produce a conservative result"),
-            TransitionCandidateProbe::Ambiguous
-        );
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_reports_missing_without_an_exact_name() {
-        let (probe, _) = probe_candidate_pages("prefix/object", vec![candidate_page(&[("prefix/object-shadow", 8)], "")]).await;
-
-        assert_eq!(
-            probe.expect("a complete listing without an exact name should be definitive"),
-            TransitionCandidateProbe::Missing
-        );
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_rejects_non_positive_generations() {
-        for generation in [0, -1] {
-            let (probe, _) =
-                probe_candidate_pages("prefix/object", vec![candidate_page(&[("prefix/object", generation)], "")]).await;
-            let err = probe.expect_err("a non-positive GCS generation must fail closed");
-            assert_eq!(err.kind(), ErrorKind::InvalidData, "generation {generation}");
-        }
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_rejects_a_page_token_that_does_not_advance() {
-        let (probe, requested_tokens) =
-            probe_candidate_pages("prefix/object", vec![candidate_page(&[], "next"), candidate_page(&[], "next")]).await;
-
-        let err = probe.expect_err("a repeated GCS page token must fail closed");
-        assert_eq!(err.kind(), ErrorKind::InvalidData);
-        assert_eq!(requested_tokens, ["", "next"]);
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_rejects_a_non_adjacent_page_token_cycle() {
-        let (probe, requested_tokens) = probe_candidate_pages(
-            "prefix/object",
-            vec![candidate_page(&[], "a"), candidate_page(&[], "b"), candidate_page(&[], "a")],
-        )
-        .await;
-
-        let err = probe.expect_err("a non-adjacent GCS page token cycle must fail closed");
-        assert_eq!(err.kind(), ErrorKind::InvalidData);
-        assert_eq!(requested_tokens, ["", "a", "b"]);
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_rejects_an_unbounded_unique_token_chain() {
-        let pages = (0..MAX_GCS_CANDIDATE_PAGES)
-            .map(|index| candidate_page(&[], &format!("token-{index}")))
-            .collect();
-        let (probe, requested_tokens) = probe_candidate_pages("prefix/object", pages).await;
-
-        let err = probe.expect_err("an unbounded unique page-token chain must fail closed");
-        assert_eq!(err.kind(), ErrorKind::InvalidData);
-        assert_eq!(requested_tokens.len(), MAX_GCS_CANDIDATE_PAGES);
-    }
-
-    #[tokio::test]
-    async fn plain_bucket_reaches_gcs_put_and_get_transport_with_resource_name() {
-        let listener = TcpListener::bind("127.0.0.1:0")
-            .await
-            .expect("the GCS fixture should bind a loopback port");
-        let endpoint = format!("http://{}", listener.local_addr().expect("the GCS fixture should have a local address"));
-        let fixture = tokio::spawn(serve_data_plane_fixture(listener));
-        let credentials = Anonymous::new().build();
-        let client = Storage::builder()
-            .with_endpoint(endpoint.clone())
-            .with_credentials(credentials.clone())
-            .build()
-            .await
-            .expect("the GCS data client should build");
-        let control = StorageControl::builder()
-            .with_endpoint(endpoint)
-            .with_credentials(credentials)
-            .build()
-            .await
-            .expect("the GCS control client should build");
-        let backend = WarmBackendGCS {
-            client: Arc::new(client),
-            control: Arc::new(control),
-            bucket: "tier-bucket".to_string(),
-            prefix: String::new(),
-        };
-
-        let (version, body, oversized_error_kind, requests) = tokio::time::timeout(Duration::from_secs(5), async {
-            let version = backend
-                .put(
-                    "probe",
-                    rustfs_s3_client::transition_api::ReaderImpl::Body(bytes::Bytes::from_static(b"RustFS")),
-                    6,
-                )
-                .await
-                .expect("a plain configured bucket should reach the GCS upload transport");
-            let mut reader = backend
-                .get(
-                    "probe",
-                    &version,
-                    WarmBackendGetOpts {
-                        start_offset: 0,
-                        length: 7,
-                    },
-                )
-                .await
-                .expect("a plain configured bucket should reach the GCS read transport");
-            let mut body = Vec::new();
-            reader
-                .read_to_end(&mut body)
-                .await
-                .expect("the fixture body should be readable");
-            let oversized_error = match backend
-                .get(
-                    "probe",
-                    &version,
-                    WarmBackendGetOpts {
-                        start_offset: 0,
-                        length: 7,
-                    },
-                )
-                .await
-            {
-                Ok(_) => panic!("an eight-byte response must not pass a seven-byte collection limit"),
-                Err(err) => err,
-            };
-            let requests = fixture.await.expect("the GCS fixture task should finish");
-            (version, body, oversized_error.kind(), requests)
-        })
-        .await
-        .expect("the GCS data-plane requests should not be rejected before transport");
-
-        assert_eq!(gcs_bucket_resource_name("tier-bucket"), "projects/_/buckets/tier-bucket");
-        assert_eq!(version, "123");
-        assert_eq!(body, b"RustFS!");
-        assert_eq!(oversized_error_kind, ErrorKind::InvalidData);
-        assert!(
-            requests[0].starts_with("POST /upload/storage/v1/b/tier-bucket/o?"),
-            "unexpected upload request line: {}",
-            requests[0].lines().next().unwrap_or_default()
-        );
-        assert!(
-            requests[1].starts_with("GET /storage/v1/b/tier-bucket/o/probe?"),
-            "unexpected read request line: {}",
-            requests[1].lines().next().unwrap_or_default()
-        );
-        assert!(
-            requests[1].to_ascii_lowercase().contains("\r\nrange: bytes=0-6\r\n"),
-            "the GCS probe read must preserve its seven-byte range"
-        );
-        assert!(
-            requests[2].starts_with("GET /storage/v1/b/tier-bucket/o/probe?"),
-            "unexpected oversized read request line: {}",
-            requests[2].lines().next().unwrap_or_default()
-        );
-        assert!(
-            requests[2].to_ascii_lowercase().contains("\r\nrange: bytes=0-6\r\n"),
-            "the oversized response must be fetched under the same seven-byte request boundary"
-        );
-    }
-
    #[tokio::test]
    async fn new_rejects_loopback_endpoint_before_credential_setup() {
        let conf = TierGCS {
@@ -23,7 +23,7 @@ use std::collections::HashMap;
 use crate::services::tier::{
    tier_config::TierHuaweicloud,
    warm_backend::{
-        S3CompatibleWarmBackendParams, TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
+        S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
        new_s3_compatible_warm_backend, optimal_part_size,
    },
    warm_backend_s3::WarmBackendS3,
@@ -89,10 +89,6 @@ impl WarmBackend for WarmBackendHuaweicloud {
        self.0.remove(object, rv).await
    }

-    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        self.0.probe_transition_candidate(object).await
-    }
-
    async fn in_use(&self) -> Result<bool, std::io::Error> {
        self.0.in_use().await
    }
@@ -26,12 +26,11 @@ use crate::services::tier::{
    tier_config::TierS3,
    warm_backend::{
        TransitionCandidateIdentity, TransitionCandidateProbe, TransitionCandidateReconciler, WarmBackend, WarmBackendGetOpts,
-        build_transition_put_options, endpoint_authority,
+        build_transition_put_options,
    },
 };
 use http::HeaderMap;
 use rustfs_s3_client::{
-    api_error_response::to_error_response,
    api_get_options::GetObjectOptions,
    api_list::ListObjectsOptions,
    api_put_object::PutObjectOptions,
@@ -44,7 +43,7 @@ use rustfs_s3_client::{
 };
 use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::path::SLASH_SEPARATOR;
-use s3s::{S3ErrorCode, dto::BucketVersioningStatus};
+use s3s::dto::BucketVersioningStatus;

 pub struct WarmBackendS3 {
    pub client: Arc<TransitionClient>,
@@ -75,19 +74,6 @@ fn remote_bucket_versioning_from_status(status: Option<&str>) -> Result<RemoteBu
    })
 }

-fn bounded_get_range(opts: &WarmBackendGetOpts) -> Result<Option<(i64, i64)>, std::io::Error> {
-    if opts.start_offset < 0 || opts.length <= 0 {
-        return Ok(None);
-    }
-    usize::try_from(opts.length)
-        .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "invalid range: length does not fit in memory"))?;
-    let end_offset = opts
-        .start_offset
-        .checked_add(opts.length - 1)
-        .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "invalid range: end offset overflow"))?;
-    Ok(Some((opts.start_offset, end_offset)))
-}
-
 impl WarmBackendS3 {
    pub async fn new(conf: &TierS3, _tier: &str) -> Result<Self, std::io::Error> {
        Self::new_with_bucket_lookup(conf, BucketLookupType::BucketLookupAuto, "s3").await
@@ -146,8 +132,10 @@ impl WarmBackendS3 {
            bucket_lookup,
            ..Default::default()
        };
-        let endpoint = endpoint_authority(&u)?;
-        let client = TransitionClient::new(&endpoint, opts, tier_type).await?;
+        let host = u
+            .host()
+            .ok_or_else(|| std::io::Error::other("Invalid endpoint URL: missing host"))?;
+        let client = TransitionClient::new(&host.to_string(), opts, tier_type).await?;

        let client = Arc::new(client);
        let core = TransitionCore(Arc::clone(&client));
@@ -189,8 +177,10 @@ impl WarmBackendS3 {
        if !rv.is_empty() {
            gopts.version_id = rv.to_string();
        }
-        if let Some((start_offset, end_offset)) = bounded_get_range(&opts)? {
-            gopts.set_range(start_offset, end_offset)?;
+        if opts.start_offset >= 0 && opts.length > 0 {
+            gopts
+                .set_range(opts.start_offset, opts.start_offset + opts.length - 1)
+                .map_err(std::io::Error::other)?;
        }
        let (_, headers, reader) = self.core.get_object(&self.bucket, &self.get_dest(object), &gopts).await?;
        Ok((headers, reader))
@@ -201,62 +191,34 @@ impl WarmBackendS3 {
        remote_bucket_versioning_from_status(config.status.as_ref().map(|status| status.as_str()))
    }

-    async fn probe_current_transition_candidate_with_header(
+    async fn probe_transition_candidate_versions(
        &self,
        object: &str,
-        raw_version_header: Option<&'static str>,
+        bucket_versioning: RemoteBucketVersioning,
    ) -> Result<TransitionCandidateProbe, std::io::Error> {
-        match self
-            .get_with_headers(
-                object,
-                "",
-                WarmBackendGetOpts {
-                    start_offset: 0,
-                    length: 1,
-                },
-            )
-            .await
-        {
-            Ok((headers, _)) => {
-                let version_id = match raw_version_header {
-                    Some(header_name) => match headers.get(header_name) {
-                        Some(value) => {
-                            let version_id = value.to_str().map_err(|_| {
-                                std::io::Error::new(
-                                    std::io::ErrorKind::InvalidData,
-                                    "remote object version id is not valid ASCII",
-                                )
-                            })?;
-                            validate_remote_version_id(version_id)?;
-                            Some(version_id)
-                        }
-                        None => None,
-                    },
-                    None => self.client.raw_version_id(&headers)?,
-                };
-                Ok(match version_id {
-                    Some(version_id) => TransitionCandidateProbe::VersionedPresent(version_id.to_string()),
-                    None => TransitionCandidateProbe::UnversionedPresent,
-                })
-            }
-            Err(err) => {
-                let response = to_error_response(&err);
-                if response.code == S3ErrorCode::NoSuchKey {
-                    Ok(TransitionCandidateProbe::Missing)
-                } else {
-                    Err(err)
-                }
-            }
-        }
-    }
+        let remote_object = self.get_dest(object);
+        let mut opts = ListObjectsOptions::default();
+        opts.set("prefix", &remote_object);
+        opts.set("max-keys", "1000");

-    pub(crate) async fn probe_transition_candidate_with_raw_version_header(
-        &self,
-        object: &str,
-        raw_version_header: &'static str,
-    ) -> Result<TransitionCandidateProbe, std::io::Error> {
-        self.probe_current_transition_candidate_with_header(object, Some(raw_version_header))
-            .await
+        let mut key_marker = String::new();
+        let mut version_id_marker = String::new();
+        let mut candidates = TransitionCandidateVersions::default();
+        loop {
+            let versions = self
+                .client
+                .list_object_versions_query(&self.bucket, &opts, &key_marker, &version_id_marker, "")
+                .await?;
+            candidates.extend(&remote_object, &versions);
+            if candidates.is_ambiguous() {
+                return Ok(TransitionCandidateProbe::Ambiguous);
+            }
+            if !versions.is_truncated {
+                return classify_transition_candidates(candidates, bucket_versioning);
+            }
+
+            advance_version_markers(&mut key_marker, &mut version_id_marker, &versions)?;
+        }
    }

    async fn probe_transition_candidate_identity(
@@ -381,7 +343,6 @@ struct TransitionCandidateVersions {
 }

 impl TransitionCandidateVersions {
-    #[cfg(test)]
    fn extend(&mut self, remote_object: &str, versions: &ListVersionsResult) {
        for version in versions.versions.iter().filter(|version| version.key == remote_object) {
            if self.version_id.is_some() {
@@ -392,6 +353,10 @@ impl TransitionCandidateVersions {
        }
    }

+    fn is_ambiguous(&self) -> bool {
+        self.ambiguous
+    }
+
    fn classify(self, bucket_versioning: RemoteBucketVersioning) -> TransitionCandidateProbe {
        if self.ambiguous {
            return TransitionCandidateProbe::Ambiguous;
@@ -415,8 +380,6 @@ impl TransitionCandidateVersions {
 mod tests {
    use super::*;
    use rustfs_s3_client::api_s3_datatypes::{ListVersionsResult, Version};
-    use std::time::Duration;
-    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    #[tokio::test]
    async fn new_rejects_loopback_endpoint_before_network_setup() {
@@ -435,204 +398,6 @@ mod tests {
        }
    }

-    #[tokio::test]
-    async fn new_preserves_an_explicit_endpoint_port() {
-        let conf = TierS3 {
-            endpoint: "https://tier.example.com:9443".to_string(),
-            bucket: "tier-bucket".to_string(),
-            access_key: "access".to_string(),
-            secret_key: "secret".to_string(),
-            region: "us-east-1".to_string(),
-            ..Default::default()
-        };
-
-        let backend = WarmBackendS3::new(&conf, "tier")
-            .await
-            .expect("a well-formed S3 endpoint should initialize without network I/O");
-        assert_eq!(backend.client.endpoint_url.host_str(), Some("tier.example.com"));
-        assert_eq!(backend.client.endpoint_url.port(), Some(9443));
-    }
-
-    #[tokio::test]
-    async fn overflowing_get_range_is_rejected_before_network_io() {
-        let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
-            Err(err) => panic!("test listener should bind: {err}"),
-        };
-        let endpoint = listener
-            .local_addr()
-            .expect("listener local address should be available")
-            .to_string();
-        let client = Arc::new(
-            TransitionClient::new(
-                &endpoint,
-                Options {
-                    creds: Credentials::new(Static(Value {
-                        access_key_id: "access-key".to_string(),
-                        secret_access_key: "secret-key".to_string(),
-                        signer_type: SignatureType::SignatureV4,
-                        ..Default::default()
-                    })),
-                    region: "us-east-1".to_string(),
-                    bucket_lookup: BucketLookupType::BucketLookupPath,
-                    max_retries: 1,
-                    ..Default::default()
-                },
-                "s3",
-            )
-            .await
-            .expect("fixture client should build"),
-        );
-        let backend = WarmBackendS3 {
-            core: TransitionCore(Arc::clone(&client)),
-            client,
-            bucket: "bucket".to_string(),
-            prefix: String::new(),
-            storage_class: String::new(),
-        };
-
-        let err = backend
-            .get_with_headers(
-                "probe",
-                "",
-                WarmBackendGetOpts {
-                    start_offset: i64::MAX,
-                    length: 2,
-                },
-            )
-            .await
-            .expect_err("an overflowing range must fail before issuing a GET");
-
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
-        assert!(
-            tokio::time::timeout(Duration::from_millis(100), listener.accept())
-                .await
-                .is_err()
-        );
-    }
-
-    async fn candidate_probe_fixture() -> Option<(WarmBackendS3, tokio::task::JoinHandle<Vec<String>>)> {
-        let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
-            Err(err) => panic!("test listener should bind: {err}"),
-        };
-        let endpoint = listener
-            .local_addr()
-            .expect("listener local address should be available")
-            .to_string();
-        let fixture = tokio::spawn(async move {
-            let responses = [
-                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nx-amz-version-id: opaque-version\r\nConnection: close\r\n\r\nx",
-                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nConnection: close\r\n\r\nx",
-                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
-                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 66\r\nConnection: close\r\n\r\n<Error><Code>NoSuchObject</Code><Message>missing</Message></Error>",
-                "HTTP/1.1 403 Forbidden\r\nContent-Type: application/xml\r\nContent-Length: 65\r\nConnection: close\r\n\r\n<Error><Code>AccessDenied</Code><Message>denied</Message></Error>",
-            ];
-            let mut requests = Vec::new();
-            for response in responses {
-                let (mut stream, _) = listener.accept().await.expect("fixture should accept candidate GET");
-                let mut request = Vec::new();
-                let mut buffer = [0; 1024];
-                loop {
-                    let read = stream.read(&mut buffer).await.expect("fixture should read request headers");
-                    assert_ne!(read, 0, "connection closed before request headers were received");
-                    request.extend_from_slice(&buffer[..read]);
-                    if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                        break;
-                    }
-                }
-                requests.push(String::from_utf8_lossy(&request).into_owned());
-                stream
-                    .write_all(response.as_bytes())
-                    .await
-                    .expect("fixture should write candidate response");
-            }
-            requests
-        });
-        let client = Arc::new(
-            TransitionClient::new(
-                &endpoint,
-                Options {
-                    creds: Credentials::new(Static(Value {
-                        access_key_id: "access-key".to_string(),
-                        secret_access_key: "secret-key".to_string(),
-                        signer_type: SignatureType::SignatureV4,
-                        ..Default::default()
-                    })),
-                    region: "us-east-1".to_string(),
-                    bucket_lookup: BucketLookupType::BucketLookupPath,
-                    max_retries: 1,
-                    ..Default::default()
-                },
-                "s3",
-            )
-            .await
-            .expect("fixture client should build"),
-        );
-        Some((
-            WarmBackendS3 {
-                core: TransitionCore(Arc::clone(&client)),
-                client,
-                bucket: "bucket".to_string(),
-                prefix: String::new(),
-                storage_class: String::new(),
-            },
-            fixture,
-        ))
-    }
-
-    #[tokio::test]
-    async fn candidate_probe_uses_only_exact_bounded_get_permissions() {
-        let Some((backend, fixture)) = candidate_probe_fixture().await else {
-            return;
-        };
-
-        assert_eq!(
-            backend
-                .probe_transition_candidate("versioned-probe")
-                .await
-                .expect("versioned candidate should be discovered"),
-            TransitionCandidateProbe::VersionedPresent("opaque-version".to_string())
-        );
-        assert_eq!(
-            backend
-                .probe_transition_candidate("unversioned-probe")
-                .await
-                .expect("unversioned candidate should be discovered"),
-            TransitionCandidateProbe::UnversionedPresent
-        );
-        assert_eq!(
-            backend
-                .probe_transition_candidate("missing-probe")
-                .await
-                .expect("a missing key should be classified"),
-            TransitionCandidateProbe::Missing
-        );
-        assert_eq!(
-            backend
-                .probe_transition_candidate("provider-missing-probe")
-                .await
-                .expect("a provider-specific missing code should be classified"),
-            TransitionCandidateProbe::Missing
-        );
-        let err = backend
-            .probe_transition_candidate("forbidden-probe")
-            .await
-            .expect_err("an authorization failure must not be mistaken for a missing key");
-        assert_eq!(to_error_response(&err).code, S3ErrorCode::AccessDenied);
-
-        let requests = fixture.await.expect("candidate fixture should join");
-        for request in requests {
-            let request = request.to_ascii_lowercase();
-            assert!(request.starts_with("get /bucket/"), "candidate discovery must use object GET");
-            assert!(request.contains("\r\nrange: bytes=0-0\r\n"));
-            assert!(!request.contains("?versioning"));
-            assert!(!request.contains("?versions"));
-        }
-    }
-
    fn list_versions(versions: &[(&str, &str)], delete_markers: &[(&str, &str)], is_truncated: bool) -> ListVersionsResult {
        ListVersionsResult {
            versions: versions
@@ -866,7 +631,8 @@ impl WarmBackend for WarmBackendS3 {
    }

    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        self.probe_current_transition_candidate_with_header(object, None).await
+        let bucket_versioning = self.remote_bucket_versioning().await?;
+        self.probe_transition_candidate_versions(object, bucket_versioning).await
    }

    async fn in_use(&self) -> Result<bool, std::io::Error> {
@@ -23,7 +23,7 @@ use std::collections::HashMap;
 use crate::services::tier::{
    tier_config::TierTencent,
    warm_backend::{
-        S3CompatibleWarmBackendParams, TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
+        S3CompatibleWarmBackendParams, WarmBackend, WarmBackendGetOpts, build_transition_put_options,
        new_s3_compatible_warm_backend, optimal_part_size,
    },
    warm_backend_s3::WarmBackendS3,
@@ -89,10 +89,6 @@ impl WarmBackend for WarmBackendTencent {
        self.0.remove(object, rv).await
    }

-    async fn probe_transition_candidate(&self, object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
-        self.0.probe_transition_candidate(object).await
-    }
-
    async fn in_use(&self) -> Result<bool, std::io::Error> {
        self.0.in_use().await
    }
@@ -23,7 +23,7 @@ use uuid::Uuid;

 use crate::services::tier::{
    tier_config::{TierS3, TierWasabi},
-    warm_backend::{TransitionCandidateProbe, WarmBackend, WarmBackendGetOpts},
+    warm_backend::{WarmBackend, WarmBackendGetOpts},
    warm_backend_s3::WarmBackendS3,
 };
 use rustfs_s3_client::transition_api::{BucketLookupType, ReadCloser, ReaderImpl};
@@ -169,10 +169,6 @@ impl WarmBackend for WarmBackendWasabi {
        self.s3.remove(object, rv).await
    }

-    async fn probe_transition_candidate(&self, object: &str) -> io::Result<TransitionCandidateProbe> {
-        self.s3.probe_transition_candidate(object).await
-    }
-
    async fn in_use(&self) -> io::Result<bool> {
        self.check_remote_bucket_unversioned().await?;
        let in_use = self.s3.in_use().await?;
@@ -10339,17 +10339,9 @@ mod tests {
    }

    #[cfg(feature = "test-util")]
-    #[test]
+    #[tokio::test]
    #[serial_test::serial(storage_class_env)]
-    fn v6_decommission_checkpoint_no_lock_put_rejects_lost_publication_fence() {
-        run_large_stack_async_test(
-            "v6-checkpoint-fence-loss",
-            v6_decommission_checkpoint_no_lock_put_rejects_lost_publication_fence_case,
-        );
-    }
-
-    #[cfg(feature = "test-util")]
-    async fn v6_decommission_checkpoint_no_lock_put_rejects_lost_publication_fence_case() {
+    async fn v6_decommission_checkpoint_no_lock_put_rejects_lost_publication_fence() {
        let temp_dir = tempfile::tempdir().expect("create v6 checkpoint fence-loss store dir");
        let (_ctx, store, _shutdown) =
            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "v6-checkpoint-fence-loss", &[4, 4])).await;
@@ -57,11 +57,7 @@ fn deserialize_code<'de, D>(d: D) -> Result<S3ErrorCode, D::Error>
 where
    D: Deserializer<'de>,
 {
-    let code = String::deserialize(d)?;
-    if code == "NoSuchObject" {
-        return Ok(S3ErrorCode::NoSuchKey);
-    }
-    Ok(S3ErrorCode::from_bytes(code.as_bytes()).unwrap_or(S3ErrorCode::Custom("".into())))
+    Ok(S3ErrorCode::from_bytes(String::deserialize(d)?.as_bytes()).unwrap_or(S3ErrorCode::Custom("".into())))
 }

 impl Default for ErrorResponse {
@@ -329,21 +325,4 @@ mod tests {
        assert_eq!(response.code, S3ErrorCode::NoSuchVersion);
        assert_eq!(response.status_code, StatusCode::NOT_FOUND);
    }
-
-    #[test]
-    fn normalizes_provider_specific_missing_object_code() {
-        let mut headers = HeaderMap::new();
-        headers.insert("x-amz-request-id", "request-id".parse().expect("request ID header should parse"));
-
-        let response = http_resp_to_error_response(
-            StatusCode::NOT_FOUND,
-            &headers,
-            b"<Error><Code>NoSuchObject</Code><Message>remote detail</Message></Error>".to_vec(),
-            "bucket",
-            "object",
-        );
-
-        assert_eq!(response.code, S3ErrorCode::NoSuchKey);
-        assert_eq!(response.status_code, StatusCode::NOT_FOUND);
-    }
 }
@@ -30,9 +30,7 @@ use tokio_util::io::StreamReader;
 use crate::{
    api_error_response::err_invalid_argument,
    api_get_options::GetObjectOptions,
-    transition_api::{
-        ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, collect_response_body, to_object_info_for_provider,
-    },
+    transition_api::{ObjectInfo, ReadCloser, ReaderImpl, RequestMetadata, TransitionClient, to_object_info_for_provider},
 };
 use futures_util::StreamExt;
 use http_body_util::BodyExt;
@@ -41,42 +39,6 @@ use hyper::body::Bytes;
 use rustfs_utils::hash::EMPTY_STRING_SHA256_HASH;
 use tokio_util::io::ReaderStream;

-fn response_limit_from_range(opts: &GetObjectOptions) -> Result<Option<usize>, std::io::Error> {
-    let Some(range) = opts
-        .headers
-        .iter()
-        .find_map(|(name, value)| name.eq_ignore_ascii_case("range").then_some(value.as_str()))
-    else {
-        return Ok(None);
-    };
-    let Some((unit, bounds)) = range.split_once('=') else {
-        return Ok(None);
-    };
-    if !unit.eq_ignore_ascii_case("bytes") {
-        return Ok(None);
-    }
-    let Some((start, end)) = bounds.split_once('-') else {
-        return Ok(None);
-    };
-    if start.is_empty() || end.is_empty() || end.contains(',') {
-        return Ok(None);
-    }
-    let start = start
-        .parse::<u64>()
-        .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "closed response range start is invalid"))?;
-    let end = end
-        .parse::<u64>()
-        .map_err(|_| std::io::Error::new(std::io::ErrorKind::InvalidInput, "closed response range end is invalid"))?;
-    let length = end
-        .checked_sub(start)
-        .and_then(|length| length.checked_add(1))
-        .ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidInput, "closed response range length overflows"))?;
-    let limit = usize::try_from(length).map_err(|_| {
-        std::io::Error::new(std::io::ErrorKind::InvalidInput, "closed response range length does not fit in memory")
-    })?;
-    Ok(Some(limit))
-}
-
 impl TransitionClient {
    pub fn get_object(&self, bucket_name: &str, object_name: &str, opts: &GetObjectOptions) -> Result<Object, std::io::Error> {
        let _ = opts;
@@ -92,7 +54,6 @@ impl TransitionClient {
        object_name: &str,
        opts: &GetObjectOptions,
    ) -> Result<(ObjectInfo, HeaderMap, ReadCloser), std::io::Error> {
-        let max_response_bytes = response_limit_from_range(opts)?;
        let resp = self
            .execute_method(
                http::Method::GET,
@@ -120,214 +81,18 @@ impl TransitionClient {

        let h = resp.headers().clone();

+        let mut body_vec = Vec::new();
        let mut body = resp.into_body();
-        let body_vec = if let Some(limit) = max_response_bytes {
-            collect_response_body(body, limit).await?
-        } else {
-            let mut body_vec = Vec::new();
-            while let Some(frame) = body.frame().await {
-                let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
-                if let Some(data) = frame.data_ref() {
-                    body_vec.extend_from_slice(data);
-                }
+        while let Some(frame) = body.frame().await {
+            let frame = frame.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e.to_string()))?;
+            if let Some(data) = frame.data_ref() {
+                body_vec.extend_from_slice(data);
            }
-            body_vec
-        };
+        }
        Ok((object_stat, h, BufReader::new(Cursor::new(body_vec))))
    }
 }

-#[cfg(test)]
-mod bounded_response_tests {
-    use super::response_limit_from_range;
-    use crate::{
-        api_get_options::GetObjectOptions,
-        credentials::{Credentials, SignatureType, Static, Value},
-        transition_api::{BucketLookupType, Options, TransitionClient, collect_response_body},
-    };
-    use http_body_util::Full;
-    use hyper::body::Bytes;
-    use std::time::Duration;
-    use tokio::{
-        io::{AsyncReadExt, AsyncWriteExt},
-        net::TcpListener,
-    };
-
-    #[test]
-    fn closed_range_derives_a_collection_limit_without_new_public_options() {
-        let mut opts = GetObjectOptions::default();
-        opts.set_range(5, 11).expect("the closed range should be valid");
-
-        assert_eq!(response_limit_from_range(&opts).expect("the range should parse"), Some(7));
-    }
-
-    #[tokio::test]
-    async fn response_collection_rejects_the_body_that_exceeds_its_range_limit() {
-        let mut opts = GetObjectOptions::default();
-        opts.set_range(0, 6).expect("the probe range should be valid");
-        let max_response_bytes = response_limit_from_range(&opts)
-            .expect("the range should parse")
-            .expect("the closed range should have a limit");
-        let err = collect_response_body(Full::new(Bytes::from_static(b"RustFSxx")), max_response_bytes)
-            .await
-            .expect_err("the collection layer must reject a response larger than its limit");
-
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-    }
-
-    async fn bounded_get_fixture(body: &'static [u8]) -> Option<(TransitionClient, tokio::task::JoinHandle<String>)> {
-        let listener = match TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
-            Err(err) => panic!("test listener should bind: {err}"),
-        };
-        let endpoint = listener
-            .local_addr()
-            .expect("listener local address should be available")
-            .to_string();
-        let request = tokio::spawn(async move {
-            let (mut stream, _) = listener.accept().await.expect("fixture should accept one GET");
-            let mut request = Vec::new();
-            let mut buffer = [0; 1024];
-            loop {
-                let read = stream.read(&mut buffer).await.expect("fixture should read request headers");
-                assert_ne!(read, 0, "connection closed before request headers were received");
-                request.extend_from_slice(&buffer[..read]);
-                if request.windows(4).any(|window| window == b"\r\n\r\n") {
-                    break;
-                }
-            }
-            let request = String::from_utf8_lossy(&request).into_owned();
-            let response = format!(
-                "HTTP/1.1 206 Partial Content\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
-                body.len()
-            );
-            stream
-                .write_all(response.as_bytes())
-                .await
-                .expect("fixture should write response headers");
-            stream.write_all(body).await.expect("fixture should write response body");
-            request
-        });
-        let client = TransitionClient::new(
-            &endpoint,
-            Options {
-                creds: Credentials::new(Static(Value {
-                    access_key_id: "access-key".to_string(),
-                    secret_access_key: "secret-key".to_string(),
-                    signer_type: SignatureType::SignatureV4,
-                    ..Default::default()
-                })),
-                region: "us-east-1".to_string(),
-                bucket_lookup: BucketLookupType::BucketLookupPath,
-                max_retries: 1,
-                ..Default::default()
-            },
-            "",
-        )
-        .await
-        .expect("fixture client should build");
-        Some((client, request))
-    }
-
-    #[tokio::test]
-    async fn real_transport_accepts_the_exact_closed_range_length() {
-        let Some((client, request)) = bounded_get_fixture(b"RustFS!").await else {
-            return;
-        };
-        let mut opts = GetObjectOptions::default();
-        opts.set_range(0, 6).expect("the probe range should be valid");
-
-        let (_, _, mut reader) = client
-            .get_object_inner("bucket", "probe", &opts)
-            .await
-            .expect("a seven-byte response should fit the requested range");
-        let mut body = Vec::new();
-        reader
-            .read_to_end(&mut body)
-            .await
-            .expect("bounded response should be readable");
-
-        assert_eq!(body, b"RustFS!");
-        assert!(
-            request
-                .await
-                .expect("fixture should join")
-                .to_ascii_lowercase()
-                .contains("\r\nrange: bytes=0-6\r\n")
-        );
-    }
-
-    #[tokio::test]
-    async fn real_transport_rejects_a_body_larger_than_the_closed_range() {
-        let Some((client, request)) = bounded_get_fixture(b"RustFS!!").await else {
-            return;
-        };
-        let mut opts = GetObjectOptions::default();
-        opts.set_range(0, 6).expect("the probe range should be valid");
-
-        let err = client
-            .get_object_inner("bucket", "probe", &opts)
-            .await
-            .expect_err("an eight-byte response must exceed the seven-byte range limit");
-
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
-        assert!(
-            request
-                .await
-                .expect("fixture should join")
-                .to_ascii_lowercase()
-                .contains("\r\nrange: bytes=0-6\r\n")
-        );
-    }
-
-    #[tokio::test]
-    async fn overflowing_closed_range_is_rejected_before_network_io() {
-        let listener = match TcpListener::bind("127.0.0.1:0").await {
-            Ok(listener) => listener,
-            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return,
-            Err(err) => panic!("test listener should bind: {err}"),
-        };
-        let endpoint = listener
-            .local_addr()
-            .expect("listener local address should be available")
-            .to_string();
-        let client = TransitionClient::new(
-            &endpoint,
-            Options {
-                creds: Credentials::new(Static(Value {
-                    access_key_id: "access-key".to_string(),
-                    secret_access_key: "secret-key".to_string(),
-                    signer_type: SignatureType::SignatureV4,
-                    ..Default::default()
-                })),
-                region: "us-east-1".to_string(),
-                bucket_lookup: BucketLookupType::BucketLookupPath,
-                max_retries: 1,
-                ..Default::default()
-            },
-            "",
-        )
-        .await
-        .expect("fixture client should build");
-        let mut opts = GetObjectOptions::default();
-        opts.headers
-            .insert("range".to_string(), "bytes=0-18446744073709551615".to_string());
-
-        let err = client
-            .get_object_inner("bucket", "probe", &opts)
-            .await
-            .expect_err("an overflowing closed range must be rejected locally");
-
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
-        assert!(
-            tokio::time::timeout(Duration::from_millis(100), listener.accept())
-                .await
-                .is_err()
-        );
-    }
-}
-
 #[derive(Default)]
 pub struct GetRequest {
    pub buffer: Vec<u8>,
@@ -105,12 +105,6 @@ struct DirtyUsageSnapshot {
    covers_all_pending: bool,
 }

-#[derive(Clone, Debug, Default)]
-pub(crate) struct ScannerBucketScanScope {
-    selected_buckets: Option<Arc<HashSet<String>>>,
-    baseline_scan_plan_digest: Option<DataUsageScanPlanDigest>,
-}
-
 pub(crate) fn is_scanner_metadata_corrupt_error(err: &StorageError) -> bool {
    matches!(err, StorageError::Io(io) if io.to_string().starts_with(SCANNER_METADATA_CORRUPT_ERROR))
 }
@@ -152,7 +146,6 @@ fn object_lock_config_enabled(config: &ObjectLockConfiguration) -> bool {
 pub struct ScannerBucketScanPlan {
    buckets: Vec<BucketInfo>,
    all_buckets: Arc<Vec<BucketInfo>>,
-    scope: ScannerBucketScanScope,
    digest: DataUsageScanPlanDigest,
    leader_epoch: u64,
    tier_registry_generation: u64,
@@ -739,8 +732,6 @@ mod dirty_usage;
 mod guards;
 mod io_cache;
 mod io_cycle;
-#[cfg(test)]
-use io_cache::{ScannerSetCacheGeneration, prepare_scoped_set_scan};
 pub(crate) use io_cycle::nsscanner_with_storage_status;
 mod io_disk;
 #[cfg(test)]
@@ -14,93 +14,6 @@
 /// ScannerIOCache implementation for SetDisks: bucket ordering, worker fan-out, merge, and publish.
 use super::*;

-#[derive(Clone, Copy)]
-pub(super) struct ScannerSetCacheGeneration {
-    pub(super) want_cycle: u64,
-    pub(super) leader_epoch: u64,
-    pub(super) tier_registry_generation: u64,
-    pub(super) source: DataUsageCacheSource,
-    pub(super) scan_plan_digest: DataUsageScanPlanDigest,
-}
-
-pub(super) struct PreparedScopedSetScan {
-    pub(super) buckets: Vec<BucketInfo>,
-    pub(super) cache: DataUsageCache,
-}
-
-pub(super) fn prepare_scoped_set_scan(
-    old_cache: &DataUsageCache,
-    set_buckets: &[BucketInfo],
-    all_buckets: &[BucketInfo],
-    scope: &ScannerBucketScanScope,
-    generation: ScannerSetCacheGeneration,
-) -> Option<PreparedScopedSetScan> {
-    let (Some(selected_buckets), Some(baseline_scan_plan_digest)) = (&scope.selected_buckets, scope.baseline_scan_plan_digest)
-    else {
-        return None;
-    };
-    if selected_buckets.is_empty()
-        || !old_cache.info.snapshot_complete
-        || old_cache.info.last_update.is_none()
-        || old_cache.info.name != DATA_USAGE_ROOT
-        || old_cache.info.next_cycle > generation.want_cycle
-        || old_cache.info.leader_epoch != generation.leader_epoch
-        || old_cache.info.tier_registry_generation != Some(generation.tier_registry_generation)
-        || old_cache.info.source != Some(generation.source)
-        || old_cache.info.scan_plan_digest != Some(baseline_scan_plan_digest)
-        || old_cache.info.cache_key_format != DATA_USAGE_CACHE_KEY_FORMAT
-        || old_cache.checked_flatten_complete_scope(DATA_USAGE_ROOT).is_none()
-    {
-        return None;
-    }
-
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: DATA_USAGE_ROOT.to_string(),
-            next_cycle: generation.want_cycle,
-            leader_epoch: generation.leader_epoch,
-            tier_registry_generation: Some(generation.tier_registry_generation),
-            source: Some(generation.source),
-            snapshot_complete: false,
-            scan_plan_digest: Some(generation.scan_plan_digest),
-            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-            lkg_snapshot_complete: true,
-            lkg_next_cycle: Some(old_cache.info.next_cycle),
-            lkg_last_update: old_cache.info.last_update,
-            lkg_leader_epoch: Some(old_cache.info.leader_epoch),
-            lkg_scan_plan_digest: old_cache.info.scan_plan_digest,
-            ..Default::default()
-        },
-        cache: HashMap::new(),
-    };
-    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-    let root_hash = crate::hash_path(DATA_USAGE_ROOT);
-    let mut current_bucket_names = HashSet::with_capacity(all_buckets.len());
-    for bucket in all_buckets {
-        if !current_bucket_names.insert(bucket.name.as_str()) {
-            return None;
-        }
-        if selected_buckets.contains(&bucket.name) {
-            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
-            continue;
-        }
-
-        let bucket_hash = crate::hash_path(&bucket.name);
-        old_cache.find(&bucket.name)?;
-        cache.copy_with_children(old_cache, &bucket_hash, &Some(root_hash.clone()));
-        cache.find(&bucket.name)?;
-    }
-
-    Some(PreparedScopedSetScan {
-        buckets: set_buckets
-            .iter()
-            .filter(|bucket| selected_buckets.contains(&bucket.name))
-            .cloned()
-            .collect(),
-        cache,
-    })
-}
-
 #[async_trait::async_trait]
 impl ScannerIOCache for SetDisks {
    #[tracing::instrument(skip(self, budget, scan_plan, updates))]
@@ -114,9 +27,8 @@ impl ScannerIOCache for SetDisks {
        scan_mode: HealScanMode,
    ) -> Result<()> {
        let ScannerBucketScanPlan {
-            mut buckets,
+            buckets,
            all_buckets,
-            scope,
            digest: scan_plan_digest,
            leader_epoch,
            tier_registry_generation,
@@ -151,57 +63,26 @@ impl ScannerIOCache for SetDisks {
                "Scanner old data usage cache load failed; rebuilding from bucket caches"
            );
        }
-        let scoped_scan = prepare_scoped_set_scan(
-            &old_cache,
-            &buckets,
-            &all_buckets,
-            &scope,
-            ScannerSetCacheGeneration {
-                want_cycle,
-                leader_epoch,
-                tier_registry_generation,
-                source,
-                scan_plan_digest,
-            },
-        );
-        let mut scoped_cache = scoped_scan.map(|prepared| {
-            buckets = prepared.buckets;
-            prepared.cache
-        });
        if buckets.is_empty() {
            let now = SystemTime::now();
-            let mut cache = match scoped_cache.take() {
-                Some(cache) => cache,
-                None => {
-                    let mut cache = DataUsageCache {
-                        info: DataUsageCacheInfo {
-                            name: DATA_USAGE_ROOT.to_string(),
-                            next_cycle: want_cycle,
-                            leader_epoch,
-                            tier_registry_generation: Some(tier_registry_generation),
-                            source: Some(source),
-                            scan_plan_digest: Some(scan_plan_digest),
-                            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                            ..Default::default()
-                        },
-                        cache: HashMap::new(),
-                    };
-                    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-                    for bucket in all_buckets.iter() {
-                        cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
-                    }
-                    cache
-                }
+            let mut cache = DataUsageCache {
+                info: DataUsageCacheInfo {
+                    name: DATA_USAGE_ROOT.to_string(),
+                    next_cycle: want_cycle,
+                    last_update: Some(now),
+                    leader_epoch,
+                    tier_registry_generation: Some(tier_registry_generation),
+                    source: Some(source),
+                    snapshot_complete: true,
+                    scan_plan_digest: Some(scan_plan_digest),
+                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                    ..Default::default()
+                },
+                cache: HashMap::new(),
            };
-            cache.info.last_update = Some(now);
-            cache.info.snapshot_complete = true;
-            cache.info.lkg_snapshot_complete = false;
-            cache.info.lkg_next_cycle = None;
-            cache.info.lkg_last_update = None;
-            cache.info.lkg_leader_epoch = None;
-            cache.info.lkg_scan_plan_digest = None;
-            if cache.find(DATA_USAGE_ROOT).is_none() {
-                cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+            for bucket in all_buckets.iter() {
+                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
            }
            reset_disk_bucket_scan_gauges(&pool_label, &set_label);
            return persist_and_publish_cache_snapshot(
@@ -388,102 +269,92 @@ impl ScannerIOCache for SetDisks {
        record_disk_bucket_scans_active(0, &pool_label, &set_label);
        let _reset_disk_bucket_scan_gauges = DiskBucketScanGaugeReset::new(pool_label.clone(), set_label.clone());

-        let mut cache = if let Some(cache) = scoped_cache.take() {
-            cache
-        } else {
-            // Fence a stale set aggregate before copying entries into per-bucket work caches.
-            if old_cache.info.next_cycle <= want_cycle
-                && old_cache.info.leader_epoch <= leader_epoch
-                && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
-            {
-                old_cache.info.scan_plan_digest = None;
+        // Fence a stale set aggregate before copying entries into per-bucket work caches.
+        if old_cache.info.next_cycle <= want_cycle
+            && old_cache.info.leader_epoch <= leader_epoch
+            && old_cache.info.tier_registry_generation != Some(tier_registry_generation)
+        {
+            old_cache.info.scan_plan_digest = None;
+        }
+        let old_lkg = old_cache.info.snapshot_complete.then_some({
+            (
+                old_cache.info.next_cycle,
+                old_cache.info.last_update,
+                old_cache.info.leader_epoch,
+                old_cache.info.scan_plan_digest,
+            )
+        });
+        let prepare_outcome = match old_cache.prepare_for_scan(
+            DATA_USAGE_ROOT,
+            want_cycle,
+            leader_epoch,
+            source,
+            scan_plan_digest,
+            require_cache_source,
+        ) {
+            DataUsageCachePrepareOutcome::RejectedNewerCycle => {
+                cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
+                warn!(
+                    target: "rustfs::scanner::io",
+                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_IO,
+                    pool = self.pool_index,
+                    set = self.set_index,
+                    cache_name = DATA_USAGE_CACHE_NAME,
+                    requested_cycle = want_cycle,
+                    cached_cycle = old_cache.info.next_cycle,
+                    state = "stale_cycle_rejected",
+                    "Scanner rejected a set cache cycle regression"
+                );
+                return Ok(());
            }
-            let old_lkg = old_cache.info.snapshot_complete.then_some({
-                (
-                    old_cache.info.next_cycle,
-                    old_cache.info.last_update,
-                    old_cache.info.leader_epoch,
-                    old_cache.info.scan_plan_digest,
-                )
-            });
-            let prepare_outcome = match old_cache.prepare_for_scan(
-                DATA_USAGE_ROOT,
-                want_cycle,
-                leader_epoch,
-                source,
-                scan_plan_digest,
-                require_cache_source,
-            ) {
-                DataUsageCachePrepareOutcome::RejectedNewerCycle => {
-                    cache_cycle_floor.fetch_max(old_cache.info.next_cycle, Ordering::AcqRel);
-                    warn!(
-                        target: "rustfs::scanner::io",
-                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                        component = LOG_COMPONENT_SCANNER,
-                        subsystem = LOG_SUBSYSTEM_IO,
-                        pool = self.pool_index,
-                        set = self.set_index,
-                        cache_name = DATA_USAGE_CACHE_NAME,
-                        requested_cycle = want_cycle,
-                        cached_cycle = old_cache.info.next_cycle,
-                        state = "stale_cycle_rejected",
-                        "Scanner rejected a set cache cycle regression"
-                    );
-                    return Ok(());
-                }
-                DataUsageCachePrepareOutcome::RejectedNewerLeader => {
-                    warn!(
-                        target: "rustfs::scanner::io",
-                        event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                        component = LOG_COMPONENT_SCANNER,
-                        subsystem = LOG_SUBSYSTEM_IO,
-                        pool = self.pool_index,
-                        set = self.set_index,
-                        cache_name = DATA_USAGE_CACHE_NAME,
-                        requested_epoch = leader_epoch,
-                        cached_epoch = old_cache.info.leader_epoch,
-                        state = "stale_leader_rejected",
-                        "Scanner rejected work from an older leader epoch"
-                    );
-                    return Ok(());
-                }
-                outcome => outcome,
-            };
-            if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
-                && let Some((cycle, last_update, epoch, digest)) = old_lkg
-            {
-                old_cache.info.lkg_snapshot_complete = true;
-                old_cache.info.lkg_next_cycle = Some(cycle);
-                old_cache.info.lkg_last_update = last_update;
-                old_cache.info.lkg_leader_epoch = Some(epoch);
-                old_cache.info.lkg_scan_plan_digest = digest;
+            DataUsageCachePrepareOutcome::RejectedNewerLeader => {
+                warn!(
+                    target: "rustfs::scanner::io",
+                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                    component = LOG_COMPONENT_SCANNER,
+                    subsystem = LOG_SUBSYSTEM_IO,
+                    pool = self.pool_index,
+                    set = self.set_index,
+                    cache_name = DATA_USAGE_CACHE_NAME,
+                    requested_epoch = leader_epoch,
+                    cached_epoch = old_cache.info.leader_epoch,
+                    state = "stale_leader_rejected",
+                    "Scanner rejected work from an older leader epoch"
+                );
+                return Ok(());
            }
-
-            let mut cache = DataUsageCache {
-                info: DataUsageCacheInfo {
-                    name: DATA_USAGE_ROOT.to_string(),
-                    next_cycle: want_cycle,
-                    leader_epoch,
-                    tier_registry_generation: Some(tier_registry_generation),
-                    source: Some(source),
-                    snapshot_complete: false,
-                    scan_plan_digest: Some(scan_plan_digest),
-                    cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-                    lkg_snapshot_complete: old_cache.info.lkg_snapshot_complete,
-                    lkg_next_cycle: old_cache.info.lkg_next_cycle,
-                    lkg_last_update: old_cache.info.lkg_last_update,
-                    lkg_leader_epoch: old_cache.info.lkg_leader_epoch,
-                    lkg_scan_plan_digest: old_cache.info.lkg_scan_plan_digest,
-                    ..Default::default()
-                },
-                cache: HashMap::new(),
-            };
-            cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-            for bucket in all_buckets.iter() {
-                cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
-            }
-            cache
+            outcome => outcome,
        };
+        if matches!(prepare_outcome, DataUsageCachePrepareOutcome::Reused)
+            && let Some((cycle, last_update, epoch, digest)) = old_lkg
+        {
+            old_cache.info.lkg_snapshot_complete = true;
+            old_cache.info.lkg_next_cycle = Some(cycle);
+            old_cache.info.lkg_last_update = last_update;
+            old_cache.info.lkg_leader_epoch = Some(epoch);
+            old_cache.info.lkg_scan_plan_digest = digest;
+        }
+
+        let mut cache = DataUsageCache {
+            info: DataUsageCacheInfo {
+                name: DATA_USAGE_ROOT.to_string(),
+                next_cycle: want_cycle,
+                leader_epoch,
+                tier_registry_generation: Some(tier_registry_generation),
+                source: Some(source),
+                snapshot_complete: false,
+                scan_plan_digest: Some(scan_plan_digest),
+                cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
+                ..Default::default()
+            },
+            cache: HashMap::new(),
+        };
+        cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
+        for bucket in all_buckets.iter() {
+            cache.replace(&bucket.name, DATA_USAGE_ROOT, DataUsageEntry::default());
+        }

        let (bucket_tx, bucket_rx) = mpsc::channel::<BucketInfo>(buckets.len());

@@ -1386,6 +1257,11 @@ impl ScannerIOCache for SetDisks {
            incomplete_scope.info.snapshot_complete = false;
            incomplete_scope.info.scan_plan_digest = Some(scan_plan_digest);
            incomplete_scope.info.cache_key_format = DATA_USAGE_CACHE_KEY_FORMAT;
+            incomplete_scope.info.lkg_snapshot_complete = old_cache.info.lkg_snapshot_complete;
+            incomplete_scope.info.lkg_next_cycle = old_cache.info.lkg_next_cycle;
+            incomplete_scope.info.lkg_last_update = old_cache.info.lkg_last_update;
+            incomplete_scope.info.lkg_leader_epoch = old_cache.info.lkg_leader_epoch;
+            incomplete_scope.info.lkg_scan_plan_digest = old_cache.info.lkg_scan_plan_digest;
            if let Err(e) = updates.send(incomplete_scope).await {
                error!(
                    target: "rustfs::scanner::io",
@@ -63,41 +63,6 @@ pub(crate) async fn nsscanner_with_storage_status<S>(
 where
    S: ScannerStorage,
 {
-    let request = ScannerCycleRequest {
-        ctx,
-        budget,
-        updates,
-        want_cycle,
-        leader_epoch,
-        scan_mode,
-        scan_scope: ScannerBucketScanScope::default(),
-    };
-    nsscanner_with_storage_status_scoped(store, request).await
-}
-
-pub(crate) struct ScannerCycleRequest {
-    pub(crate) ctx: CancellationToken,
-    pub(crate) budget: Arc<ScannerCycleBudget>,
-    pub(crate) updates: mpsc::Sender<DataUsageInfo>,
-    pub(crate) want_cycle: u64,
-    pub(crate) leader_epoch: u64,
-    pub(crate) scan_mode: HealScanMode,
-    pub(crate) scan_scope: ScannerBucketScanScope,
-}
-
-pub(crate) async fn nsscanner_with_storage_status_scoped<S>(store: &S, request: ScannerCycleRequest) -> Result<ScannerCycleResult>
-where
-    S: ScannerStorage,
-{
-    let ScannerCycleRequest {
-        ctx,
-        budget,
-        updates,
-        want_cycle,
-        leader_epoch,
-        scan_mode,
-        scan_scope,
-    } = request;
    let child_token = ctx.child_token();
    let _tier_cycle_guard = begin_tier_registry_cycle(want_cycle, leader_epoch);

@@ -315,7 +280,6 @@ where
        let scan_plan = ScannerBucketScanPlan {
            buckets: set_buckets,
            all_buckets: Arc::clone(&all_buckets),
-            scope: scan_scope.clone(),
            digest: scan_plan_digest,
            leader_epoch,
            tier_registry_generation,
@@ -765,155 +765,6 @@ fn bucket_usage_scan_order_prioritizes_dirty_buckets() {
    assert_eq!(names, vec!["dirty", "missing", "cached"]);
 }

-fn complete_set_usage_cache(buckets: &[(&str, usize)], scan_plan_digest: DataUsageScanPlanDigest) -> DataUsageCache {
-    let mut cache = DataUsageCache {
-        info: DataUsageCacheInfo {
-            name: DATA_USAGE_ROOT.to_string(),
-            next_cycle: 7,
-            last_update: Some(SystemTime::now()),
-            leader_epoch: 11,
-            source: Some(DataUsageCacheSource::new(1, 2)),
-            snapshot_complete: true,
-            scan_plan_digest: Some(scan_plan_digest),
-            cache_key_format: DATA_USAGE_CACHE_KEY_FORMAT,
-            tier_registry_generation: Some(13),
-            ..Default::default()
-        },
-        ..Default::default()
-    };
-    cache.replace(DATA_USAGE_ROOT, "", DataUsageEntry::default());
-    for (bucket, size) in buckets {
-        cache.replace(
-            bucket,
-            DATA_USAGE_ROOT,
-            DataUsageEntry {
-                size: *size,
-                objects: 1,
-                ..Default::default()
-            },
-        );
-    }
-    cache
-}
-
-#[test]
-fn scoped_set_scan_preserves_unselected_usage_and_drops_deleted_buckets() {
-    let baseline_digest = DataUsageScanPlanDigest([1; 32]);
-    let current_digest = DataUsageScanPlanDigest([2; 32]);
-    let mut old_cache = complete_set_usage_cache(&[("stable", 10), ("dirty", 20), ("deleted", 30)], baseline_digest);
-    old_cache.replace(
-        "stable/prefix",
-        "stable",
-        DataUsageEntry {
-            size: 5,
-            objects: 1,
-            ..Default::default()
-        },
-    );
-    let all_buckets = vec![bucket_info("stable"), bucket_info("dirty")];
-    let selected_buckets = Arc::new(HashSet::from(["dirty".to_string(), "deleted".to_string()]));
-
-    let prepared = prepare_scoped_set_scan(
-        &old_cache,
-        &all_buckets,
-        &all_buckets,
-        &ScannerBucketScanScope {
-            selected_buckets: Some(selected_buckets),
-            baseline_scan_plan_digest: Some(baseline_digest),
-        },
-        ScannerSetCacheGeneration {
-            want_cycle: 8,
-            leader_epoch: 11,
-            tier_registry_generation: 13,
-            source: DataUsageCacheSource::new(1, 2),
-            scan_plan_digest: current_digest,
-        },
-    )
-    .expect("complete matching set cache should support a scoped scan");
-
-    assert_eq!(prepared.buckets.iter().map(|bucket| bucket.name.as_str()).collect::<Vec<_>>(), ["dirty"]);
-    let stable = prepared
-        .cache
-        .checked_flatten("stable")
-        .expect("unselected bucket subtree should be retained");
-    assert_eq!((stable.size, stable.objects), (15, 2));
-    assert_eq!(prepared.cache.find("dirty").map(|entry| (entry.size, entry.objects)), Some((0, 0)));
-    assert!(prepared.cache.find("deleted").is_none());
-    assert_eq!(prepared.cache.info.scan_plan_digest, Some(current_digest));
-    assert_eq!(prepared.cache.info.next_cycle, 8);
-    assert!(!prepared.cache.info.snapshot_complete);
-    assert!(prepared.cache.info.lkg_snapshot_complete);
-    assert_eq!(prepared.cache.info.lkg_next_cycle, Some(7));
-    assert_eq!(prepared.cache.info.lkg_scan_plan_digest, Some(baseline_digest));
-}
-
-#[test]
-fn scoped_set_scan_falls_back_when_an_unselected_bucket_has_no_baseline() {
-    let baseline_digest = DataUsageScanPlanDigest([3; 32]);
-    let old_cache = complete_set_usage_cache(&[("stable", 10)], baseline_digest);
-    let all_buckets = vec![bucket_info("stable"), bucket_info("new")];
-
-    assert!(
-        prepare_scoped_set_scan(
-            &old_cache,
-            &all_buckets,
-            &all_buckets,
-            &ScannerBucketScanScope {
-                selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
-                baseline_scan_plan_digest: Some(baseline_digest),
-            },
-            ScannerSetCacheGeneration {
-                want_cycle: 8,
-                leader_epoch: 11,
-                tier_registry_generation: 13,
-                source: DataUsageCacheSource::new(1, 2),
-                scan_plan_digest: DataUsageScanPlanDigest([4; 32]),
-            },
-        )
-        .is_none()
-    );
-}
-
-#[test]
-fn scoped_set_scan_requires_an_exact_complete_baseline() {
-    let baseline_digest = DataUsageScanPlanDigest([5; 32]);
-    let all_buckets = vec![bucket_info("dirty")];
-    let scope = ScannerBucketScanScope {
-        selected_buckets: Some(Arc::new(HashSet::from(["dirty".to_string()]))),
-        baseline_scan_plan_digest: Some(baseline_digest),
-    };
-    let generation = ScannerSetCacheGeneration {
-        want_cycle: 8,
-        leader_epoch: 11,
-        tier_registry_generation: 13,
-        source: DataUsageCacheSource::new(1, 2),
-        scan_plan_digest: DataUsageScanPlanDigest([6; 32]),
-    };
-
-    let mut incomplete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    incomplete.info.snapshot_complete = false;
-    assert!(prepare_scoped_set_scan(&incomplete, &all_buckets, &all_buckets, &scope, generation).is_none());
-
-    let mut not_durable = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    not_durable.info.last_update = None;
-    assert!(prepare_scoped_set_scan(&not_durable, &all_buckets, &all_buckets, &scope, generation).is_none());
-
-    let mut wrong_digest = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    wrong_digest.info.scan_plan_digest = Some(DataUsageScanPlanDigest([7; 32]));
-    assert!(prepare_scoped_set_scan(&wrong_digest, &all_buckets, &all_buckets, &scope, generation).is_none());
-
-    let empty_scope = ScannerBucketScanScope {
-        selected_buckets: Some(Arc::new(HashSet::new())),
-        baseline_scan_plan_digest: Some(baseline_digest),
-    };
-    let complete = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    assert!(prepare_scoped_set_scan(&complete, &all_buckets, &all_buckets, &empty_scope, generation).is_none());
-
-    let mut future_cache = complete_set_usage_cache(&[("dirty", 10)], baseline_digest);
-    future_cache.info.next_cycle = generation.want_cycle.saturating_add(1);
-    assert!(prepare_scoped_set_scan(&future_cache, &all_buckets, &all_buckets, &scope, generation).is_none());
-}
-
 #[test]
 fn record_set_scan_failure_preserves_first_error() {
    let mut first = None;
@@ -327,89 +327,9 @@ Recursive prefix/delete-all is the exception because physical directory removal

 - Every code path that removes or overwrites transitioned metadata either atomically leaves an exact free-version owner or enters an already-authorized v6 dispatch.
 - If both a journal and free-version are observable, recovery preserves the remote object until version-specific authority and all-pool absence prove a single owner.
- Missing `transitioned-version-state` remains unknown. Compatibility work must not synthesize known-disabled or exact semantics merely from an empty stored version ID.
+- Missing `transition-version-state` remains unknown. Compatibility work must not synthesize known-disabled or exact semantics merely from an empty stored version ID.

-The following single-record protocol approves how historical objects without RustFS `transitioned-version-state` can be upgraded. It does not approve a bulk scanner or allow destructive cleanup to consume an unproven record.
-
-## Legacy transitioned-version-state reconciliation
-
-### Current
-
-An absent `transitioned-version-state` key decodes as `TransitionVersionState::Unknown`. The current GET and free-version cleanup paths reject that state rather than interpreting an empty remote version as unversioned. There is no admin route that repairs this field in `xl.meta`. The existing transition-transaction reconcile route operates on expired `UploadOutcomeUnknown` transaction records and can exact-delete their canonical candidates; it is a separate protocol and must not be reused for metadata reconciliation.
-
-An explicitly persisted `unknown`, a malformed state, conflicting RustFS/MinIO compatibility keys, an invalid or nil version identifier, and a partial transition tuple are not legacy absence. They remain invalid or ambiguous and fail closed.
-
-### Approved single-record control surface
-
-The approved target is one synchronous, exact logical-version operation. It does not list a bucket or prefix and does not create a durable job:
-
-```text
-GET  /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<object>&versionId=<local-version-id>
-POST /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<object>&versionId=<local-version-id>
-```
-
-`versionId` is required; the literal `null` is the explicit selector for a locally unversioned object, while an omitted or empty selector is invalid. GET requires `admin:ListTier`. It performs an authoritative all-pool metadata read and a bounded server-side live backend probe, but it never writes metadata or mutates the remote tier. Its response includes the canonical immutable source tuple, every original per-set missing-state representation, the proposed target tuple when one is provable, an opaque reconciliation digest over all three, fleet/topology and tier-generation readiness, and whether POST is ready to attempt migration. GET never labels an unmodified legacy record `migrated`.
-
-POST requires `admin:SetTier`. Its body contains `confirm: true`, the complete source and target tuples, every original per-set representation, and their reconciliation digest returned by a recent GET. The server does not trust a client-supplied state or external assertion: it rereads the object, requires every immutable source field to match exactly after canonicalization, repeats the bounded live probe, and derives the target state itself. For the mutable state/version/destination fields, each set must equal either its original missing-state representation bound by the digest or the exact newly proven target; this is the only accepted partial-retry shape. A missing confirmation, any other stale tuple or digest, or a widened selector is rejected before any write.
-
-The expected tuple binds all evidence whose change could redirect the repair:
-
- bucket name and incarnation; exact object name, local version ID, data directory, modification time, size, and ETag;
- transition completion status, tier name, canonical remote object name, raw remote-version key presence/value, and raw state-key presence/value under both compatibility prefixes;
- tier-config generation and the `transition-tier-destination-id` binding, including backend type, endpoint/bucket/prefix identity, and the credential-independent backend fingerprint;
- topology generation and the generation/digest of every authoritative `xl.meta` copy found across pools and sets.
-
-The only allowed state derivations are:
-
-| Live proof | Persisted state | Persisted remote version | Remote request meaning |
-|---|---|---|---|
-| Provider proves versioning disabled and the candidate is present | `KnownDisabled` | Absent/empty | Send no `versionId` |
-| Provider proves suspended-version null semantics and the exact candidate is present | `SuspendedNull` | Literal `null` | Send the provider's null-version form |
-| Provider proves one exact, nonempty, non-`null` opaque version | `Exact` | That exact opaque identifier | Send that exact `versionId` |
-
-Missing, multiple, changing, or unsupported probe results do not select a state. In particular, a preexisting empty remote-version field is not evidence for `KnownDisabled`, and a client may not nominate `Exact` or `SuspendedNull`.
-
-The POST may write only the derived `transitioned-version-state`, its corresponding `transitioned-versionID` value when the proven model requires one, and the exact `transition-tier-destination-id` binding under both compatibility prefixes in the matching `xl.meta` version. It does not issue remote GET beyond the proof probe, remote PUT, remote DELETE, local object DELETE, free-version cleanup, transaction/journal cleanup, tier-config mutation, restore, or source-payload rewrite. Reconciliation establishes metadata meaning; a later ordinary owner may perform cleanup under its own destructive protocol.
-
-### Outcome contract
-
-POST returns exactly one of the following outcomes and whether it changed bytes. GET uses the same diagnostic names for non-applicable cases, returns `ready-to-migrate` when a missing state is provable, and returns `migrated` only when strong readback shows the record was already explicit and converged:
-
-| Outcome | Meaning and permitted effect |
-|---|---|
-| `migrated` | All authoritative copies already contain, or were monotonically advanced to, the same proven state and destination identity. Only this outcome makes the record eligible for later ordinary read/delete semantics. |
-| `retained-ambiguous` | The tuple is structurally legacy-compatible, but the live probe is missing, multiple, changing, unsupported, or otherwise cannot prove exactly one state. No metadata or remote object is changed. |
-| `corrupt` | Explicit `Unknown`, malformed/contradictory compatibility keys, nil/invalid identifiers, partial transition metadata, or authoritative copies outside the one allowed `{original missing representation, exact proven target}` retry subset were observed. No backend probe is required after corruption is established, and nothing is changed. |
-| `backend-unavailable` | The bound tier generation/destination cannot be acquired, the bounded probe fails, or a metadata quorum/strong readback needed to complete the operation is unavailable. Any already-persisted monotonic subset is retained for an idempotent retry; it is never rolled back. |
-
-HTTP failure detail may distinguish a stale expected tuple, lost fence, timeout, or unavailable quorum, but it must preserve one of these machine-readable outcomes. Logs and audit events include request identity, object identity, tier, generations, outcome, and whether bytes changed; they never include credentials or raw credential-derived configuration.
-
-### Fence, write, and retry order
-
-The approved POST executes the following order. A step that cannot be proven stops the operation without remote mutation:
-
-1. Authenticate `admin:SetTier`, validate the exact single-record selector, `confirm: true`, expected tuple, and digest.
-2. Prove every required node advertises the reconciliation format and destination-identity capability; capture the fleet and topology generation. Unknown or unsupported nodes block the writer.
-3. Acquire the bucket-lifecycle WRITE fence and validate the bucket incarnation.
-4. Acquire the exact tier-config generation lease bound to the expected destination identity.
-5. Acquire exact object-version WRITE locks for every owning physical pool/set in stable pool/set order.
-6. Perform an authoritative all-pool read, reject duplicate/conflicting ownership, validate every compatibility key, and require each set to match the immutable source tuple plus either its digest-bound original missing-state representation or the exact proposed target.
-7. Run one bounded, cancellation-aware live probe through the leased backend. No client or cached probe result is authority.
-8. Before writing, revalidate the fleet/topology generation, bucket incarnation and lifecycle fence, tier lease/destination identity, physical owner set, and complete metadata tuple.
-9. Write the same derived state and destination identity to each authoritative set with that set's metadata quorum and conditional generation. A timeout or response loss is resolved only by a strong read of that exact set.
-10. Strongly reread every authoritative set and revalidate the full tuple, state, destination identity, and topology before returning `migrated`. Release locks and leases in reverse order.
-
-Cross-pool and cross-set partial success is monotonic. The only legal repair edge is `missing state -> one proven {state, remote version, destination identity}`. A retry may accept an already-written subset only when every known copy equals the newly proven target, every remaining copy equals its original missing-state representation captured by the reconciliation digest, and all immutable source fields still match; it then fills only the missing copies. This exact target-plus-original subset is neither stale nor corrupt. The retry never clears a known state, rewrites it to another state, changes destination identity, or rolls a successful set back to missing/`Unknown`. Any other divergent value produces `corrupt`; an unavailable set/readback produces `backend-unavailable`, and destructive cleanup remains blocked until a later strong all-pool read proves complete convergence.
-
-GET takes the same fleet/topology snapshot and authoritative all-pool read but no write locks that imply mutation authority. Because GET is advisory, POST always repeats every fence, read, and live proof rather than promoting the GET result.
-
-### Mixed-version and future batch work
-
-The writer gate requires every node that can serve, rewrite, heal, decommission, or recover the affected `xl.meta` to preserve the explicit state and destination identity. A rolling fleet with an unknown/unsupported node is inspect-only. Downgrade is blocked while reconciled records could be rewritten by readers that erase or misinterpret those fields. Cross-pool movement must either copy the proven tuple unchanged or block reconciliation; a first-match lookup is never sufficient.
-
-Explicit `Unknown`, corruption, and ambiguity remain fail closed for reads that cannot prove non-destructive semantics and for every destructive path. A migrated record becomes ordinary explicit metadata, but reconciliation itself never transfers remote DELETE ownership.
-
-A bucket/prefix/fleet batch reconcile is still an **open design**. It requires a separate durable job identity, create-only admission, lease/CAS checkpoint, bounded pages, per-record expected tuples and outcomes, cancellation/restart semantics, retention, fleet rollout negotiation, and status counters. Implementations must not approximate that protocol by adding a list selector or background loop to the synchronous route.
+How historical objects without RustFS transition-version-state can be upgraded safely is an **open design**. It requires a read/repair rule with rollback behavior before destructive cleanup may consume those records.

 ## Durable namespace receipts during decommission

@@ -39,11 +39,11 @@ Internal metadata is stored under both `x-rustfs-internal-<suffix>` and `x-minio
 |--------|---------|
 | `transition-status` | `"complete"` when tiered |
 | `transitioned-object` | tier key path (stored without the tier prefix; `get_dest` adds it) |
-| `transitioned-versionID` | Provider version identifier: current exact UTF-8 text, legacy RustFS raw UUID bytes, MinIO's empty unversioned value, or absent for some historical unversioned records. Interpret it only with `transitioned-version-state` or a live compatibility probe. |
+| `transitioned-versionID` | S3 version_id returned by tier PUT (16 raw UUID bytes, or absent) |
 | `transition-tier` | tier name |
 | `tier-free-versionID` | delete-marker version for free-version sweep |

-Legacy raw UUID values must reject empty, malformed, and nil UUIDs (regression covered in `crates/filemeta/src/filemeta/version.rs` tests):
+Reading binary values must reject empty, malformed, and nil values (regression covered in `crates/filemeta/src/filemeta/version.rs` tests):

 ```rust
 get_bytes(&self.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID)
@@ -52,7 +52,7 @@ get_bytes(&self.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID)
 // None for: absent key, wrong-length bytes, nil UUID
 ```

-`transition_version_id == None` means only that no usable legacy UUID projection exists; it does not prove the remote bucket's versioning model. Only an explicit `KnownDisabled` state authorizes ordinary GET/DELETE to omit `versionId`. A missing state with an absent or empty version key remains `Unknown` and requires the bounded compatibility probe or the approved reconcile workflow; it never directly authorizes cleanup. A nil UUID (`00000000-...`) sent as `?versionId=` causes `NoSuchVersion`. Do not use `Uuid::from_slice(..).unwrap_or_default()` here: it converts an empty metadata value into `Uuid::nil()`, which is exactly that failure.
+`transition_version_id == None` means the tier bucket is unversioned; the GET/DELETE against the tier must then send no `versionId` parameter. A nil UUID (`00000000-...`) sent as `?versionId=` causes `NoSuchVersion`. Do not use `Uuid::from_slice(..).unwrap_or_default()` here: it converts an empty metadata value into `Uuid::nil()`, which is exactly that failure.

 ## Inspect xl.meta directly

@@ -64,8 +64,8 @@ cargo build -p rustfs-filemeta --example dump_fileinfo

 | Output | Meaning |
 |---|---|
-| `transition_ver_id: <none>` | No usable legacy UUID projection exists. Inspect `transitioned-version-state` and the raw compatibility keys; do not infer unversioned semantics. |
-| `transition_ver_id: <uuid>` | A legacy UUID representation decoded successfully. It is not destructive authority unless the persisted state or reconcile proof establishes the exact remote model. |
+| `transition_ver_id: <none>` | No versionId will be sent to the tier (correct for a non-versioned tier bucket). |
+| `transition_ver_id: <uuid>` | That UUID will be sent as `?versionId=<uuid>`. |

 There is one `xl.meta` per erasure shard disk (`{disk}/{bucket}/{object}/xl.meta`); all shards of a healthy object should be identical. `dump_versions` (same crate) lists every version in a file.

@@ -80,7 +80,7 @@ RUST_LOG=rustfs_ecstore::bucket::lifecycle=debug rustfs ...
 | `fetching transitioned object from tier` | DEBUG | Emitted before the tier request. |
 | `tier GET failed` | ERROR | Includes `tier_version_id`. |

-If the version keys are empty while `transitioned-version-state` is absent, the record has the historical MinIO unversioned shape but still remains `Unknown`; only the compatibility probe or reconcile protocol may prove `KnownDisabled`. If state is explicitly `KnownDisabled`, no `versionId` is sent.
+If both `x-rustfs-internal-transitioned-versionID` and `x-minio-internal-transitioned-versionID` are the empty string, the object was transitioned to a non-versioned tier bucket and no versionId must be sent.

 ## Manual transition run

@@ -154,35 +154,6 @@ Historical transition transactions in `upload_outcome_unknown` state can use an

 `finalize_missing` re-runs the provider probe and fails closed for `unversioned_present`, `versioned_present`, `ambiguous`, `unsupported`, or probe errors. It never accepts an operator assertion in place of a live `missing` result. Providers without an authoritative probe or exact version deletion remain pending; the endpoint does not infer provider capabilities, accept external absence assertions, or select a candidate automatically.

-## Reconcile legacy transition-version metadata
-
-This section describes an **approved target that is not implemented yet**. The current server has no admin route that backfills a missing `transitioned-version-state` in `xl.meta`. Do not use the transaction reconcile route above for this purpose: that route owns an upload transaction candidate and may delete it, while legacy metadata reconciliation is non-destructive and may update only the exact local metadata version.
-
-The approved interface is synchronous and accepts exactly one bucket/object/local-version tuple:
-
-```text
-GET  /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<object>&versionId=<local-version-id>
-POST /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<object>&versionId=<local-version-id>
-```
-
-`versionId` is required; use the literal `null` for a locally unversioned object. An omitted or empty selector is invalid. GET requires `admin:ListTier`. It reports the authoritative all-pool tuple, destination identity, fleet/topology readiness, live probe classification, opaque expected-tuple digest, and a machine-readable diagnosis. It returns `ready-to-migrate`, not `migrated`, when a missing state is provable because GET is read-only.
-
-POST requires `admin:SetTier`, `confirm: true`, and the complete immutable source tuple, original per-set missing-state representations, proposed target, and reconciliation digest returned by GET. The server rereads every authoritative copy and repeats the bounded live backend probe; provider console output or an operator-supplied state is diagnostic evidence only, never write authority. A retry accepts only copies that still match their digest-bound original representation or already equal the exact proven target; any other divergence is stale or corrupt. The server may persist only one of these exact state/version pairs, together with the bound destination identity:
-
-| Proven remote model | State | Version value |
-|---|---|---|
-| Versioning disabled | `KnownDisabled` | Empty/absent; later requests omit `versionId` |
-| Versioning suspended null object | `SuspendedNull` | Literal `null` |
-| One exact version | `Exact` | Exact nonempty, non-`null` opaque identifier |
-
-The response outcome is `migrated`, `retained-ambiguous`, `corrupt`, or `backend-unavailable`. `migrated` means strong all-pool readback proved the same state and destination identity on every authoritative copy; it can be idempotent with `changed=false`. Ambiguous/missing/multiple probe results are retained, and explicit `Unknown`, malformed or conflicting dual keys, nil identifiers, partial tuples, or copies outside the exact `{original missing representation, proven target}` retry subset fail closed. An unavailable backend, tier generation, metadata quorum, or required strong readback reports `backend-unavailable`; a monotonic partial write is retained for retry and never rolled back.
-
-The POST does not issue remote DELETE or PUT, remove local data, create a free-version, clean a transaction/journal, or change tier configuration. It holds the approved fleet/topology, bucket-lifecycle, exact tier-generation/destination, and stable all-pool object-version fences across authoritative reread and the bounded probe; it rechecks them before quorum writes and after strong readback. A fleet containing a node that cannot preserve the explicit state/destination binding is inspect-only, and a cross-pool first match is never enough.
-
-There is intentionally no bucket, prefix, or fleet selector. Batch repair requires a separate durable, resumable job protocol and remains future work. Until the single-record route is implemented, retain affected metadata, use external inspection only for diagnosis, and never hand-edit `xl.meta` or enable remote cleanup by assuming that an empty version field means an unversioned tier.
-
-The full approved fence, quorum, cross-set retry, destination-binding, and mixed-version contract is specified in [../architecture/ilm-tiering-persistence-contracts.md](../architecture/ilm-tiering-persistence-contracts.md#legacy-transitioned-version-state-reconciliation).
-
 ## Invariant: local-first expiry ordering

 `expire_transitioned_object` deletes local metadata first (making the object unreachable) and leaves a persisted free-version for remote-tier cleanup. `tier_free_version_recovery.rs` scans and re-enqueues that record; the lifecycle worker's `cleanup_free_version_exact` in `bucket_lifecycle_ops.rs` performs the fenced remote delete, local-marker cleanup, and rescan. Never remove a remote tier version while live local metadata still points at it: doing so lets a concurrent GET read a stored version_id whose remote version is already gone and fail with `NoSuchVersion`.
@@ -80,82 +80,6 @@ fn wasabi_payload_name(config: &TierConfig) -> S3Result<String> {
        .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Wasabi configuration"))
 }

-fn normalize_add_tier_payload_name(config: &mut TierConfig) -> S3Result<()> {
-    match config.tier_type {
-        TierType::S3 => {
-            let _ = config
-                .s3
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?;
-        }
-        TierType::Wasabi => config.name = wasabi_payload_name(config)?,
-        TierType::RustFS => {
-            config.name = config
-                .rustfs
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?
-                .name
-                .clone();
-        }
-        TierType::MinIO => {
-            config.name = config
-                .minio
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?
-                .name
-                .clone();
-        }
-        TierType::Aliyun => {
-            config.name = config
-                .aliyun
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?
-                .name
-                .clone();
-        }
-        TierType::Tencent => {
-            config.name = config
-                .tencent
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?
-                .name
-                .clone();
-        }
-        TierType::Huaweicloud => {
-            config.name = config
-                .huaweicloud
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?
-                .name
-                .clone();
-        }
-        TierType::Azure => {
-            config.name = config
-                .azure
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?
-                .name
-                .clone();
-        }
-        TierType::GCS => {
-            let _ = config
-                .gcs
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?;
-        }
-        TierType::R2 => {
-            config.name = config
-                .r2
-                .as_ref()
-                .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?
-                .name
-                .clone();
-        }
-        TierType::Unsupported => {}
-    }
-    Ok(())
-}
-
 fn spawn_transition_tier_config_propagation(action: &'static str) {
    if let Some(notification_sys) = current_notification_system() {
        debug!(
@@ -339,7 +263,75 @@ impl Operation for AddTier {
        let mut args: TierConfig = serde_json::from_slice(&body)
            .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRequest, format!("invalid JSON: {e}")))?;

-        normalize_add_tier_payload_name(&mut args)?;
+        match args.tier_type {
+            TierType::S3 => {
+                args.name = args
+                    .s3
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing S3 configuration"))?
+                    .name;
+            }
+            TierType::Wasabi => {
+                args.name = wasabi_payload_name(&args)?;
+            }
+            TierType::RustFS => {
+                args.name = args
+                    .rustfs
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing RustFS configuration"))?
+                    .name;
+            }
+            TierType::MinIO => {
+                args.name = args
+                    .minio
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing MinIO configuration"))?
+                    .name;
+            }
+            TierType::Aliyun => {
+                args.name = args
+                    .aliyun
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Aliyun configuration"))?
+                    .name;
+            }
+            TierType::Tencent => {
+                args.name = args
+                    .tencent
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Tencent configuration"))?
+                    .name;
+            }
+            TierType::Huaweicloud => {
+                args.name = args
+                    .huaweicloud
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Huawei Cloud configuration"))?
+                    .name;
+            }
+            TierType::Azure => {
+                args.name = args
+                    .azure
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing Azure configuration"))?
+                    .name;
+            }
+            TierType::GCS => {
+                args.name = args
+                    .gcs
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing GCS configuration"))?
+                    .name;
+            }
+            TierType::R2 => {
+                args.name = args
+                    .r2
+                    .clone()
+                    .ok_or_else(|| S3Error::with_message(S3ErrorCode::InvalidRequest, "missing R2 configuration"))?
+                    .name;
+            }
+            _ => (),
+        }
        debug!(
            event = EVENT_ADMIN_TIER_STATE,
            component = LOG_COMPONENT_ADMIN,
@@ -1158,44 +1150,6 @@ mod tests {
        assert_eq!(err.message(), Some("missing Wasabi configuration"));
    }

-    #[test]
-    fn add_tier_payload_preserves_canonical_madmin_s3_and_gcs_names() {
-        for (provider, wire) in [
-            (
-                "S3",
-                serde_json::json!({
-                    "Type": "s3",
-                    "Name": "COLD-S3",
-                    "S3": {
-                        "Endpoint": "https://s3.example.invalid",
-                        "AccessKey": "access",
-                        "SecretKey": "secret",
-                        "Bucket": "archive"
-                    }
-                }),
-            ),
-            (
-                "GCS",
-                serde_json::json!({
-                    "Type": "gcs",
-                    "Name": "COLD-GCS",
-                    "GCS": {
-                        "Endpoint": "https://storage.googleapis.com",
-                        "Creds": "e30=",
-                        "Bucket": "archive"
-                    }
-                }),
-            ),
-        ] {
-            let mut config: TierConfig = serde_json::from_value(wire).expect("canonical madmin payload should decode");
-            let expected = config.name.clone();
-
-            normalize_add_tier_payload_name(&mut config).expect("canonical madmin payload should pass the handler boundary");
-
-            assert_eq!(config.name, expected, "{provider} top-level Name must not be cleared");
-        }
-    }
-
    #[test]
    fn resolve_tier_name_prefers_path_parameter() {
        let uri: Uri = "/rustfs/admin/v3/tier/HOT?tier=COLD".parse().expect("uri should parse");
@@ -1802,11 +1756,5 @@ mod tests {
        }

        assert!(!production.contains("check_key_valid(get_session_token"));
-
-        let add_tier = source_block(production, "impl Operation for AddTier");
-        assert!(
-            add_tier.contains("normalize_add_tier_payload_name(&mut args)?;"),
-            "AddTier must preserve canonical top-level provider names through the tested boundary helper"
-        );
    }
 }
@@ -170,12 +170,12 @@ use s3s::dto::{
    DeleteObjectsOutput, DeletedObject, ETag, GetObjectAttributesInput, GetObjectAttributesOutput, GetObjectAttributesParts,
    GetObjectInput, GetObjectOutput, HeadObjectInput, HeadObjectOutput, MetadataDirective, ObjectAttributes, ObjectLockLegalHold,
    ObjectLockLegalHoldStatus, ObjectLockMode, ObjectLockRetention, ObjectLockRetentionMode, ObjectPart, PutObjectInput,
-    PutObjectOutput, Range, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreRequestType, RestoreStatus,
-    SSECustomerAlgorithm, SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput,
-    ServerSideEncryption, ServerSideEncryptionConfiguration, StorageClass, StreamingBlob, TaggingDirective, TaggingHeader,
-    Timestamp, TimestampFormat, WebsiteRedirectLocation,
+    PutObjectOutput, Range, RequestCharged, RestoreObjectInput, RestoreObjectOutput, RestoreStatus, SSECustomerAlgorithm,
+    SSECustomerKeyMD5, SSEKMSKeyId, SelectObjectContentInput, SelectObjectContentOutput, ServerSideEncryption,
+    ServerSideEncryptionConfiguration, StorageClass, StreamingBlob, TaggingDirective, TaggingHeader, Timestamp, TimestampFormat,
+    WebsiteRedirectLocation,
 };
-use s3s::header::X_AMZ_RESTORE;
+use s3s::header::{X_AMZ_RESTORE, X_AMZ_RESTORE_OUTPUT_PATH};
 use s3s::stream::{ByteStream, DynByteStream, RemainingLength};
 use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};

@@ -165,49 +165,18 @@ impl DefaultObjectUsecase {

        validate_table_catalog_object_mutation(&bucket, &object).await?;

-        // Typed S3 errors on every RestoreObject failure (backlog#2205): a
-        // `Custom` code serializes as a generic 500, which makes SDK clients
-        // retry client errors and conflicts alike.
-        let rreq = rreq.ok_or_else(|| S3Error::with_message(S3ErrorCode::MalformedXML, "restore request is required"))?;
-
-        // SELECT-type restore is not supported (backlog#1341). The restore
-        // path can only write the retrieved bytes back to the source key, so
-        // honouring a SELECT request overwrote the source object with
-        // SELECT-only metadata (dropping `x-amz-restore`, user metadata and
-        // tags on an unversioned bucket, or publishing a bogus latest version
-        // on a versioned one) while never writing anything to
-        // `OutputLocation.S3`. Reject before any guard, metadata write or
-        // fabricated `x-amz-restore-output-path` response header.
-        if rreq
-            .type_
-            .as_ref()
-            .is_some_and(|type_| type_.as_str() == RestoreRequestType::SELECT)
-        {
-            return Err(S3Error::with_message(
-                S3ErrorCode::NotImplemented,
-                "SELECT restore requests are not supported.",
-            ));
-        }
+        let rreq = rreq.ok_or_else(|| {
+            S3Error::with_message(S3ErrorCode::Custom("ErrValidRestoreObject".into()), "restore request is required")
+        })?;

        let Some(store) = self.object_store() else {
            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
        };

-        // Validate the request shape before taking any lock or reading the
-        // object: a malformed request or an illegal `Days` value is a client
-        // error, and the validator messages are static — they carry no
-        // backend or credential detail.
-        if let Err(e) = validate_restore_request(&rreq, store.clone()) {
-            return Err(S3Error::with_message(
-                S3ErrorCode::InvalidRequest,
-                format!("Restore object validation failed: {e}"),
-            ));
-        }
-
        let version_id_str = version_id.clone().unwrap_or_default();
        let mut opts = post_restore_opts(&version_id_str, &bucket, &object)
            .await
-            .map_err(ApiError::from)?;
+            .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrPostRestoreOpts".into()), "restore object failed."))?;
        apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
        // `apply_bucket_generation_guard` deliberately tolerates a missing guard
        // (only the S3 access layer installs one), so this must not hard-require
@@ -223,7 +192,11 @@ impl DefaultObjectUsecase {
            }
        };

-        let restore_operation_id = Some(Uuid::new_v4());
+        // SELECT-type restores skip both the ongoing check and the metadata
+        // write below, so the accept guard would protect nothing for them —
+        // they keep the plain (read-locked) accept path.
+        let is_select = rreq.type_.as_ref().is_some_and(|t| t.as_str() == "SELECT");
+        let restore_operation_id = (!is_select).then(Uuid::new_v4);
        let mut restore_worker_guard = if let Some(operation_id) = restore_operation_id {
            Some(
                store
@@ -237,10 +210,10 @@ impl DefaultObjectUsecase {

        // Hold the restore-accept guard across the restore-status read, the
        // ongoing/already-restored decision, and the metadata write below, so
-        // two concurrent POST ?restore cannot both observe ongoing=false and
-        // both start a copy-back (backlog#1304). Reads and writes inside this
-        // scope run with no_lock; the guard is dropped before the copy-back is
-        // spawned so it never blocks readers.
+        // two concurrent (non-SELECT) POST ?restore cannot both observe
+        // ongoing=false and both start a copy-back (backlog#1304). Reads and
+        // writes inside this scope run with no_lock; the guard is dropped
+        // before the copy-back is spawned so it never blocks readers.
        // Contention on the accept guard (e.g. a concurrent accept or an
        // in-flight commit on the same object) is transient — answer 503
        // SlowDown so SDK clients back off and retry instead of treating it
@@ -249,7 +222,9 @@ impl DefaultObjectUsecase {
        if store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)? != restore_bucket_incarnation_id {
            return Err(ApiError::from(StorageError::BucketNotFound(bucket.clone())).into());
        }
-        let mut accept_guard = {
+        let mut accept_guard = if is_select {
+            None
+        } else {
            let guard = store
                .acquire_restore_accept_guard(&bucket, &object)
                .await
@@ -258,17 +233,24 @@ impl DefaultObjectUsecase {
            Some(guard)
        };

-        // A missing key or version must stay NoSuchKey / NoSuchVersion, and an
-        // authorization or storage failure must keep its own identity, so map
-        // the storage error instead of flattening it (backlog#2205).
-        let mut obj_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
+        let mut obj_info = store
+            .get_object_info(&bucket, &object, &opts)
+            .await
+            .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrInvalidObjectState".into()), "restore object failed."))?;

-        // Restoring an object that was never transitioned is the S3
-        // InvalidObjectState case, not an internal error.
+        // Check if object is in a transitioned state
        if obj_info.transitioned_object.status != lifecycle::TRANSITION_COMPLETE {
            return Err(S3Error::with_message(
-                S3ErrorCode::InvalidObjectState,
-                "The operation is not valid for the object's storage class.",
+                S3ErrorCode::Custom("ErrInvalidTransitionedState".into()),
+                "restore object failed.",
+            ));
+        }
+
+        // Validate restore request
+        if let Err(e) = validate_restore_request(&rreq, store.clone()) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::Custom("ErrValidRestoreObject".into()),
+                format!("Restore object validation failed: {}", e),
            ));
        }

@@ -278,7 +260,7 @@ impl DefaultObjectUsecase {
        // would create an ABBA cycle. If the probe succeeds, reacquire and
        // re-read the object before replacing the exact orphan generation.
        let mut superseded_worker_guard = None;
-        if obj_info.restore_ongoing {
+        if obj_info.restore_ongoing && !is_select {
            match classify_ongoing_restore(obj_info.user_defined.as_ref(), OffsetDateTime::now_utc()) {
                OngoingRestoreRecovery::ActiveOrUnsafe => {
                    return Err(S3Error::with_message(
@@ -311,11 +293,13 @@ impl DefaultObjectUsecase {
                            .map_err(|_| S3Error::with_message(S3ErrorCode::SlowDown, "restore object failed."))?,
                    );
                    opts.no_lock = true;
-                    obj_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?;
+                    obj_info = store.get_object_info(&bucket, &object, &opts).await.map_err(|_| {
+                        S3Error::with_message(S3ErrorCode::Custom("ErrInvalidObjectState".into()), "restore object failed.")
+                    })?;
                    if obj_info.transitioned_object.status != lifecycle::TRANSITION_COMPLETE {
                        return Err(S3Error::with_message(
-                            S3ErrorCode::InvalidObjectState,
-                            "The operation is not valid for the object's storage class.",
+                            S3ErrorCode::Custom("ErrInvalidTransitionedState".into()),
+                            "restore object failed.",
                        ));
                    }
                    if obj_info.restore_ongoing {
@@ -343,11 +327,11 @@ impl DefaultObjectUsecase {
        remove_str(&mut metadata, SUFFIX_RESTORE_OPERATION_ID);
        remove_str(&mut metadata, SUFFIX_RESTORE_WORKER_LOCK);

+        let mut header = HeaderMap::new();
+
        let event_object_info = obj_info.clone();
        let obj_info_ = obj_info.clone();
-        // Scopes the accept-guarded metadata write: everything below runs
-        // inside the accept critical section, which is released right after.
-        {
+        if !is_select {
            obj_info.metadata_only = true;
            metadata.insert(AMZ_RESTORE_EXPIRY_DAYS.to_string(), rreq.days.unwrap_or(1).to_string());
            let request_date = OffsetDateTime::now_utc().format(&Rfc3339).map_err(|e| {
@@ -419,7 +403,7 @@ impl DefaultObjectUsecase {
                    &restore_dst_opts,
                )
                .await
-                .map_err(ApiError::from)?;
+                .map_err(|_| S3Error::with_message(S3ErrorCode::Custom("ErrCopyObject".into()), "restore object failed."))?;
            rustfs_scanner::record_dirty_usage_bucket(&bucket);
            #[cfg(test)]
            maybe_pause_after_restore_status_commit(&bucket, &object).await;
@@ -445,6 +429,17 @@ impl DefaultObjectUsecase {
        drop(accept_guard);
        drop(restore_bucket_lifecycle_guard);

+        // Handle output location for SELECT requests
+        if let Some(output_location) = &rreq.output_location
+            && let Some(s3) = &output_location.s3
+            && !s3.bucket_name.is_empty()
+        {
+            let restore_object = Uuid::new_v4().to_string();
+            if let Ok(header_value) = format!("{}{}{}", s3.bucket_name, s3.prefix, restore_object).parse() {
+                header.insert(X_AMZ_RESTORE_OUTPUT_PATH, header_value);
+            }
+        }
+
        // Spawn restoration task in the background. Pin the copy-back to the
        // version the accept resolved and flagged: with a versionless request
        // on a versioned bucket, a PUT landing between the accept and the
@@ -504,7 +499,7 @@ impl DefaultObjectUsecase {
            restore_output_path: None,
        };
        helper = helper.object(event_object_info).version_id(version_id_str);
-        let result = Ok(S3Response::new(output));
+        let result = Ok(S3Response::with_headers(output, header));
        let _ = helper.complete(&result);
        result
    }
@@ -634,28 +629,6 @@ mod tests {
        assert_eq!(classify_ongoing_restore(&conflicting_date, now), OngoingRestoreRecovery::ActiveOrUnsafe);
    }

-    fn restore_request(days: Option<i32>) -> RestoreRequest {
-        RestoreRequest {
-            days,
-            description: None,
-            glacier_job_parameters: None,
-            output_location: None,
-            select_parameters: None,
-            tier: None,
-            type_: None,
-        }
-    }
-
-    fn restore_input(bucket: &str, key: &str, rreq: RestoreRequest) -> RestoreObjectInput {
-        RestoreObjectInput::builder()
-            .bucket(bucket.to_string())
-            .key(key.to_string())
-            .restore_request(Some(rreq))
-            .build()
-            .expect("restore input should build")
-    }
-
-    /// backlog#2205: a missing restore body is a client error, not a 500.
    #[tokio::test]
    async fn execute_restore_object_rejects_missing_restore_request() {
        let input = RestoreObjectInput::builder()
@@ -668,92 +641,10 @@ mod tests {
        let usecase = DefaultObjectUsecase::without_context();

        let err = usecase.execute_restore_object(req).await.unwrap_err();
-        assert_eq!(err.code(), &S3ErrorCode::MalformedXML);
-    }
-
-    /// backlog#1341: a SELECT restore must be rejected outright — the restore
-    /// path can only write back to the source key, never to
-    /// `OutputLocation.S3`. Rejection happens before the store is resolved, so
-    /// an uninitialized usecase still answers NotImplemented rather than the
-    /// InternalError every request that gets past this point returns.
-    #[tokio::test]
-    async fn execute_restore_object_rejects_select_type() {
-        let mut rreq = restore_request(None);
-        rreq.type_ = Some(s3s::dto::RestoreRequestType::from_static(s3s::dto::RestoreRequestType::SELECT));
-
-        let req = build_request(restore_input("test-bucket", "test-key", rreq), Method::POST);
-        let usecase = DefaultObjectUsecase::without_context();
-
-        let err = usecase.execute_restore_object(req).await.unwrap_err();
-        assert_eq!(err.code(), &S3ErrorCode::NotImplemented);
-    }
-
-    /// backlog#2205: every RestoreObject failure that reaches storage must
-    /// keep its typed S3 identity. Before this, a missing key, a malformed
-    /// version-id, an illegal `Days` and an object that was never transitioned
-    /// all collapsed into `Custom(...)` codes, which serialize as a retryable
-    /// HTTP 500.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn execute_restore_object_maps_failures_to_typed_s3_errors() {
-        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
-
-        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
-        let context = crate::app::gating_test_env::shared_gating_ambient().await;
-        let bucket = format!("restore-typed-errors-{}", Uuid::new_v4().simple());
-        store
-            .make_bucket(&bucket, &MakeBucketOptions::default())
-            .await
-            .expect("create restore test bucket");
-        let mut reader = PutObjReader::from_vec(b"never transitioned".to_vec());
-        store
-            .put_object(&bucket, "local-object", &mut reader, &ObjectOptions::default())
-            .await
-            .expect("put untransitioned test object");
-
-        let usecase = DefaultObjectUsecase::with_context(Some(context));
-
-        // An illegal `Days` is a client error, rejected before any lock or
-        // object read.
-        let err = usecase
-            .execute_restore_object(build_request(
-                restore_input(&bucket, "local-object", restore_request(Some(0))),
-                Method::POST,
-            ))
-            .await
-            .expect_err("days=0 must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
-
-        // A malformed version-id keeps InvalidArgument instead of being
-        // flattened inside `post_restore_opts`.
-        let mut input = restore_input(&bucket, "local-object", restore_request(Some(1)));
-        input.version_id = Some("not-a-uuid".to_string());
-        let err = usecase
-            .execute_restore_object(build_request(input, Method::POST))
-            .await
-            .expect_err("malformed version-id must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
-
-        // A missing key stays NoSuchKey.
-        let err = usecase
-            .execute_restore_object(build_request(
-                restore_input(&bucket, "missing-object", restore_request(Some(1))),
-                Method::POST,
-            ))
-            .await
-            .expect_err("missing key must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::NoSuchKey);
-
-        // Restoring an object that was never transitioned is the S3
-        // InvalidObjectState case, not an internal error.
-        let err = usecase
-            .execute_restore_object(build_request(
-                restore_input(&bucket, "local-object", restore_request(Some(1))),
-                Method::POST,
-            ))
-            .await
-            .expect_err("untransitioned object must be rejected");
-        assert_eq!(err.code(), &S3ErrorCode::InvalidObjectState);
+        match err.code() {
+            S3ErrorCode::Custom(code) => assert_eq!(code, "ErrValidRestoreObject"),
+            code => panic!("unexpected error code: {:?}", code),
+        }
    }

    #[tokio::test]
@@ -368,8 +368,7 @@ pub(crate) mod bucket {
                version_id: &str,
                bucket: &str,
                object: &str,
-            ) -> Result<crate::storage::storage_api::StorageObjectOptions, crate::storage::storage_api::StorageError>
-            {
+            ) -> Result<crate::storage::storage_api::StorageObjectOptions, std::io::Error> {
                crate::storage::storage_api::ecstore_bucket::lifecycle::bucket_lifecycle_ops::post_restore_opts(
                    version_id, bucket, object,
                )
@@ -14,8 +14,6 @@

 #[cfg(all(feature = "hotpath", feature = "hotpath-alloc", not(target_os = "windows")))]
 use std::alloc::{GlobalAlloc, Layout};
-#[cfg(all(feature = "hotpath", feature = "hotpath-alloc", not(target_os = "windows")))]
-use std::ptr::NonNull;

 #[cfg(all(feature = "hotpath", feature = "hotpath-alloc", not(target_os = "windows")))]
 #[derive(Default)]
@@ -37,10 +35,8 @@ unsafe impl GlobalAlloc for MiMallocAllocator {
    }

    unsafe fn dealloc(&self, ptr: *mut u8, layout: Layout) {
-        // SAFETY: ptr came from this allocator, is non-null by GlobalAlloc's
-        // dealloc contract, and layout.size() is the original allocation size.
-        let ptr = unsafe { NonNull::new_unchecked(ptr) };
-        unsafe { rustfs_mimalloc::MiMalloc::free_csize_nonnull(ptr, layout.size()) }
+        // SAFETY: ptr came from this allocator and layout.size() is the original allocation size.
+        unsafe { rustfs_mimalloc::MiMalloc::free_csize(ptr, layout.size()) }
    }

    unsafe fn realloc(&self, ptr: *mut u8, layout: Layout, new_size: usize) -> *mut u8 {
@@ -14,6 +14,8 @@

 use const_str::concat;
 use shadow_rs::shadow;
+use std::path::Path;
+use std::process::Command;

 shadow!(build);

@@ -45,6 +47,10 @@ pub const DISPLAY_VERSION: &str = {

 type VersionParseResult = Result<(u32, u32, u32, Option<String>), Box<dyn std::error::Error>>;

+fn build_version_override() -> Option<&'static str> {
+    BUILD_VERSION_OVERRIDE.filter(|version| !version.is_empty())
+}
+
 fn version_ref(version: &str) -> String {
    if version.starts_with("refs/tags/") || version.starts_with('@') {
        version.to_string()
@@ -55,7 +61,91 @@ fn version_ref(version: &str) -> String {

 #[allow(clippy::const_is_empty)]
 pub fn get_version() -> String {
-    version_ref(DISPLAY_VERSION)
+    if let Some(version) = build_version_override() {
+        return version_ref(version);
+    }
+
+    // Get the latest tag
+    if let Ok(latest_tag) = get_latest_tag() {
+        // Check if current commit is newer than the latest tag
+        if is_head_newer_than_tag(&latest_tag) {
+            // If current commit is newer, increment the version number
+            if let Ok(new_version) = increment_version(&latest_tag) {
+                return format!("refs/tags/{new_version}");
+            }
+        }
+
+        // If current commit is the latest tag, or version increment failed, return current tag
+        return format!("refs/tags/{latest_tag}");
+    }
+
+    // If no tag exists, use original logic
+    if !build::TAG.is_empty() {
+        format!("refs/tags/{}", build::TAG)
+    } else if !build::SHORT_COMMIT.is_empty() {
+        format!("@{}", build::SHORT_COMMIT)
+    } else {
+        format!("refs/tags/{}", build::PKG_VERSION)
+    }
+}
+
+/// Get the latest git tag
+fn get_latest_tag() -> Result<String, Box<dyn std::error::Error>> {
+    let output = Command::new("git").args(["describe", "--tags", "--abbrev=0"]).output()?;
+
+    if output.status.success() {
+        let tag = String::from_utf8(output.stdout)?;
+        Ok(tag.trim().to_string())
+    } else {
+        Err("Failed to get latest tag".into())
+    }
+}
+
+/// Check if current HEAD is newer than specified tag
+fn is_head_newer_than_tag(tag: &str) -> bool {
+    is_head_newer_than_tag_in(Path::new("."), tag)
+}
+
+fn is_head_newer_than_tag_in(repo: &Path, tag: &str) -> bool {
+    let head = Command::new("git").current_dir(repo).args(["rev-parse", "HEAD"]).output();
+    let tag_commit = Command::new("git")
+        .current_dir(repo)
+        .args(["rev-list", "-n", "1", tag])
+        .output();
+
+    let (Ok(head), Ok(tag_commit)) = (head, tag_commit) else {
+        return false;
+    };
+
+    if !head.status.success() || !tag_commit.status.success() || head.stdout == tag_commit.stdout {
+        return false;
+    }
+
+    let output = Command::new("git")
+        .current_dir(repo)
+        .args(["merge-base", "--is-ancestor", tag, "HEAD"])
+        .output();
+
+    match output {
+        Ok(result) => result.status.success(),
+        Err(_) => false,
+    }
+}
+
+/// Increment version number (increase patch version)
+fn increment_version(version: &str) -> Result<String, Box<dyn std::error::Error>> {
+    // Parse version number, e.g. "1.0.0-alpha.19" -> (1, 0, 0, Some("alpha.19"))
+    let (major, minor, patch, pre_release) = parse_version(version)?;
+
+    // If there's a pre-release identifier, increment the pre-release version number
+    if let Some(pre) = pre_release
+        && let Some(new_pre) = increment_pre_release(&pre)
+    {
+        return Ok(format!("{major}.{minor}.{patch}-{new_pre}"));
+    }
+
+    // Otherwise increment patch version number
+    Ok(format!("{major}.{minor}.{}", patch + 1))
 }

 /// Parse version number
@@ -76,6 +166,28 @@ pub fn parse_version(version: &str) -> VersionParseResult {
    Ok((major, minor, patch, pre_release))
 }

+/// Increment pre-release version number
+fn increment_pre_release(pre_release: &str) -> Option<String> {
+    // Handle pre-release versions like "alpha.19"
+    let parts: Vec<&str> = pre_release.split('.').collect();
+    if parts.len() == 2
+        && let Ok(num) = parts[1].parse::<u32>()
+    {
+        return Some(format!("{}.{}", parts[0], num + 1));
+    }
+
+    // Handle pre-release versions like "alpha19"
+    if let Some(pos) = pre_release.rfind(|c: char| c.is_alphabetic()) {
+        let prefix = &pre_release[..=pos];
+        let suffix = &pre_release[pos + 1..];
+        if let Ok(num) = suffix.parse::<u32>() {
+            return Some(format!("{prefix}{}", num + 1));
+        }
+    }
+
+    None
+}
+
 /// Clean version string - removes common prefixes
 pub fn clean_version(version: &str) -> String {
    version
@@ -172,6 +284,34 @@ mod tests {
    use super::*;
    use tracing::debug;

+    fn run_git(repo: &Path, args: &[&str]) {
+        let status = Command::new("git").current_dir(repo).args(args).status().unwrap();
+        assert!(status.success(), "git command failed: git {}", args.join(" "));
+    }
+
+    #[test]
+    fn test_is_head_newer_than_tag_requires_strict_descendant() {
+        let repo = tempfile::tempdir().unwrap();
+        run_git(repo.path(), &["init", "--quiet"]);
+        run_git(repo.path(), &["config", "user.name", "RustFS Tests"]);
+        run_git(repo.path(), &["config", "user.email", "rustfs@example.com"]);
+        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "tagged commit"]);
+        run_git(repo.path(), &["tag", "--annotate", "1.2.3", "--message", "1.2.3"]);
+
+        assert!(!is_head_newer_than_tag_in(repo.path(), "1.2.3"));
+
+        run_git(repo.path(), &["commit", "--allow-empty", "--quiet", "-m", "newer commit"]);
+
+        assert!(is_head_newer_than_tag_in(repo.path(), "1.2.3"));
+    }
+
+    #[test]
+    fn build_version_override_is_used_for_current_version_when_set() {
+        if let Some(version) = build_version_override() {
+            assert_eq!(get_version(), version_ref(version));
+        }
+    }
+
    #[test]
    fn version_ref_keeps_existing_ref_prefixes() {
        assert_eq!(version_ref("1.2.3"), "refs/tags/1.2.3");
@@ -179,11 +319,6 @@ mod tests {
        assert_eq!(version_ref("@abc123"), "@abc123");
    }

-    #[test]
-    fn get_version_uses_build_metadata() {
-        assert_eq!(get_version(), version_ref(DISPLAY_VERSION));
-    }
-
    #[test]
    fn test_parse_version() {
        // Test standard version parsing
@@ -201,6 +336,27 @@ mod tests {
        assert_eq!(pre_release, Some("alpha.19".to_string()));
    }

+    #[test]
+    fn test_increment_pre_release() {
+        // Test alpha.19 -> alpha.20
+        assert_eq!(increment_pre_release("alpha.19"), Some("alpha.20".to_string()));
+
+        // Test beta.5 -> beta.6
+        assert_eq!(increment_pre_release("beta.5"), Some("beta.6".to_string()));
+
+        // Test unparsable case
+        assert_eq!(increment_pre_release("unknown"), None);
+    }
+
+    #[test]
+    fn test_increment_version() {
+        // Test pre-release version increment
+        assert_eq!(increment_version("1.0.0-alpha.19").unwrap(), "1.0.0-alpha.20");
+
+        // Test standard version increment
+        assert_eq!(increment_version("1.0.0").unwrap(), "1.0.1");
+    }
+
    #[test]
    fn test_version_format() {
        // Test if version format starts with refs/tags/