chore(deps): update crates and fix storage CI regressions (#9111 )

* chore(deps): refresh crates and adopt openidconnect 4.1 Use published OIDC/OAuth2 releases and reject malformed HMAC signature lengths before upstream verification. Refresh workspace and fuzz locks, and advance the local zstd compatibility patch to 0.13.4 while retaining the fixed 0.14.3 codec backend. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * chore(deps): refresh compatible lockfile resolutions Resolve Prost build and derive dependencies with itertools 0.14 and prefer windows-sys 0.61.2 within the existing supported version ranges. Retain all package versions, sources and checksums. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(connect): separate cleanup checks from short probe windows Use the functional measurement budget for the cleanup HTTP fixture so setup and measured I/O can finish before checking delete status handling. Exercise both normal scheduling and an in-flight PUT clock advance, and include target classification in unexpected-failure diagnostics. Retain the cleanup deadline, outer timeout and delete-stage assertions. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(ecstore): isolate stalled multipart disk timeouts Apply the short injected timeout only to the deliberately stalled disk. Healthy disks retain their normal operation budgets, while logical lock, physical owner, immediate CopyObject and final shard assertions remain. Keep overrides scoped across cloned and reconnected handles, preserve shorter caller deadlines and explicit zero timeouts, and report drive health when the real CopyObject assertion fails. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(rebalance): retry metadata write lock quorum admission Recognize typed write-lock quorum failures for rebalance.bin within the existing bounded metadata admission retry policy. Each attempt must acquire quorum before loading or saving metadata; data writes are not replayed by this retry. Cover recovered and persistent quorum failures, cancellation after a failed attempt, and rejection of unrelated resource or read-lock errors. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
feat(tls): add opt-in fips feature for aws-lc-rs and rustls (#9084 )
2026-11-02 09:04:14 +00:00 · 2026-11-02 06:49:24 +00:00 · 2026-11-02 04:27:05 +00:00 · 2026-11-02 03:59:27 +00:00 · 2026-11-02 02:43:28 +00:00 · 2026-11-02 09:03:15 +08:00
42 changed files with 2344 additions and 250 deletions
@@ -234,7 +234,7 @@ test-group = 'embedded-test-ports'
 # processes cannot exhaust those budgets. Keep deadlines and assertions intact.
 # The short HTTP fixtures are tracked in rustfs/backlog#2714.
 [[profile.default.overrides]]
-filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|oversized_response_is_rejected_without_buffering_it)$/)))'
+filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|cleanup_accepts_missing_object_and_still_deletes_the_bucket|cleanup_diagnostics_identify_each_disconnected_transport_stage|cleanup_diagnostics_identify_each_failed_http_stage_without_private_context|cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_deadline|oversized_response_is_rejected_without_buffering_it)$/)))'
 threads-required = "num-test-threads"

 # This cancellation probe hashes the serving executable before acquiring its
@@ -548,7 +548,7 @@ filter = 'package(rustfs) & (binary(/^embedded.*_test$/) | binary(admin_diagnost
 test-group = 'embedded-test-ports'

 [[profile.ci.overrides]]
-filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|oversized_response_is_rejected_without_buffering_it)$/)))'
+filter = 'package(rustfs) & (binary(/^connect_registration$/) | binary(connect_inventory) | binary(connect_perf_drive) | (binary(connect_perf_client) & test(=configured_proxy_and_direct_endpoint_failures_are_distinct)) | (binary(connect_perf_object) & test(/^(real_rustfs_endpoint_and_production_cli_support_bounded_get_and_put|cleanup_uses_a_budget_independent_of_the_measurement_window|cleanup_accepts_missing_object_and_still_deletes_the_bucket|cleanup_diagnostics_identify_each_disconnected_transport_stage|cleanup_diagnostics_identify_each_failed_http_stage_without_private_context|cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_deadline|oversized_response_is_rejected_without_buffering_it)$/)))'
 threads-required = "num-test-threads"

 # Preserve the cancellation probe's startup budget and assertions in CI too.
@@ -976,6 +976,12 @@ test-group = 'e2e-vault'
 filter = 'package(e2e_test) & (test(/^list_objects_v2_pagination_test::tests::(test_list_objects_v2_delimiter_small_page_traverses_all|test_list_objects_v2_max_keys_above_limit_returns_token|test_list_objects_v2_maxkeys_above_limit_with_delimiter)$/) | test(=reliant::tiering::test_hermetic_transition_restore_failure_expiry_and_retry))'
 threads-required = "num-test-threads"

+# The mc fixture uploads hundreds of objects concurrently before measuring listing.
+# Reserve the runner so other server fixtures do not starve its disk operations.
+[[profile.e2e-full.overrides]]
+filter = 'package(e2e_test) & test(=mc_mirror_small_bucket_test::test_mc_mirror_small_bucket_completes_without_list_timeout)'
+threads-required = "num-test-threads"
+
 # Fixed, small contract for every executable server change. These existing
 # scenarios exercise write/read, anonymous denial and retained bytes on restart.
 [profile.core-contract]
@@ -296,6 +296,32 @@
      "origin": "baseline",
      "alert_contract": "scheduled-validation"
    },
+    {
+      "workflow": ".github/workflows/rustfs-validation-sweep.yml",
+      "name": "RustFS Validation Sweep",
+      "roles": [
+        "periodic",
+        "manual"
+      ],
+      "events": [
+        "schedule",
+        "workflow_dispatch"
+      ],
+      "cadence": "nightly",
+      "timezone": "UTC",
+      "schedule": [
+        "30 19 * * *"
+      ],
+      "owner": "@overtrue",
+      "policy": "active",
+      "monitor": "workflow",
+      "max_age_hours": 60,
+      "result_owner": "rustfs-validation-sweep.yml",
+      "disposition": "canonical validation cases executed in rotating batches",
+      "failure_owner": "workflow",
+      "origin": "baseline",
+      "alert_contract": "scheduled-validation"
+    },
    {
      "workflow": ".github/workflows/btf-prototype-reference.yml",
      "name": "BTF prototype reference",
@@ -0,0 +1,421 @@
+# 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.
+
+name: RustFS Validation Sweep
+
+# Executes the migrated canonical validation cases (rustfs/auto-testing#113)
+# on the shared functional fleet. Everything with a lane executor is swept in
+# rotating nightly batches so the whole set is covered roughly weekly; a
+# single case or one module can be dispatched manually. That is the
+# implemented catalog (66 today) plus the eight planned cases whose dedicated
+# executors already exist and have live PASS evidence - fresh nightly
+# evidence feeds their acceptance review. heal.* cases stay on the
+# interactive validation flow: they require reviewed native/tool collectors
+# built by separate validation-tools runs before fleet execution.
+
+on:
+  schedule:
+    # After the nightly functional chain settles (chain starts ~16:48 UTC and
+    # runs for about two hours).
+    - cron: "30 19 * * *"
+  workflow_dispatch:
+    inputs:
+      case:
+        description: 'Single canonical validation case id (overrides module and rotation), e.g. pool.decommission'
+        type: string
+        required: false
+        default: ''
+      module:
+        description: 'Restrict the sweep to one catalog module, e.g. bucket-replication'
+        type: string
+        required: false
+        default: ''
+      limit:
+        description: 'Maximum cases per sweep (0 = run everything selected)'
+        type: number
+        required: false
+        default: 12
+      build_run_id:
+        description: 'Successful nightly package run; empty resolves the latest'
+        type: string
+        required: false
+        default: ''
+      build_run_attempt:
+        description: 'Attempt of that package run'
+        type: string
+        required: false
+        default: '1'
+
+permissions:
+  contents: read
+
+concurrency:
+  group: rustfs-validation-sweep
+  cancel-in-progress: false
+
+env:
+  RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES || 'vm000 vm001 vm002 vm003' }}
+  RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER || 'azureuser' }}
+  RUSTFS_ACCESS_KEY: ${{ secrets.RUSTFS_ACCESS_KEY }}
+  RUSTFS_SECRET_KEY: ${{ secrets.RUSTFS_SECRET_KEY }}
+  RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
+  PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+
+jobs:
+  sweep:
+    if: github.repository == 'rustfs/rustfs'
+    runs-on: smoke-testing
+    timeout-minutes: 150
+    steps:
+      - name: Checkout chain tooling
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Resolve nightly candidate
+        id: candidate
+        env:
+          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          BUILD_RUN_ID: ${{ inputs.build_run_id }}
+          BUILD_RUN_ATTEMPT: ${{ inputs.build_run_attempt }}
+          CHAIN_OUTPUT: ${{ runner.temp }}/validation-sweep-candidate.json
+        run: |
+          set -euo pipefail
+          python3 scripts/resolve_functional_candidate.py
+          python3 - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+          root = Path(os.environ["RUNNER_TEMP"]) / "validation-sweep"
+          root.mkdir(parents=True, exist_ok=True)
+          candidate = json.loads((Path(os.environ["RUNNER_TEMP"]) / "validation-sweep-candidate.json").read_text())
+          manifest = candidate["manifest"]
+          (root / "candidate.json").write_text(json.dumps({
+              "package_url": manifest["package_url"],
+              "package_sha256": manifest["package_sha256"],
+              "source_sha": manifest["source_sha"],
+          }, sort_keys=True) + "\n")
+          with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+              output.write("testing_sha=" + candidate["testing_sha"] + "\n")
+              output.write("package_url=" + manifest["package_url"] + "\n")
+              output.write("source_sha=" + manifest["source_sha"] + "\n")
+          PY
+
+      - name: Checkout auto-testing scripts
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          repository: rustfs/auto-testing
+          ref: ${{ steps.candidate.outputs.testing_sha }}
+          token: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          path: auto-testing
+          persist-credentials: false
+
+      - name: Select sweep cases
+        id: select
+        env:
+          SELECTED_CASE: ${{ inputs.case }}
+          SELECTED_MODULE: ${{ inputs.module }}
+          CASE_LIMIT: ${{ inputs.limit }}
+          SWEEP_EVENT: ${{ github.event_name }}
+          SWEEP_ROOT: ${{ runner.temp }}/validation-sweep
+        run: |
+          set -euo pipefail
+          python3 - <<'PY'
+          import datetime
+          import json
+          import os
+          from pathlib import Path
+          sweep_root = Path(os.environ["SWEEP_ROOT"])
+
+          catalog = json.loads(Path("auto-testing/validation-cases.json").read_text())
+          def sweepable(case):
+              status = case.get("source_status")
+              if status == "implemented":
+                  return True
+              # The remaining planned non-heal cases route to the lane
+              # executors and already have live PASS evidence; sweeping them
+              # nightly produces fresh evidence for their acceptance review.
+              return status == "planned" and not case["id"].startswith("heal.")
+
+          selected = [(case["id"], module["id"]) for module in catalog["modules"]
+                      for case in module["cases"]
+                      if sweepable(case)]
+          if not selected:
+              raise SystemExit("catalog has no sweepable cases")
+
+          one_case = os.environ.get("SELECTED_CASE", "").strip()
+          module_filter = os.environ.get("SELECTED_MODULE", "").strip()
+          known = {case_id for case_id, _ in selected}
+          modules = {module_id for _, module_id in selected}
+          if one_case:
+              if one_case not in known:
+                  raise SystemExit(f"unknown case or no lane executor for it: {one_case}")
+              selected = [(one_case, dict(selected)[one_case])]
+          elif module_filter:
+              if module_filter not in modules:
+                  raise SystemExit(f"unknown module or no implemented cases: {module_filter}")
+              selected = [pair for pair in selected if pair[1] == module_filter]
+          elif os.environ.get("SWEEP_EVENT") == "schedule":
+              # Stateless rotation: every night take one window of the catalog
+              # so a full sweep needs len(selected)/limit nights and no state
+              # has to be carried between runs.
+              limit = max(1, int(os.environ.get("CASE_LIMIT") or 12))
+              day = datetime.datetime.now(datetime.timezone.utc).timetuple().tm_yday
+              start = (day * limit) % len(selected)
+              selected = (selected[start:] + selected[:start])[:limit]
+          limit = int(os.environ.get("CASE_LIMIT") or 0)
+          if one_case or module_filter or (not os.environ.get("SWEEP_EVENT") == "schedule"):
+              if limit > 0:
+                  selected = selected[:limit]
+
+          (sweep_root / "sweep-cases.txt").write_text("".join(f"{case_id}\n" for case_id, _ in selected))
+          with open(os.environ["GITHUB_OUTPUT"], "a") as output:
+              output.write(f"count={len(selected)}\n")
+          print(f"selected {len(selected)} case(s):", ", ".join(c for c, _ in selected))
+          PY
+
+      - name: Admit clean shared environment
+        id: environment
+        timeout-minutes: 5
+        env:
+          FUNCTIONAL_ENVIRONMENT_RECEIPT: ${{ runner.temp }}/environment-validation-sweep.json
+        run: python3 scripts/functional_environment.py enter --suite heal
+
+      - name: Run validation cases
+        id: run_cases
+        timeout-minutes: 120
+        env:
+          AUTO_TESTING_SHA: ${{ steps.candidate.outputs.testing_sha }}
+          SWEEP_ROOT: ${{ runner.temp }}/validation-sweep
+        run: |
+          set -uo pipefail
+          mkdir -p "${SWEEP_ROOT}/results"
+          : > "${SWEEP_ROOT}/results.txt"
+          while read -r CASE_ID; do
+            [ -n "${CASE_ID}" ] || continue
+            echo "::group::case ${CASE_ID}"
+            rm -rf "${SWEEP_ROOT}/run"
+            if ./auto-testing/scripts/run_validation_case.sh \
+                 "${CASE_ID}" "${SWEEP_ROOT}/candidate.json" \
+                 "${AUTO_TESTING_SHA}" "${SWEEP_ROOT}/run" \
+                 > "${SWEEP_ROOT}/results/${CASE_ID//\//_}.log" 2>&1; then
+              printf '%s\tPASS\n' "${CASE_ID}" >> "${SWEEP_ROOT}/results.txt"
+              echo "case ${CASE_ID}: PASS"
+            else
+              printf '%s\tFAIL\n' "${CASE_ID}" >> "${SWEEP_ROOT}/results.txt"
+              echo "case ${CASE_ID}: FAIL (log retained in the sweep artifact)"
+            fi
+            if [ -f "${SWEEP_ROOT}/run/report.md" ]; then
+              cp "${SWEEP_ROOT}/run/report.md" "${SWEEP_ROOT}/results/${CASE_ID//\//_}.report.md" 2>/dev/null || true
+            fi
+            echo "::endgroup::"
+            # Fleet hygiene between cases: the case runners set up their own
+            # topology, this purge guarantees the next case starts clean even
+            # when the previous one died mid-run.
+            for node in ${RUSTFS_NODES}; do
+              ssh -o BatchMode=yes -o ConnectTimeout=10 "${RUSTFS_SSH_USER}@${node}" '
+                SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
+                ${SUDO} systemctl stop rustfs 2>/dev/null || true
+                if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
+                  ${SUDO} dpkg -P rustfs
+                fi
+                for i in 1 2 3 4; do ${SUDO} rm -rf /data/rustfs${i}/mnmd; done
+                ${SUDO} rm -f /etc/default/rustfs
+              ' || echo "inter-case cleanup failed on ${node} (continuing)"
+            done
+          done < "${SWEEP_ROOT}/sweep-cases.txt"
+          echo "sweep finished: $(wc -l < "${SWEEP_ROOT}/results.txt") case(s) executed"
+
+      - name: Generate sweep report
+        if: ${{ always() && steps.environment.outcome == 'success' }}
+        env:
+          SWEEP_ROOT: ${{ runner.temp }}/validation-sweep
+          PACKAGE_URL: ${{ steps.candidate.outputs.package_url }}
+          SOURCE_SHA: ${{ steps.candidate.outputs.source_sha }}
+        run: |
+          set -euo pipefail
+          python3 - <<'PY'
+          import json
+          import os
+          from pathlib import Path
+
+          root = Path(os.environ["SWEEP_ROOT"])
+          catalog = json.loads(Path("auto-testing/validation-cases.json").read_text())
+          module_of = {case["id"]: module["id"] for module in catalog["modules"] for case in module["cases"]}
+          rows = []
+          for line in (root / "results.txt").read_text().splitlines():
+            if not line:
+                continue
+            case_id, status = line.split("\t")
+            rows.append((case_id, module_of.get(case_id, "?"), status))
+          passed = sum(1 for _, _, s in rows if s == "PASS")
+          failed = len(rows) - passed
+          run_url = (f"{os.environ.get('GITHUB_SERVER_URL', 'https://github.com')}/"
+                     f"{os.environ.get('GITHUB_REPOSITORY', 'rustfs/rustfs')}/actions/runs/"
+                     f"{os.environ.get('GITHUB_RUN_ID', '')}")
+
+          report = [
+              "# RustFS validation sweep report", "",
+              f"- Run: {run_url}",
+              f"- Attempt: {os.environ.get('GITHUB_RUN_ATTEMPT', '')}",
+              f"- Trigger: {os.environ.get('GITHUB_EVENT_NAME', '')}",
+              f"- Package: {os.environ.get('PACKAGE_URL', 'nightly (R2 latest)')}",
+              f"- Source SHA: {os.environ.get('SOURCE_SHA', '')}",
+              f"- auto-testing: {os.environ.get('AUTO_TESTING_SHA', '')}", "",
+              "Canonical validation cases (rustfs/auto-testing#113) executed by the sweep, grouped by catalog module.", "",
+          ]
+          # Group the executed cases by catalog module (the major category:
+          # pool lifecycle, heal, ILM expiration, bucket/site replication, kms,
+          # data services, identity, s3 compatibility). Per-module subtotals
+          # deliberately use "- Subtotal:" so the dashboard overview regexes
+          # (which anchor on "- Total:") keep reading the grand totals below.
+          by_module = {}
+          module_order = []
+          for case_id, module, status in rows:
+              if module not in by_module:
+                  by_module[module] = []
+                  module_order.append(module)
+              by_module[module].append((case_id, status))
+          report += ["## Case Summary", ""]
+          for module in module_order:
+              cases = by_module[module]
+              mod_pass = sum(1 for _, st in cases if st == "PASS")
+              mod_fail = len(cases) - mod_pass
+              report += [f"### {module} ({len(cases)})", "",
+                         "| Case | Status |", "| --- | --- |"]
+              report += [f"| {case_id} | {status} |" for case_id, status in cases]
+              report += ["", f"- Subtotal: {mod_pass} passed, {mod_fail} failed", ""]
+          report += [
+              "## Totals", "",
+              f"- Total: {len(rows)}",
+              f"- PASS: {passed}",
+              f"- FAIL: {failed}",
+              "- UNSUPPORTED: 0", "",
+              "Failing cases keep this workflow red only as a report; the sweep artifact",
+              "retains every case log and per-case report for diagnosis.", "",
+          ]
+          (root / "report.md").write_text("\n".join(report))
+          (root / "cases.md").write_text("\n".join(f"| {c} | {m} | {s} |" for c, m, s in rows) + "\n")
+          PY
+          cat "${SWEEP_ROOT}/report.md" >> "${GITHUB_STEP_SUMMARY}"
+
+      - name: Upload functional report to dashboard
+        if: ${{ always() && steps.environment.outcome == 'success' }}
+        timeout-minutes: 2
+        continue-on-error: true
+        env:
+          GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: ${{ runner.temp }}/validation-sweep/report.md
+          SUITE: validation
+        run: |
+          set -euo pipefail
+          if [ -z "${GH_TOKEN:-}" ]; then
+            echo "PF_TESTING_GH_TOKEN is not configured; skipping dashboard upload"
+            exit 0
+          fi
+          DATE="$(date -u +%Y-%m-%d)"
+          REPORT_PATH="functional-reports/${SUITE}/${DATE}.md"
+          B64_FILE="$(mktemp)"
+          python3 -c 'import base64,sys;print(base64.b64encode(open(sys.argv[1],"rb").read()).decode())' "${REPORT_FILE}" > "${B64_FILE}"
+          SHA="$(gh api "repos/rustfs/dashboard/contents/${REPORT_PATH}" -q '.sha' 2>/dev/null || true)"
+          if [ -n "${SHA}" ]; then
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --rawfile content "${B64_FILE}" --arg sha "${SHA}" \
+              '{message:$msg, content:($content|rtrimstr("\n")), sha:$sha}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          else
+            jq -n --arg msg "report(${SUITE}): ${DATE}" --rawfile content "${B64_FILE}" \
+              '{message:$msg, content:($content|rtrimstr("\n"))}' \
+              | gh api --method PUT "repos/rustfs/dashboard/contents/${REPORT_PATH}" --input - >/dev/null
+          fi
+          echo "dashboard updated: ${REPORT_PATH}"
+
+      - name: Retain sweep evidence
+        if: ${{ always() }}
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: rustfs-validation-sweep-${{ github.run_id }}-${{ github.run_attempt }}
+          path: |
+            ${{ runner.temp }}/validation-sweep/report.md
+            ${{ runner.temp }}/validation-sweep/results.txt
+            ${{ runner.temp }}/validation-sweep/sweep-cases.txt
+            ${{ runner.temp }}/validation-sweep/candidate.json
+            ${{ runner.temp }}/validation-sweep/results/
+          if-no-files-found: error
+          retention-days: 30
+
+      - name: Cleanup environment (after)
+        id: environment_cleanup
+        timeout-minutes: 5
+        if: ${{ always() && steps.environment.outcome == 'success' }}
+        run: |
+          set -uo pipefail
+          cleanup_status=0
+          for node in ${RUSTFS_NODES}; do
+            ssh -o BatchMode=yes -o ConnectTimeout=10 "${RUSTFS_SSH_USER}@${node}" '
+              set -euo pipefail
+              SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
+              ${SUDO} systemctl stop rustfs 2>/dev/null || true
+              if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
+                ${SUDO} dpkg -P rustfs
+              fi
+              for i in 1 2 3 4; do ${SUDO} rm -rf /data/rustfs${i}/mnmd; done
+              ${SUDO} rm -f /etc/default/rustfs
+            ' || cleanup_status=$?
+          done
+          exit "${cleanup_status}"
+
+      - name: Release environment only after successful cleanup
+        if: ${{ always() && steps.environment.outcome == 'success' }}
+        timeout-minutes: 5
+        env:
+          FUNCTIONAL_ENVIRONMENT_RECEIPT: ${{ runner.temp }}/environment-validation-sweep.json
+          CLEANUP_OUTCOME: ${{ steps.environment_cleanup.outcome }}
+        run: python3 scripts/functional_environment.py finish --suite heal --cleanup-outcome "$CLEANUP_OUTCOME"
+
+      - name: Notify on failure
+        if: failure()
+        run: echo "RustFS validation sweep failed (harness/environment breakdown); see the sweep artifact for case logs."
+
+      - name: Enforce sweep result
+        if: ${{ always() && steps.environment.outcome == 'success' }}
+        env:
+          SWEEP_ROOT: ${{ runner.temp }}/validation-sweep
+        run: |
+          set -euo pipefail
+          if grep -q $'\tFAIL$' "${SWEEP_ROOT}/results.txt"; then
+            echo "validation sweep has failing cases:"
+            grep $'\tFAIL$' "${SWEEP_ROOT}/results.txt"
+            exit 1
+          fi
+
+  alert-on-failure:
+    name: Alert on scheduled failure
+    # A scheduled sweep failure usually means a canonical validation case
+    # regressed or the fleet environment broke; surface it as a tracking
+    # issue instead of leaving it in the Actions tab.
+    needs: [sweep]
+    if: github.repository == 'rustfs/rustfs' && (always() && github.event_name == 'schedule' && contains(needs.*.result, 'failure'))
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    permissions:
+      contents: read
+      issues: write
+    steps:
+      - uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v5.0.0
+        with:
+          persist-credentials: false
+      - name: Open or update failure-tracking issue
+        uses: ./.github/actions/schedule-failure-issue
+        with:
+          github-token: ${{ secrets.BACKLOG_ISSUE_TOKEN }}
@@ -299,7 +299,7 @@ dependencies = [
 "strum",
 "thiserror 2.0.21",
 "uuid",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]

 [[package]]
@@ -492,7 +492,7 @@ dependencies = [
 "arrow-select",
 "flatbuffers",
 "lz4_flex",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]

 [[package]]
@@ -908,12 +908,28 @@ dependencies = [
 "zeroize",
 ]

+[[package]]
+name = "aws-lc-fips-sys"
+version = "0.14.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b744eed18cd75d814effdf2aba2880f912abe0fe1c327b361341d31aba9a7005"
+dependencies = [
+ "bindgen",
+ "cc",
+ "cmake",
+ "dunce",
+ "fs_extra",
+ "pkg-config",
+ "regex",
+]
+
 [[package]]
 name = "aws-lc-rs"
 version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "048006790404103c488e6c5616b3e7e702a71292aadda08e8f7698a9e0d2eb53"
 dependencies = [
+ "aws-lc-fips-sys",
 "aws-lc-sys",
 "untrusted 0.7.1",
 "zeroize",
@@ -960,9 +976,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-kms"
-version = "1.128.0"
+version = "1.129.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef519fda62fd0fdc066948f5ef33ecb8ae1bf0187bf1a12e8a775c93c4c47e7e"
+checksum = "5c5cdb29ad25d71c9ee54c408a672c98d90a24a0ea6eb063025483e0e894b411"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -985,9 +1001,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.156.0"
+version = "1.157.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1072c9aa20159b9b417115342d38c4c53108fc98e5359fcb86b7d4575747eb59"
+checksum = "6ea9794c741520d8baec9462bcd0ad12164d4f5f78d008a8da9d6a0b7222caa8"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1021,9 +1037,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sso"
-version = "1.118.0"
+version = "1.119.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4a89786db8ee6dff2c9302a354ef081354fcb0f527e16ee3cb249a6a5cb6df7"
+checksum = "3c414f25ec4aa9dc18f5cc8f55e18848bb509fcda4ce7c77bee6072e95806bc9"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1046,9 +1062,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.120.0"
+version = "1.121.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "521c6a11a25f1244efddf947523f277a51a63f0a7ec228f0f62d826e88106366"
+checksum = "bc638bb4355e6514334dcc3228cd4622f5ee54d704bd6a9010a5a91d7856d9d9"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1071,9 +1087,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-sts"
-version = "1.123.0"
+version = "1.124.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8754d38d29448bd1d4aa4a6312ed010dfa296b3436eb2c551cc6e42fc6a66a38"
+checksum = "3f8a0dd8fbde04ba58ad722113c7addb11bf94cf549b1ea0eb9ea427fb87198a"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1524,6 +1540,8 @@ dependencies = [
 "bitflags 2.13.2",
 "cexpr",
 "clang-sys",
+ "log",
+ "prettyplease 0.3.0",
 "proc-macro2",
 "quote",
 "regex",
@@ -3942,6 +3960,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
 "block-buffer 0.10.4",
+ "const-oid 0.9.6",
 "crypto-common 0.1.7",
 "subtle",
 ]
@@ -5277,9 +5296,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b39e941010d86063ff233bf8325269bf68097fb24ba67000fbddc0a410e8056e"
+checksum = "d9632c1fd213e63c1b88d8f3e59543da4999b17f859cc1e1d67cf8f2207e8935"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5312,15 +5331,15 @@ dependencies = [

 [[package]]
 name = "hotpath-drain"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d0faae1a9e1fd8ce5debaaaf452dcbe833fc8a8a4f3779d2ec8828b116c3c16a"
+checksum = "738324f75aec847dea984f6806402f4f1bdfab437b4700fc8c3ebcad2dff6424"

 [[package]]
 name = "hotpath-macros"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a24f5179ff680132003306354efd0ac8cde12e87b1a788719c836e91af35cb58"
+checksum = "5d92fdfec803932e5764ea93c201cf7c6c2d8cff1871da44f64f5094c9c3ef8a"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -5941,7 +5960,7 @@ dependencies = [
 "lz4",
 "snap",
 "uuid",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]

 [[package]]
@@ -6034,6 +6053,9 @@ name = "lazy_static"
 version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb"
+dependencies = [
+ "spin 0.9.9",
+]

 [[package]]
 name = "lexical-core"
@@ -6243,9 +6265,9 @@ dependencies = [

 [[package]]
 name = "log"
-version = "0.4.34"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c"

 [[package]]
 name = "lru"
@@ -6609,9 +6631,9 @@ dependencies = [

 [[package]]
 name = "ml-kem"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9ba01ebcf90f6aff8e9b9f55bb69b19e6250e4ade5787d0040eaa27dd226238"
+checksum = "84ca78b66f46050d49fa21693ae8adab2081d8101b65d8067316b6b70b64a0d1"
 dependencies = [
 "hybrid-array",
 "kem",
@@ -6624,9 +6646,9 @@ dependencies = [

 [[package]]
 name = "module-lattice"
-version = "0.2.3"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c61b87c9683ab7cb1c6871d261ad5479b6b10ceb52c4352aaca3b5d35a8febe"
+checksum = "c4a1e412c8ee83a1c678f427806219c49c96b2511b9f07497514f0e43fb26155"
 dependencies = [
 "ctutils",
 "hybrid-array",
@@ -6922,6 +6944,22 @@ dependencies = [
 "rand_core 0.10.1",
 ]

+[[package]]
+name = "num-bigint-dig"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7"
+dependencies = [
+ "lazy_static",
+ "libm",
+ "num-integer",
+ "num-iter",
+ "num-traits",
+ "rand 0.8.8",
+ "smallvec",
+ "zeroize",
+]
+
 [[package]]
 name = "num-complex"
 version = "0.4.6"
@@ -6946,6 +6984,16 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "num-iter"
+version = "0.1.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c92800bd69a1eac91786bcfe9da64a897eb72911b8dc3095decbd07429e8048b"
+dependencies = [
+ "num-integer",
+ "num-traits",
+]
+
 [[package]]
 name = "num-rational"
 version = "0.4.2"
@@ -7017,18 +7065,19 @@ dependencies = [

 [[package]]
 name = "oauth2"
-version = "5.1.0"
-source = "git+https://github.com/ramosbugs/oauth2-rs?rev=89585f4a280afd7b285eaf1c9be0068874e03d22#89585f4a280afd7b285eaf1c9be0068874e03d22"
+version = "5.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
 dependencies = [
 "base64 0.22.1",
 "chrono",
 "getrandom 0.2.17",
 "http 1.5.0",
- "rand 0.10.3",
+ "rand 0.8.8",
 "serde",
 "serde_json",
 "serde_path_to_error",
- "sha2 0.11.1",
+ "sha2 0.10.9",
 "thiserror 1.0.69",
 "url",
 ]
@@ -7190,29 +7239,31 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"

 [[package]]
 name = "openidconnect"
-version = "4.0.1"
-source = "git+https://github.com/ramosbugs/openidconnect-rs?rev=2f77ee73bc4554e1cb6279adb386005429fb4e41#2f77ee73bc4554e1cb6279adb386005429fb4e41"
+version = "4.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df0d098a59353679624d406825940139e2b2d554a52b64c9fb1b1bc42a11fe5c"
 dependencies = [
 "base64 0.22.1",
 "chrono",
 "dyn-clone",
 "ed25519-dalek 3.0.0",
- "hmac 0.13.0",
+ "getrandom 0.4.3",
+ "hmac 0.12.1",
 "http 1.5.0",
 "itertools 0.14.0",
 "log",
 "oauth2",
 "p256",
 "p384",
- "rand 0.10.3",
- "rsa",
+ "rand 0.8.8",
+ "rsa 0.9.10",
 "serde",
 "serde-value",
 "serde_json",
 "serde_path_to_error",
 "serde_plain",
 "serde_with",
- "sha2 0.11.1",
+ "sha2 0.10.9",
 "subtle",
 "thiserror 1.0.69",
 "url",
@@ -7564,7 +7615,7 @@ dependencies = [
 "snap",
 "tokio",
 "twox-hash",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]

 [[package]]
@@ -7816,6 +7867,17 @@ dependencies = [
 "futures-io",
 ]

+[[package]]
+name = "pkcs1"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f"
+dependencies = [
+ "der 0.7.10",
+ "pkcs8 0.10.2",
+ "spki 0.7.3",
+]
+
 [[package]]
 name = "pkcs1"
 version = "0.8.0-rc.4"
@@ -8914,9 +8976,9 @@ dependencies = [

 [[package]]
 name = "reflink-copy"
-version = "0.1.30"
+version = "0.1.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9dd7ab4af0363d5ccfd2838d782a28196cf32a5cc2e4fe3c5dc83f2be588b8b"
+checksum = "3dd31f458a66eeda05a9e837fc5f0effbaacbe8565868ecae5b37bf0544c19f5"
 dependencies = [
 "cfg-if",
 "libc",
@@ -9078,6 +9140,26 @@ dependencies = [
 "xmlparser",
 ]

+[[package]]
+name = "rsa"
+version = "0.9.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d"
+dependencies = [
+ "const-oid 0.9.6",
+ "digest 0.10.7",
+ "num-bigint-dig",
+ "num-integer",
+ "num-traits",
+ "pkcs1 0.7.5",
+ "pkcs8 0.10.2",
+ "rand_core 0.6.4",
+ "signature 2.2.0",
+ "spki 0.7.3",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "rsa"
 version = "0.10.0-rc.18"
@@ -9088,7 +9170,7 @@ dependencies = [
 "crypto-bigint",
 "crypto-primes",
 "digest 0.11.3",
- "pkcs1",
+ "pkcs1 0.8.0-rc.4",
 "pkcs8 0.11.0",
 "rand_core 0.10.1",
 "sha2 0.11.1",
@@ -9193,13 +9275,13 @@ dependencies = [
 "p521",
 "pageant",
 "pbkdf2 0.13.0",
- "pkcs1",
+ "pkcs1 0.8.0-rc.4",
 "pkcs5 0.8.1",
 "pkcs8 0.11.0",
 "polyval",
 "rand 0.10.3",
 "rand_core 0.10.1",
- "rsa",
+ "rsa 0.10.0-rc.18",
 "russh-cryptovec",
 "russh-util",
 "salsa20 0.11.0",
@@ -9391,7 +9473,7 @@ dependencies = [
 "regex",
 "reqwest",
 "rmp-serde",
- "rsa",
+ "rsa 0.10.0-rc.18",
 "rust-embed",
 "rustfs-audit",
 "rustfs-common",
@@ -9594,7 +9676,7 @@ dependencies = [
 "jsonwebtoken",
 "pbkdf2 0.13.0",
 "rand 0.10.3",
- "rsa",
+ "rsa 0.10.0-rc.18",
 "serde",
 "serde_json",
 "sha2 0.11.1",
@@ -11988,7 +12070,7 @@ dependencies = [
 "p384",
 "p521",
 "rand_core 0.10.1",
- "rsa",
+ "rsa 0.10.0-rc.18",
 "sec1",
 "serde",
 "sha1 0.11.0",
@@ -13096,9 +13178,9 @@ dependencies = [

 [[package]]
 name = "unicode-properties"
-version = "0.1.4"
+version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
+checksum = "993ad287eb20e8ba755312801cfe1413e0f25d604075863acaddca10b04c1833"

 [[package]]
 name = "unicode-segmentation"
@@ -14000,9 +14082,9 @@ dependencies = [

 [[package]]
 name = "zip"
-version = "9.1.3"
+version = "9.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1156b39d142a34b999c2310a50455e0cb3760bf7efad0122f796d813bee26ccb"
+checksum = "848dad21639597378c3be94c59fd4de683f0f6d8a5387575896e1245737d4f37"
 dependencies = [
 "aes 0.9.3",
 "bzip2",
@@ -14053,7 +14135,7 @@ dependencies = [

 [[package]]
 name = "zstd"
-version = "0.13.3"
+version = "0.13.4"
 dependencies = [
 "apache-avro",
 "arrow-array",
@@ -217,7 +217,7 @@ chacha20poly1305 = { version = "0.11.0" }
 crc-fast = "1.10.0"
 hmac = { version = "0.13.0" }
 jsonwebtoken = { version = "11.1.0" }
-openidconnect = { default-features = false, version = "4.0" }
+openidconnect = { default-features = false, version = "4.1.0" }
 pbkdf2 = "0.13.0"
 p256 = { version = "0.14.0", features = ["ecdsa", "pkcs8"] }
 rsa = { version = "=0.10.0-rc.18" }
@@ -253,9 +253,9 @@ atoi = "3.1.1"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.13.0" }
 aws-credential-types = { version = "1.3.1" }
-aws-sdk-kms = { default-features = false, version = "1.128.0" }
-aws-sdk-s3 = { default-features = false, version = "1.156.0" }
-aws-sdk-sts = { default-features = false, version = "1.123.0" }
+aws-sdk-kms = { default-features = false, version = "1.129.0" }
+aws-sdk-s3 = { default-features = false, version = "1.157.0" }
+aws-sdk-sts = { default-features = false, version = "1.124.0" }
 aws-smithy-async = { version = "1.3.1" }
 aws-smithy-http-client = { default-features = false, version = "1.5.2" }
 aws-smithy-runtime-api = { version = "1.19.2" }
@@ -356,7 +356,7 @@ winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.19" }
-zip = "9.1.3"
+zip = "9.2.1"
 zstd = "0.14.3"

 # Observability and Metrics
@@ -386,7 +386,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 rustfs-mimalloc = { version = "0.7.2" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
-hotpath = { version = "0.30.2", default-features = false }
+hotpath = { version = "0.30.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.49" }

@@ -427,7 +427,3 @@ strip = "none"
 # Pin upstream fixes until compatible crates.io releases are available.
 [patch.crates-io]
 zstd = { path = "crates/zstd-compat" }
-# Preserve RustFS OIDC flows while migrating the old elliptic-curve dependency chain.
-# Upstream: openidconnect-rs#244 and oauth2-rs#352.
-openidconnect = { git = "https://github.com/ramosbugs/openidconnect-rs", rev = "2f77ee73bc4554e1cb6279adb386005429fb4e41" }
-oauth2 = { git = "https://github.com/ramosbugs/oauth2-rs", rev = "89585f4a280afd7b285eaf1c9be0068874e03d22" }
@@ -3720,33 +3720,45 @@ fn is_decommission_copy_quorum_deferral(err: &Error) -> bool {
    data_movement::data_movement_stage_source_as::<DecommissionCopyQuorumFailure>(err).is_some()
 }

-/// The storage error of a version copy that kept timing out on a namespace
-/// lock through its in-place retries, kept typed so the entry failure policy
-/// defers it.
+/// A copy whose object namespace lock remained unavailable through its
+/// in-place retries, retained as a typed deferral.
 #[derive(Debug, thiserror::Error)]
 #[error("{0}")]
-struct DecommissionCopyLockTimeout(#[source] Error);
+struct DecommissionCopyLockUnavailable(#[source] Error);

-fn is_decommission_copy_lock_timeout_error(err: &Error) -> bool {
-    let err = data_movement::data_movement_stage_source(err).unwrap_or(err);
-    matches!(err, Error::Lock(rustfs_lock::LockError::Timeout { .. }))
+fn is_decommission_copy_lock_unavailable_error(err: &Error, bucket: &str, object: &str) -> bool {
+    let source = data_movement::data_movement_stage_source(err).unwrap_or(err);
+    match source {
+        Error::Lock(rustfs_lock::LockError::Timeout { .. }) => true,
+        Error::NamespaceLockQuorumUnavailable {
+            mode: "read",
+            bucket: lock_bucket,
+            object: lock_object,
+            ..
+        } => {
+            data_movement::data_movement_error_is_stage(err, "prepare_put_object")
+                && lock_bucket == bucket
+                && lock_object == object
+        }
+        _ => false,
+    }
 }

-/// A version copy that timed out acquiring a namespace lock leaves the source
+/// A version copy that could not acquire its namespace lock leaves the source
 /// in place; the next attempt copies the version again or finds an equivalent
 /// committed target. The lock is held by another mutation of the key or, when
 /// this node restarted mid-copy, by its previous incarnation until that lock's
 /// lease expires on the peers. Either clears by waiting, so the entry is
 /// deferred instead of being counted as a permanent per-object failure.
-fn decommission_copy_lock_timeout_deferral(stage: &str, bucket: &str, object: &str, err: Error) -> Error {
+fn decommission_copy_lock_unavailable_deferral(stage: &str, bucket: &str, object: &str, err: Error) -> Error {
    data_movement::data_movement_context_error(
-        format!("decommission {stage} timed out on a namespace lock for {bucket}/{object}: {err}"),
-        DecommissionCopyLockTimeout(err),
+        format!("decommission {stage} could not acquire a namespace lock for {bucket}/{object}: {err}"),
+        DecommissionCopyLockUnavailable(err),
    )
 }

-fn is_decommission_copy_lock_timeout_deferral(err: &Error) -> bool {
-    data_movement::data_movement_stage_source_as::<DecommissionCopyLockTimeout>(err).is_some()
+fn is_decommission_copy_lock_unavailable_deferral(err: &Error) -> bool {
+    data_movement::data_movement_stage_source_as::<DecommissionCopyLockUnavailable>(err).is_some()
 }

 /// The deferral of a version copy whose in-place retries ended on `error`, if
@@ -3756,8 +3768,8 @@ fn take_decommission_copy_deferral(stage: &str, bucket: &str, object: &str, erro
        return Some(decommission_copy_quorum_deferral(stage, bucket, object, err));
    }
    error
-        .take_if(|err| is_decommission_copy_lock_timeout_error(err))
-        .map(|err| decommission_copy_lock_timeout_deferral(stage, bucket, object, err))
+        .take_if(|err| is_decommission_copy_lock_unavailable_error(err, bucket, object))
+        .map(|err| decommission_copy_lock_unavailable_deferral(stage, bucket, object, err))
 }

 fn resolve_decommission_entry_reload_result(result: Result<()>, bucket: &str, object_name: &str) -> Result<()> {
@@ -5822,13 +5834,13 @@ enum DecommissionEntryOutcome {
 /// entry is done and must not be counted as a failure or escalated. A quorum
 /// failure of the source cleanup is deferred too: the copied source is still
 /// in place and the next attempt cleans it up again. So is a version copy that
-/// lost drive quorum or timed out on a namespace lock through its in-place
+/// lost drive quorum or could not acquire an object namespace lock through its in-place
 /// retries: the source still holds the version and the next attempt copies it
 /// again. Every other failure stays terminal.
 fn resolve_decommission_entry_failure(err: Error) -> Result<DecommissionEntryOutcome> {
    if is_decommission_source_cleanup_retryable_error(&err)
        || is_decommission_copy_quorum_deferral(&err)
-        || is_decommission_copy_lock_timeout_deferral(&err)
+        || is_decommission_copy_lock_unavailable_deferral(&err)
    {
        return Ok(DecommissionEntryOutcome::Deferred(err));
    }
@@ -28353,7 +28365,7 @@ mod tests {
            "one version must commit before the next version times out"
        );
        assert!(
-            matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
+            matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_unavailable_deferral(err)),
            "{outcome:?}"
        );
        assert_eq!(
@@ -28387,6 +28399,88 @@ mod tests {
        assert_eq!(actual, b"written after versioning was enabled");
    }

+    /// A target lookup that loses its read-lock quorum must retain the source
+    /// and resume an entry after an earlier version has already committed.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn decommission_defers_target_lookup_read_lock_quorum_and_resumes() {
+        use tokio::io::AsyncReadExt as _;
+
+        let fixture = NullVersionDecommission::new(NullVersionTop::NewerVersion).await;
+        let object = fixture.object;
+        let attempted = Arc::new(AtomicUsize::new(0));
+        let fault = DecommissionTestFaultGuard::install_with_error(
+            Arc::new({
+                let attempted = Arc::clone(&attempted);
+                move |stage, _, faulted, _, success| {
+                    stage == "before_target_commit"
+                        && faulted == object
+                        && success
+                        && attempted.fetch_add(1, Ordering::SeqCst) > 0
+                }
+            }),
+            Some(Arc::new(|_, bucket, object| {
+                data_movement::data_movement_stage_error_for_test(
+                    "decommission_object",
+                    "prepare_put_object",
+                    bucket,
+                    object,
+                    Error::NamespaceLockQuorumUnavailable {
+                        mode: "read",
+                        bucket: bucket.to_owned(),
+                        object: object.to_owned(),
+                        required: 1,
+                        achieved: 0,
+                    },
+                )
+            })),
+        );
+        let (outcome, _) = fixture.run_entry().await;
+        assert!(
+            matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_unavailable_deferral(err)),
+            "{outcome:?}"
+        );
+        fixture.assert_source_retained().await;
+        assert_eq!(
+            fixture
+                .target_versions()
+                .await
+                .expect("first version committed")
+                .versions
+                .len(),
+            1
+        );
+        assert_eq!(
+            fixture.store.pool_meta.read().await.pools[0]
+                .decommission
+                .as_ref()
+                .expect("active decommission")
+                .items_decommission_failed,
+            0
+        );
+
+        drop(fault);
+        let (outcome, _) = fixture.run_entry().await;
+        assert!(matches!(outcome, DecommissionEntryOutcome::Complete), "{outcome:?}");
+        fixture.assert_moved().await;
+        let mut reader = fixture.store.pools[1]
+            .get_object_reader(
+                fixture.bucket,
+                fixture.object,
+                None,
+                HeaderMap::new(),
+                &ObjectOptions {
+                    version_id: Some(fixture.seeded.versions[0].version_id.expect("newer version").to_string()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("the newer version must remain readable");
+        let mut actual = Vec::new();
+        reader.stream.read_to_end(&mut actual).await.expect("read the newer version");
+        assert_eq!(actual, b"written after versioning was enabled");
+    }
+
    /// A namespace lock still held on the target key, as by a copy whose node
    /// was killed mid-write, makes target selection fail with a typed
    /// object-lock timeout. A real held lock, not an injected error, must defer
@@ -28429,7 +28523,7 @@ mod tests {

            let (outcome, _) = fixture.run_entry().await;
            assert!(
-                matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
+                matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_unavailable_deferral(err)),
                "{outcome:?}"
            );
            assert_eq!(
@@ -28838,17 +28932,17 @@ mod tests {
            timeout(),
        );
        for err in [timeout(), prepare] {
-            assert!(is_decommission_copy_lock_timeout_error(&err), "{err}");
+            assert!(is_decommission_copy_lock_unavailable_error(&err, "bucket-a", "object-a"), "{err}");
            assert!(!is_decommission_copy_quorum_error(&err), "{err}");
            let mut error = Some(err);
            let deferred = take_decommission_copy_deferral("migrate_object", "bucket-a", "object-a", &mut error)
                .expect("a lock timeout copy must be deferred");
            assert!(error.is_none());
-            assert!(is_decommission_copy_lock_timeout_deferral(&deferred));
+            assert!(is_decommission_copy_lock_unavailable_deferral(&deferred));
            assert!(!is_decommission_copy_quorum_deferral(&deferred));
-            let failure = data_movement::data_movement_stage_source_as::<DecommissionCopyLockTimeout>(&deferred)
+            let failure = data_movement::data_movement_stage_source_as::<DecommissionCopyLockUnavailable>(&deferred)
                .expect("the deferred error must keep its lock-timeout origin");
-            assert!(is_decommission_copy_lock_timeout_error(&failure.0));
+            assert!(is_decommission_copy_lock_unavailable_error(&failure.0, "bucket-a", "object-a"));
            assert!(matches!(
                resolve_decommission_entry_failure(deferred),
                Ok(DecommissionEntryOutcome::Deferred(_))
@@ -28873,6 +28967,43 @@ mod tests {
        }
    }

+    #[test]
+    fn decommission_copy_defers_only_the_matching_object_read_lock_quorum() {
+        let quorum = |mode, bucket: &str, object: &str| Error::NamespaceLockQuorumUnavailable {
+            mode,
+            bucket: bucket.to_owned(),
+            object: object.to_owned(),
+            required: 1,
+            achieved: 0,
+        };
+        let wrap = |stage, err| {
+            data_movement::data_movement_stage_error_for_test("decommission_object", stage, "bucket-a", "object-a", err)
+        };
+        let mut error = Some(wrap("prepare_put_object", quorum("read", "bucket-a", "object-a")));
+        let deferred = take_decommission_copy_deferral("migrate_object", "bucket-a", "object-a", &mut error)
+            .expect("matching read-lock quorum must defer");
+        assert!(error.is_none());
+        assert!(matches!(
+            resolve_decommission_entry_failure(deferred),
+            Ok(DecommissionEntryOutcome::Deferred(_))
+        ));
+        for err in [
+            quorum("read", "bucket-a", "object-a"),
+            wrap("source_recheck", quorum("read", "bucket-a", "object-a")),
+            wrap("prepare_put_object", quorum("write", "bucket-a", "object-a")),
+            wrap(
+                "prepare_put_object",
+                quorum("exact_delete_capacity_reconciliation", "bucket-a", "object-a"),
+            ),
+            wrap("prepare_put_object", quorum("read", "other-bucket", "object-a")),
+            wrap("prepare_put_object", quorum("read", "bucket-a", "other-object")),
+        ] {
+            let mut error = Some(err);
+            assert!(take_decommission_copy_deferral("migrate_object", "bucket-a", "object-a", &mut error).is_none());
+            assert!(error.is_some(), "unrelated lock failures must stay terminal");
+        }
+    }
+
    #[test]
    fn decommission_target_capacity_error_accepts_wrapped_capacity_errors() {
        let disk_full = Error::other(format!("decommission_object: put_object failed for bucket/object: {}", Error::DiskFull));
@@ -605,6 +605,7 @@ fn resolve_data_movement_abort_result(
 #[derive(Debug)]
 struct DataMovementStageError {
    rendered: String,
+    stage: Option<String>,
    source: Box<dyn std::error::Error + Send + Sync>,
 }

@@ -626,6 +627,7 @@ where
 {
    Error::other(DataMovementStageError {
        rendered,
+        stage: None,
        source: Box::new(err),
    })
 }
@@ -634,7 +636,21 @@ fn data_movement_stage_error<E>(op_label: &str, stage: &str, bucket: &str, objec
 where
    E: std::error::Error + Send + Sync + 'static,
 {
-    data_movement_context_error(format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"), err)
+    Error::other(DataMovementStageError {
+        rendered: format!("{op_label}: {stage} failed for {bucket}/{object}: {err}"),
+        stage: Some(stage.to_owned()),
+        source: Box::new(err),
+    })
+}
+
+pub(crate) fn data_movement_error_is_stage(err: &Error, stage: &str) -> bool {
+    let Error::Io(io_err) = err else {
+        return false;
+    };
+    io_err
+        .get_ref()
+        .and_then(|source| source.downcast_ref::<DataMovementStageError>())
+        .is_some_and(|source| source.stage.as_deref() == Some(stage))
 }

 #[cfg(test)]
@@ -853,6 +853,8 @@ pub(crate) struct DiskHealthMetricEpoch {
    fallback_operation_metrics: StdRwLock<HashMap<&'static str, Arc<DiskOperationMetrics>>>,
    #[cfg(test)]
    multipart_cleanup_observer: std::sync::Mutex<Option<std::sync::Weak<ScopedMultipartCleanupCompletions>>>,
+    #[cfg(test)]
+    operation_timeout_override: std::sync::Mutex<Option<Duration>>,
    /// Caller API operations currently executing through the disk health wrapper.
    api_waiting: AtomicU32,
    /// Operations rejected because the disk was unavailable for the caller.
@@ -880,6 +882,8 @@ impl Default for DiskHealthMetricEpoch {
            fallback_operation_metrics: StdRwLock::new(HashMap::new()),
            #[cfg(test)]
            multipart_cleanup_observer: std::sync::Mutex::new(None),
+            #[cfg(test)]
+            operation_timeout_override: std::sync::Mutex::new(None),
            api_waiting: AtomicU32::new(0),
            total_errors_availability: AtomicU64::new(0),
            total_errors_timeout: AtomicU64::new(0),
@@ -1573,6 +1577,20 @@ impl Default for DiskHealthTracker {
    }
 }

+#[cfg(test)]
+#[derive(Debug)]
+pub(crate) struct ScopedDiskOperationTimeout {
+    metrics: Arc<DiskHealthMetricEpoch>,
+    previous: Option<Duration>,
+}
+
+#[cfg(test)]
+impl Drop for ScopedDiskOperationTimeout {
+    fn drop(&mut self) {
+        *self.metrics.operation_timeout_override.lock().expect("test timeout lock") = self.previous;
+    }
+}
+
 /// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
 /// This is similar to Go's xlStorageDiskIDCheck.
 #[derive(Debug, Clone)]
@@ -1914,6 +1932,23 @@ impl LocalDiskWrapper {
        self.health_check
    }

+    #[cfg(test)]
+    pub(crate) fn override_operation_timeout_for_test(&self, timeout: Duration) -> ScopedDiskOperationTimeout {
+        assert!(!timeout.is_zero(), "fault injection needs a positive timeout");
+        // The metrics epoch follows this drive across clones and reconnects.
+        // Its scoped override cannot shorten unrelated healthy drives' I/O.
+        let previous = self
+            .metrics
+            .operation_timeout_override
+            .lock()
+            .expect("test timeout lock")
+            .replace(timeout);
+        ScopedDiskOperationTimeout {
+            metrics: Arc::clone(&self.metrics),
+            previous,
+        }
+    }
+
    /// Enable health monitoring after disk creation.
    /// Used to defer health checks until after startup format loading completes.
    pub fn enable_health_check(&self) {
@@ -2638,6 +2673,13 @@ impl LocalDiskWrapper {
            }
            return result;
        }
+        #[cfg(test)]
+        let timeout_duration = self
+            .metrics
+            .operation_timeout_override
+            .lock()
+            .expect("test timeout lock")
+            .map_or(timeout_duration, |injected| timeout_duration.min(injected));
        // Execute the operation with timeout
        let result = tokio::time::timeout(timeout_duration, operation).await;

@@ -6044,6 +6086,67 @@ mod tests {
        assert!(health.offline_duration().is_none());
    }

+    #[tokio::test]
+    async fn scoped_operation_timeout_is_per_disk_and_preserves_caller_policy() {
+        let directories = [
+            tempfile::tempdir().expect("faulted drive"),
+            tempfile::tempdir().expect("healthy drive"),
+        ];
+        let mut wrappers = Vec::new();
+        for directory in &directories {
+            let endpoint = Endpoint::try_from(directory.path().to_str().expect("UTF-8 drive path")).expect("drive endpoint");
+            let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local drive"));
+            wrappers.push(LocalDiskWrapper::new(disk, false));
+        }
+        let reconnected = LocalDiskWrapper::new_with_reconnect_state(
+            Arc::clone(&wrappers[0].disk),
+            false,
+            Some(wrappers[0].health_tracker_epoch_for_reconnect()),
+        );
+        let scope = wrappers[0].override_operation_timeout_for_test(Duration::from_secs(1));
+        tokio::time::pause();
+        let operation = || async {
+            tokio::time::sleep(Duration::from_secs(2)).await;
+            Ok(())
+        };
+        let (faulted, healthy) = tokio::join!(
+            reconnected.track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_secs(3),
+                TimeoutHealthAction::IgnoreFailure,
+            ),
+            wrappers[1].track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_secs(3),
+                TimeoutHealthAction::IgnoreFailure,
+            ),
+        );
+        assert_eq!(faulted.expect_err("only the selected drive must time out"), DiskError::Timeout);
+        healthy.expect("healthy drive keeps its caller's deadline");
+        let shorter_started = tokio::time::Instant::now();
+        let shorter = wrappers[0]
+            .track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_millis(500),
+                TimeoutHealthAction::IgnoreFailure,
+            )
+            .await;
+        assert_eq!(shorter.expect_err("fault injection cannot extend a caller deadline"), DiskError::Timeout);
+        assert!(shorter_started.elapsed() < Duration::from_secs(1));
+        wrappers[0]
+            .track_disk_health_with_op("read_metadata", operation, Duration::ZERO)
+            .await
+            .expect("explicit unbounded physical ownership cannot acquire a test deadline");
+        drop(scope);
+        reconnected
+            .track_disk_health_with_op("read_metadata", operation, Duration::from_secs(3))
+            .await
+            .expect("dropping the scope restores reconnected callers' deadlines");
+    }
+
    #[tokio::test]
    async fn ignored_timeout_does_not_mark_drive_failure() {
        let dir = tempfile::tempdir().expect("temp dir should be created");
@@ -8033,6 +8033,14 @@ impl LocalDisk {
    fn io_get_object_open_path(&self, bucket: &str, key: &str) -> Result<PathBuf> {
        #[cfg(target_os = "linux")]
        {
+            if is_clean_relative_segment(bucket) && is_clean_relative_segment(key) {
+                let root = self.io_root();
+                let mut path = PathBuf::with_capacity(root.as_os_str().len() + bucket.len() + key.len() + 2);
+                path.push(root);
+                path.push(bucket);
+                path.push(key);
+                return Ok(path);
+            }
            let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
            check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
            Ok(path)
@@ -8047,6 +8055,11 @@ impl LocalDisk {
    fn io_get_bucket_and_object_open_path(&self, bucket: &str, key: &str) -> Result<(PathBuf, PathBuf)> {
        #[cfg(target_os = "linux")]
        {
+            if is_clean_relative_segment(bucket) && is_clean_relative_segment(key) {
+                let bucket_path = self.io_root().join(bucket);
+                let path = bucket_path.join(key);
+                return Ok((bucket_path, path));
+            }
            let (bucket_path, path) = build_local_disk_object_path(self.io_root(), bucket, key);
            let normalized = check_local_disk_object_path_components(self.io_root(), &bucket_path, &path)?;
            if matches!(normalized, std::borrow::Cow::Owned(_)) {
@@ -12027,8 +12040,7 @@ fn local_disk_object_path(root: &Path, bucket: &str, key: &str) -> Result<PathBu
 #[cfg(not(windows))]
 #[inline]
 fn is_clean_relative_segment(s: &str) -> bool {
-    let b = s.as_bytes();
-    !matches!(b.first(), None | Some(b'/' | b'.')) && !b.ends_with(b"/") && !b.windows(2).any(|w| matches!(w, b"//" | b"/."))
+    !s.starts_with('/') && rustfs_utils::path::unix_is_canonical_ascii_path(s.as_bytes())
 }

 fn build_local_disk_object_path(root: &Path, bucket: &str, key: &str) -> (PathBuf, PathBuf) {
@@ -12075,10 +12087,26 @@ fn build_local_disk_bucket_path(root: &Path, bucket: &str) -> PathBuf {
    bucket_path
 }

+#[inline]
+fn disk_path_starts_with_normalized(path: &Path, base: &Path) -> bool {
+    #[cfg(not(windows))]
+    return rustfs_utils::path::unix_normalized_path_starts_with(path, base);
+    #[cfg(windows)]
+    path.starts_with(base)
+}
+
+#[inline]
+fn disk_path_strip_prefix_normalized<'a>(path: &'a Path, base: &Path) -> Result<&'a Path> {
+    #[cfg(not(windows))]
+    return rustfs_utils::path::unix_normalized_path_strip_prefix(path, base).ok_or(DiskError::InvalidPath);
+    #[cfg(windows)]
+    path.strip_prefix(base).map_err(|_| DiskError::InvalidPath)
+}
+
 fn check_local_disk_valid_object_path(root: &Path, bucket_path: &Path, path: &Path) -> Result<()> {
    let bucket_path = normalize_path_cow(bucket_path);
    let path = normalize_path_cow(path);
-    if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
+    if !disk_path_starts_with_normalized(&bucket_path, root) || !disk_path_starts_with_normalized(&path, &bucket_path) {
        return Err(DiskError::InvalidPath);
    }

@@ -12087,7 +12115,7 @@ fn check_local_disk_valid_object_path(root: &Path, bucket_path: &Path, path: &Pa

 fn check_local_disk_valid_path(root: &Path, path: impl AsRef<Path>) -> Result<()> {
    let path = normalize_path_cow(path.as_ref());
-    if !path.starts_with(root) {
+    if !disk_path_starts_with_normalized(&path, root) {
        return Err(DiskError::InvalidPath);
    }

@@ -12108,7 +12136,7 @@ fn check_local_disk_object_path_components<'a>(
 ) -> Result<std::borrow::Cow<'a, Path>> {
    let bucket_path = normalize_path_cow(bucket_path);
    let path = normalize_path_cow(path);
-    if !bucket_path.starts_with(root) || !path.starts_with(&bucket_path) {
+    if !disk_path_starts_with_normalized(&bucket_path, root) || !disk_path_starts_with_normalized(&path, &bucket_path) {
        return Err(DiskError::InvalidPath);
    }

@@ -12118,7 +12146,7 @@ fn check_local_disk_object_path_components<'a>(
 #[cfg(target_os = "linux")]
 fn check_local_disk_valid_path_at(root: &Path, root_fd: &std::fs::File, path: impl AsRef<Path>) -> Result<()> {
    let path = normalize_path_cow(path.as_ref());
-    if !path.starts_with(root) {
+    if !disk_path_starts_with_normalized(&path, root) {
        return Err(DiskError::InvalidPath);
    }

@@ -12127,7 +12155,7 @@ fn check_local_disk_valid_path_at(root: &Path, root_fd: &std::fs::File, path: im

 #[cfg(target_os = "linux")]
 fn reject_local_disk_symlink_components_at(root: &Path, root_fd: &std::fs::File, path: &Path) -> Result<()> {
-    let relative = path.strip_prefix(root).map_err(|_| DiskError::InvalidPath)?;
+    let relative = disk_path_strip_prefix_normalized(path, root)?;
    match validate_existing_local_disk_prefix_at(root_fd, relative) {
        Ok(()) => Ok(()),
        Err(LocalDiskPathValidationAtError::Unsupported) => reject_local_disk_symlink_components(root, path),
@@ -12155,11 +12183,11 @@ fn validate_existing_local_disk_prefix_at(
        return Ok(());
    }

-    let mut candidate = relative.to_path_buf();
+    let mut candidate = relative;
    loop {
        match openat2(
            root_fd,
-            &candidate,
+            candidate,
            OFlags::PATH | OFlags::CLOEXEC,
            Mode::empty(),
            ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS,
@@ -12168,10 +12196,12 @@ fn validate_existing_local_disk_prefix_at(
            Err(Errno::NOSYS) => return Err(LocalDiskPathValidationAtError::Unsupported),
            Err(Errno::LOOP | Errno::XDEV) => return Err(LocalDiskPathValidationAtError::InvalidPath),
            Err(Errno::NOENT) => {
-                let Some(parent) = candidate.parent().filter(|parent| !parent.as_os_str().is_empty()) else {
+                let Some(parent) =
+                    rustfs_utils::path::unix_path_parent(candidate).filter(|parent| !parent.as_os_str().is_empty())
+                else {
                    return Ok(());
                };
-                candidate = parent.to_path_buf();
+                candidate = parent;
            }
            Err(err) => return Err(LocalDiskPathValidationAtError::Io(err.into())),
        }
@@ -12179,7 +12209,7 @@ fn validate_existing_local_disk_prefix_at(
 }

 fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()> {
-    let relative = path.strip_prefix(root).map_err(|_| DiskError::InvalidPath)?;
+    let relative = disk_path_strip_prefix_normalized(path, root)?;
    let mut current = root.to_path_buf();

    for component in relative.components() {
@@ -12203,12 +12233,11 @@ fn reject_local_disk_symlink_components(root: &Path, path: &Path) -> Result<()>
 fn normalize_path_cow(path: &Path) -> std::borrow::Cow<'_, Path> {
    // Windows prefixes retain the existing PathBuf reconstruction semantics.
    #[cfg(not(windows))]
-    if !path.as_os_str().as_encoded_bytes().contains(&b'.')
-        || !path
-            .components()
-            .any(|c| matches!(c, std::path::Component::ParentDir | std::path::Component::CurDir))
    {
-        return std::borrow::Cow::Borrowed(path);
+        let bytes = path.as_os_str().as_encoded_bytes();
+        if bytes != b"." && !bytes.starts_with(b"./") && !rustfs_utils::path::unix_path_has_dotdot_segment(bytes) {
+            return std::borrow::Cow::Borrowed(path);
+        }
    }
    std::borrow::Cow::Owned(normalize_path_components(path))
 }
@@ -4439,12 +4439,18 @@ fn prepare_destination_parent_with_retry(
    base_dir: &Path,
    publication_root: &PublicationRoot,
 ) -> io::Result<ExistingBaseDirectoryGuard> {
-    let prune_budget = parent_prune_budget(parent, base_dir);
+    let mut prune_budget = None;
    let mut attempt = 0;
    loop {
        match mkdir_all_below_existing_base_std(parent, base_dir, publication_root) {
            Ok(preparation) => return Ok(preparation),
-            Err(err) if should_retry_prepare(&err, attempt, prune_budget) => {
+            Err(err)
+                if should_retry_prepare(
+                    &err,
+                    attempt,
+                    *prune_budget.get_or_insert_with(|| parent_prune_budget(parent, base_dir)),
+                ) =>
+            {
                attempt += 1;
            }
            Err(err) => return Err(err),
@@ -4936,13 +4942,9 @@ impl PublicationRoot {
        #[cfg(target_os = "linux")]
        {
            use rustix::fs::{Mode, OFlags};
-            let parent = self.open_directory(
-                path.parent()
-                    .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file parent"))?,
-            )?;
-            let name = path
-                .file_name()
-                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name"))?;
+            let (parent_path, name) = rustfs_utils::path::unix_split_parent_and_file_name(path)
+                .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file parent or name"))?;
+            let parent = self.open_directory(parent_path)?;
            let file = open_linux_relative(
                &parent,
                Path::new(name),
@@ -4982,9 +4984,9 @@ impl PublicationRoot {

    #[cfg(target_os = "linux")]
    fn relative_path<'a>(&self, path: &'a Path) -> io::Result<&'a Path> {
-        path.strip_prefix(self.io_path.as_path())
-            .or_else(|_| path.strip_prefix(self.path.as_path()))
-            .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root"))
+        rustfs_utils::path::unix_normalized_path_strip_prefix(path, self.io_path.as_path())
+            .or_else(|| rustfs_utils::path::unix_normalized_path_strip_prefix(path, self.path.as_path()))
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "path must remain below its publication root"))
    }

    #[cfg(windows)]
@@ -4996,6 +4998,13 @@ impl PublicationRoot {
    }
 }

+#[cfg(target_os = "linux")]
+#[inline]
+fn linux_relative_path_has_invalid_component(relative: &Path) -> bool {
+    let bytes = relative.as_os_str().as_encoded_bytes();
+    bytes.starts_with(b"/") || rustfs_utils::path::unix_path_has_dotdot_segment(bytes)
+}
+
 /// Open the actual IO handle with the same containment policy as path validation.
 /// Older kernels retain fd-relative traversal; only ENOSYS enables the fallback.
 #[cfg(target_os = "linux")]
@@ -5006,10 +5015,7 @@ fn open_linux_relative(
    mode: rustix::fs::Mode,
 ) -> io::Result<std::fs::File> {
    use rustix::fs::{OFlags, ResolveFlags, openat2};
-    if relative
-        .components()
-        .any(|component| !matches!(component, Component::Normal(_) | Component::CurDir))
-    {
+    if linux_relative_path_has_invalid_component(relative) {
        return Err(io::Error::new(io::ErrorKind::InvalidInput, "invalid relative disk path"));
    }
    let relative = if relative.as_os_str().is_empty() {
@@ -5180,11 +5186,8 @@ impl LinuxCommitDirectories {

    fn file_name<'a>(&self, path: &'a Path, source: bool) -> io::Result<&'a std::ffi::OsStr> {
        let parent = if source { &self.source_path } else { &self.destination_path };
-        if path.parent() != Some(parent) {
-            return Err(io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard"));
-        }
-        path.file_name()
-            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "missing file name"))
+        rustfs_utils::path::unix_direct_child_file_name(path, parent)
+            .ok_or_else(|| io::Error::new(io::ErrorKind::InvalidInput, "file parent does not match its commit guard"))
    }

    fn validate(&self) -> io::Result<()> {
@@ -5196,8 +5199,8 @@ impl LinuxCommitDirectories {
            return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed"));
        }
        let destination = self.prepare_destination()?;
-        if self.source_path != self.destination_path
-            || (self.source.device, self.source.inode) != (destination.device, destination.inode)
+        if (self.source.device, self.source.inode) != (destination.device, destination.inode)
+            || self.source_path.as_os_str() != self.destination_path.as_os_str()
        {
            let current = self.root.open_directory(&self.destination_path)?.metadata()?;
            if (current.dev(), current.ino()) != (destination.device, destination.inode) {
@@ -5216,7 +5219,10 @@ impl LinuxCommitDirectories {
        loop {
            if let Some(prepared) = prepared {
                let metadata = rustix::fs::statat(&*self.source.file, source_name, rustix::fs::AtFlags::SYMLINK_NOFOLLOW)?;
-                if prepared.path != source || metadata.st_dev != prepared.device || metadata.st_ino != prepared.inode {
+                if (prepared.path.as_os_str() != source.as_os_str() && prepared.path != source)
+                    || metadata.st_dev != prepared.device
+                    || metadata.st_ino != prepared.inode
+                {
                    return Err(io::Error::new(
                        io::ErrorKind::InvalidData,
                        "rename source identity changed while publication was prepared",
@@ -6354,19 +6360,35 @@ fn windows_rename_source_is_allowed(attributes: u32, reparse_tag: u32) -> bool {
 }

 fn existing_base_directory_suffix<'a>(dir_path: &'a Path, base_dir: &Path) -> io::Result<&'a Path> {
-    let relative = dir_path
-        .strip_prefix(base_dir)
-        .map_err(|_| io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory"))?;
-    for component in relative.components() {
-        if !matches!(component, Component::Normal(_) | Component::CurDir) {
+    #[cfg(target_os = "linux")]
+    {
+        let relative = rustfs_utils::path::unix_normalized_path_strip_prefix(dir_path, base_dir).ok_or_else(|| {
+            io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory")
+        })?;
+        if linux_relative_path_has_invalid_component(relative) {
            return Err(io::Error::new(
                io::ErrorKind::InvalidInput,
                "rename destination contains an invalid path component",
            ));
        }
+        Ok(relative)
    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        let relative = dir_path.strip_prefix(base_dir).map_err(|_| {
+            io::Error::new(io::ErrorKind::InvalidInput, "rename destination must remain below its base directory")
+        })?;
+        for component in relative.components() {
+            if !matches!(component, Component::Normal(_) | Component::CurDir) {
+                return Err(io::Error::new(
+                    io::ErrorKind::InvalidInput,
+                    "rename destination contains an invalid path component",
+                ));
+            }
+        }

-    Ok(relative)
+        Ok(relative)
+    }
 }

 /// Missing state never downgrades surviving native control records to legacy.
@@ -429,6 +429,7 @@ impl WritePlan {
        let checksum = reader.content_hash().clone();
        let trailer = reader.get_trailer().cloned();

+        let delegate_inner_etag = self.compression.is_none() && reader.is_etag_reader();
        let encrypted = self.encryption.is_some();
        if let Some(algorithm) = self.compression {
            reader = HashReader::from_reader(
@@ -448,7 +449,8 @@ impl WritePlan {
                }
                WriteEncryptionMode::SinglepartObjectKey | WriteEncryptionMode::MultipartObjectKey { .. } => [0u8; 12],
            };
-            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?;
+            let protected_reader = rustfs_rio::EtagReader::new_encrypted(reader, encryption.key_bytes, etag_nonce)?
+                .with_inner_etag_delegation(delegate_inner_etag);
            reader = match encryption.mode {
                WriteEncryptionMode::SinglepartObjectKey => HashReader::from_reader(
                    #[cfg(feature = "rio-v2")]
@@ -318,6 +318,21 @@ fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool {
    }
 }

+fn is_rebalance_metadata_admission_retryable(err: &Error) -> bool {
+    match rebalance_error_source(err) {
+        Error::Lock(err) => is_rebalance_transient_lock_error(err),
+        // No metadata was written when lock admission failed. Each attempt
+        // must acquire its quorum again before loading or saving this run.
+        Error::NamespaceLockQuorumUnavailable {
+            mode: "write",
+            bucket,
+            object,
+            ..
+        } => bucket == crate::disk::RUSTFS_META_BUCKET && object == REBAL_META_NAME,
+        _ => false,
+    }
+}
+
 fn is_rebalance_transient_io_error(err: &std::io::Error) -> bool {
    if err.kind() == std::io::ErrorKind::TimedOut {
        return true;
@@ -385,9 +400,7 @@ where
        match result {
            Ok(value) => return Ok(value),
            Err(err) => {
-                if attempt.saturating_add(1) >= max_attempts.max(1)
-                    || !matches!(rebalance_error_source(&err), Error::Lock(lock_err) if is_rebalance_transient_lock_error(lock_err))
-                {
+                if attempt.saturating_add(1) >= max_attempts.max(1) || !is_rebalance_metadata_admission_retryable(&err) {
                    return Err(err);
                }
                #[cfg(test)]
@@ -719,6 +732,82 @@ mod error_source_tests {
        }
    }

+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_recovers_after_write_quorum_failure() {
+        let mut attempts = 0;
+        retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(if attempts == 1 {
+                resolve_rebalance_meta_save_result(
+                    Err(rebalance_meta_lock_error(
+                        rustfs_lock::LockError::QuorumNotReached {
+                            required: 1,
+                            achieved: 0,
+                        },
+                        "write",
+                    )),
+                    "stop_rebalance",
+                )
+            } else {
+                Ok(())
+            })
+        })
+        .await
+        .expect("a later quorum-protected admission must be allowed to succeed");
+        assert_eq!(attempts, 2);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_retains_persistent_write_quorum_failure() {
+        for max_attempts in [0, 1, 3] {
+            let mut attempts = 0;
+            let result = retry_rebalance_metadata_access(None, max_attempts, || {
+                attempts += 1;
+                std::future::ready(Err::<(), _>(rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )))
+            })
+            .await;
+            let error = result.expect_err("exhausted admission attempts must never acknowledge a write");
+            assert_eq!(attempts, max_attempts.max(1));
+            assert!(matches!(
+                rebalance_error_source(&error),
+                Error::NamespaceLockQuorumUnavailable {
+                    mode: "write",
+                    required: 1,
+                    achieved: 0,
+                    ..
+                }
+            ));
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_stops_after_admission_is_cancelled() {
+        let mut attempts = 0;
+        let result = retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(Err::<(), _>(if attempts == 1 {
+                rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )
+            } else {
+                Error::OperationCanceled
+            }))
+        })
+        .await;
+        assert_eq!(attempts, 2, "a stale or cancelled run cannot enter another admission attempt");
+        assert!(matches!(result, Err(Error::OperationCanceled)));
+    }
+
    #[tokio::test]
    async fn rebalance_metadata_retry_does_not_retry_permanent_or_untyped_errors() {
        for err in [
@@ -731,6 +820,20 @@ mod error_source_tests {
                required: 3,
                achieved: 2,
            },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: "data-bucket".to_string(),
+                object: REBAL_META_NAME.to_string(),
+                required: 1,
+                achieved: 0,
+            },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
+                object: "other-metadata.bin".to_string(),
+                required: 1,
+                achieved: 0,
+            },
            Error::other("stale rebalance run rejected: lock acquisition timed out"),
            Error::other("rebalance distributed run fence lost"),
            Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),
@@ -9038,6 +9038,13 @@ mod tests {
            "constructing an object lock must not clone each client Arc"
        );

+        // Check shared ownership before acquisition can spawn pending cleanup.
+        let read_lock = set
+            .new_ns_lock("bucket", "read-object")
+            .await
+            .expect("second namespace lock should be created");
+        assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2);
+
        let write_error = write_lock
            .get_write_lock(Duration::from_millis(500))
            .await
@@ -9052,11 +9059,6 @@ mod tests {
            ),
            "the shared client representation must preserve the exact write quorum result: {write_error}"
        );
-        let read_lock = set
-            .new_ns_lock("bucket", "read-object")
-            .await
-            .expect("second namespace lock should be created");
-        assert_eq!(Arc::strong_count(&set.shared_lockers), clients_before + 2);
        let read_guard = read_lock
            .get_read_lock(Duration::from_millis(500))
            .await
@@ -2432,7 +2432,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
            let multipart_part_size = data.size();
            let stream = mem::replace(
                &mut data.stream,
-                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
+                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, true)?,
            );

            let write_path = classify_multipart_part_write_path(multipart_part_size, fi.erasure.block_size);
@@ -10060,7 +10060,7 @@ mod tests {

        // Real disk callbacks and health probes must share wall-clock deadlines.
        // Advancing Tokio time can expire unrelated healthy-drive probes.
-        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, Some("2"))], async {
+        temp_env::async_with_vars([(rustfs_config::ENV_DRIVE_MAX_TIMEOUT_DURATION, None::<&str>)], async {
            let (dirs, store) = crate::bucket::metadata_sys::test_support::isolated_store_over_temp_disks().await;
            crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
            let bucket = "mpu-quorum-success-physical";
@@ -10087,6 +10087,8 @@ mod tests {
                local.observe_multipart_cleanup_for_test(&upload_path)
            }).collect::<Vec<_>>();
            let disk::Disk::Local(paused_local) = disks[3].as_ref() else { panic!("actual local physical tail") };
+            // Fault only the held leg; healthy disks keep their normal I/O budgets.
+            let _paused_timeout = paused_local.override_operation_timeout_for_test(Duration::from_secs(2));
            let io_target = paused_local.get_disk().get_object_path_for_io(bucket, object).expect("real IO target");
            let mutation_key = paused_local.get_disk().get_object_path(bucket, object).expect("logical native mutation key");
            let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
@@ -10243,7 +10245,15 @@ mod tests {
                let size = copy_info.get_actual_size().expect("actual source size");
                copy_info.put_object_reader = Some(PutObjReader::new(HashReader::from_stream(reader.stream, size, size, None, None, false).expect("actual Copy body stream")));
                tokio::time::timeout(Duration::from_secs(5), store.copy_object(bucket, object, bucket, copy_object, &mut copy_info, &ObjectOptions::default(), &ObjectOptions::default()))
-                    .await.expect("cross-key Copy cannot wait on the old successful logical owner").expect("actual CopyObject");
+                    .await.expect("cross-key Copy cannot wait on the old successful logical owner")
+                    .unwrap_or_else(|error| {
+                        let health = disks.iter().enumerate().map(|(slot, disk)| {
+                            let disk::Disk::Local(local) = disk.as_ref() else { panic!("actual local Copy backend") };
+                            let metrics = local.metrics_snapshot();
+                            (slot, disk.runtime_state(), metrics.total_errors_timeout, metrics.total_errors_availability, metrics.total_waiting, metrics.api_calls)
+                        }).collect::<Vec<_>>();
+                        panic!("actual CopyObject: {error:?}; drive health (slot, state, timeouts, unavailable, waiting, calls): {health:?}");
+                    });
                drop(copy_info);
                let mut copied = store.get_object_reader(bucket, copy_object, None, HeaderMap::new(), &ObjectOptions::default()).await.expect("actual copied body");
                let mut actual = Vec::new();
@@ -4534,7 +4534,7 @@ impl SetDisks {

            let stream = mem::replace(
                &mut data.stream,
-                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, false)?,
+                HashReader::from_stream(Cursor::new(Vec::new()), 0, 0, None, None, true)?,
            );

            let small_size_hint = if matches!(write_path, SmallWritePath::Inline | SmallWritePath::SingleBlockNonInline) {
@@ -19,6 +19,7 @@ use crate::set_disk::get_lock_acquire_timeout;
 use crate::storage_api_contracts::multipart::MultipartOperations as _;
 use crate::storage_api_contracts::object::ObjectOperations as _;
 use futures::{StreamExt, stream};
+use rustfs_utils::path::encode_dir_object_ref;
 use std::collections::HashSet;

 const MULTIPART_LIST_SET_CONCURRENCY: usize = 4;
@@ -1185,11 +1186,9 @@ impl ECStore {
        let oi = if require_all_pools {
            lookup_opts.metadata_cache_safe = false;
            lookup_opts.suppress_read_repair = true;
-            if let Err(err) = check_object_args(bucket, object) {
-                return Some(err);
-            }
-            let encoded_object = encode_dir_object(object);
-            self.get_latest_object_info_for_conditional_write(bucket, &encoded_object, &lookup_opts)
+            // Only completion uses this branch, after validating bucket and object.
+            let encoded_object = encode_dir_object_ref(object);
+            self.get_latest_object_info_for_conditional_write(bucket, encoded_object.as_ref(), &lookup_opts)
                .await
                .and_then(|(info, _)| {
                    lookup_opts.precondition_check(&info)?;
@@ -3775,6 +3775,74 @@ mod tests {
        assert!(corrupt_crc_entry.is_latest_delete_marker());
    }

+    #[test]
+    fn scanner_and_list_select_same_non_free_version_after_cleanup_record() {
+        let mut fm = FileMeta::load(&crate::test_data::create_realistic_object_xlmeta().expect("realistic xl.meta"))
+            .expect("load realistic xl.meta");
+        let object = fm.versions[0].into_fileinfo("bucket", "object", true).expect("decode object");
+        let mut free_meta_sys = HashMap::new();
+        rustfs_utils::http::insert_bytes(&mut free_meta_sys, rustfs_utils::http::SUFFIX_FREE_VERSION, Vec::new());
+        let free = FileMetaShallowVersion::try_from(FileMetaVersion {
+            version_type: VersionType::Delete,
+            delete_marker: Some(MetaDeleteMarker {
+                version_id: Some(Uuid::from_u128(42)),
+                mod_time: Some(OffsetDateTime::from_unix_timestamp(2_000_000_000).expect("free timestamp")),
+                meta_sys: free_meta_sys,
+            }),
+            write_version: 1,
+            ..Default::default()
+        })
+        .expect("encode cleanup record");
+        fm.versions.insert(0, free);
+        let entry = MetaCacheEntry {
+            name: "object".to_string(),
+            metadata: fm.marshal_msg().expect("encode metadata"),
+            cached: None,
+            reusable: false,
+        };
+        let scanned = fm
+            .get_file_info_versions("bucket", "object", false)
+            .expect("scanner version decode");
+        assert_eq!(scanned.free_versions.len(), 1);
+        assert_eq!(scanned.versions.len(), 1);
+        assert!(!scanned.versions[0].is_latest, "free record occupies the original first position");
+        assert!(!entry.clone().is_latest_delete_marker(), "cleanup record is not a visible delete marker");
+        let listed = entry.to_fileinfo("bucket").expect("LIST version decode");
+        assert_eq!(listed.version_id, object.version_id);
+        assert_eq!(listed.version_id, scanned.versions[0].version_id);
+        assert_eq!(listed.mod_time, scanned.versions[0].mod_time);
+        assert_eq!(listed.size, scanned.versions[0].size);
+        assert!(!listed.deleted);
+
+        let marker_id = Uuid::from_u128(43);
+        fm.add_version(FileInfo {
+            volume: "bucket".to_string(),
+            name: "object".to_string(),
+            version_id: Some(marker_id),
+            deleted: true,
+            mod_time: Some(OffsetDateTime::from_unix_timestamp(2_100_000_000).expect("marker timestamp")),
+            ..Default::default()
+        })
+        .expect("add current delete marker");
+        let mut marker_entry = MetaCacheEntry {
+            name: "object".to_string(),
+            metadata: fm.marshal_msg().expect("encode marker metadata"),
+            cached: None,
+            reusable: false,
+        };
+        let scanned = fm
+            .get_file_info_versions("bucket", "object", false)
+            .expect("scan marker stack");
+        assert_eq!(scanned.versions.len(), 2);
+        assert_eq!(scanned.versions[0].version_id, Some(marker_id));
+        assert!(scanned.versions[0].deleted);
+        assert!(!scanned.versions[1].deleted, "older object must remain hidden by the current marker");
+        assert!(marker_entry.is_latest_delete_marker());
+        let listed = marker_entry.to_fileinfo("bucket").expect("LIST marker version");
+        assert_eq!(listed.version_id, scanned.versions[0].version_id);
+        assert!(listed.deleted);
+    }
+
    #[test]
    fn file_info_versions_and_borrowed_slice_decode_match_cached_and_full_decoders() {
        let mut fm = FileMeta::load(&crate::test_data::create_realistic_object_xlmeta().expect("realistic xl.meta"))
@@ -16,9 +16,10 @@ use super::{
    transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics},
 };
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
-use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
+use openidconnect::core::{CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
 use openidconnect::{
-    AsyncHttpClient, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, ProviderMetadataWithLogout, TokenUrl,
+    AsyncHttpClient, ClaimsVerificationError, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
+    ProviderMetadataWithLogout, SignatureVerificationError, TokenUrl,
 };
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -212,6 +213,35 @@ impl DiscoveredProviderMetadata {
    }
 }

+pub(super) fn verify_id_token_claims<'a, N: NonceVerifier>(
+    token: &'a CoreIdToken,
+    verifier: &CoreIdTokenVerifier<'_>,
+    nonce_verifier: N,
+) -> Result<&'a CoreIdTokenClaims, ClaimsVerificationError> {
+    let expected_length = match token.signing_alg().map_err(ClaimsVerificationError::SignatureVerification)? {
+        CoreJwsSigningAlgorithm::HmacSha256 => 32,
+        CoreJwsSigningAlgorithm::HmacSha384 => 48,
+        CoreJwsSigningAlgorithm::HmacSha512 => 64,
+        _ => return token.claims(verifier, nonce_verifier),
+    };
+    // RUSTFS_COMPAT_TODO(openidconnect-hmac-signature-width): Reject widths that upstream converts to fixed-size arrays. Remove after upstream returns verification errors for arbitrary HMAC signature lengths.
+    // openidconnect 4.1.0 converts HMAC signatures to fixed-size arrays before verifying them.
+    let invalid_length = || {
+        ClaimsVerificationError::SignatureVerification(SignatureVerificationError::CryptoError(
+            "invalid HMAC signature length".to_owned(),
+        ))
+    };
+    let compact = token.to_string();
+    let (_, signature) = compact.rsplit_once('.').ok_or_else(invalid_length)?;
+    let signature = base64_simd::URL_SAFE_NO_PAD
+        .decode_to_vec(signature)
+        .map_err(|_| invalid_length())?;
+    if signature.len() != expected_length {
+        return Err(invalid_length());
+    }
+    token.claims(verifier, nonce_verifier)
+}
+
 // This adapter is used only for discovery/JWKS fetches, never token exchange.
 struct JwksAcceptClient<'a> {
    inner: &'a ReqwestHttpClient,
@@ -242,7 +242,7 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
    let now = time::OffsetDateTime::now_utc().unix_timestamp();
    let payload = serde_json::json!({
        "iss": config.config_url, "sub": "synthetic-subject", "aud": config.client_id,
-        "iat": now, "exp": now + 300,
+        "iat": now, "exp": now + 300, "nonce": "synthetic-nonce",
    });

    for (algorithm, name, signature_length) in [
@@ -275,22 +275,28 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
                .expect("parse configured provider metadata");
            let verifier = metadata.verifier(&config);
            assert!(
-                valid.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
+                verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
                "valid {name} token must pass for workload={workload}"
            );
+            assert!(verify_id_token_claims(&valid, &verifier, &Nonce::new("synthetic-nonce".into())).is_ok());
+            assert!(verify_id_token_claims(&valid, &verifier, &Nonce::new("wrong-nonce".into())).is_err());

            for length in [1, 0, signature_length - 1, signature_length, signature_length + 1, 128] {
                let signature = base64_simd::URL_SAFE_NO_PAD.encode_to_string(vec![0_u8; length]);
                let malformed: CoreIdToken = format!("{signing_input}.{signature}")
                    .parse()
-                    .expect("invalid signature length must still reach cryptographic verification");
+                    .expect("invalid signature length must remain parseable before verification");
                assert!(
-                    malformed.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(),
+                    verify_id_token_claims(&malformed, &verifier, |_: Option<&Nonce>| Ok(())).is_err(),
                    "{name} signature length {length} must return an error for workload={workload}"
                );
+                assert!(
+                    verify_id_token_claims(&malformed, &verifier, &Nonce::new("synthetic-nonce".into())).is_err(),
+                    "{name} signature length {length} must return an error with a nonce for workload={workload}"
+                );
            }
            assert!(
-                valid.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
+                verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
                "invalid inputs must not prevent later valid {name} verification"
            );
        }
@@ -16,7 +16,7 @@ use super::{
        OidcConfigQuery, OidcConfigSnapshot, OidcProviderConfig, OidcProviderSummary, OidcSiteReplicationSnapshot,
        SourcedOidcProviderConfig, load_oidc_config_snapshot,
    },
-    provider::{ProviderRuntime, discover_provider_for_recovery},
+    provider::{ProviderRuntime, discover_provider_for_recovery, verify_id_token_claims},
    state::{OidcAuthSession, OidcLogoutSession, OidcStateStore},
    transport::{OidcExtraRootCaProvider, ReqwestHttpClient, inspect_token_response_body, oidc_http_error_diagnostics},
 };
@@ -535,7 +535,7 @@ impl OidcSys {
        let verifier = client
            .id_token_verifier()
            .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-        let verified = id_token.claims(&verifier, &Nonce::new(session.nonce.clone()));
+        let verified = verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()));
        if let Err(e) = verified {
            let verification_error = e.to_string();
            warn!(
@@ -583,8 +583,7 @@ impl OidcSys {
            let verifier = client
                .id_token_verifier()
                .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-            id_token
-                .claims(&verifier, &Nonce::new(session.nonce.clone()))
+            verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()))
                .map_err(|retry_err| {
                    error!(
                        event = EVENT_OIDC_DIAGNOSTICS,
@@ -767,7 +766,7 @@ impl OidcSys {
            .metadata
            .verifier(&config)
            .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-        if let Err(e) = id_token.claims(&verifier, |_: Option<&Nonce>| Ok(())) {
+        if let Err(e) = verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(())) {
            state = self
                .provider_runtime
                .refresh_provider_state(&provider_id, &config)
@@ -780,8 +779,7 @@ impl OidcSys {
                .metadata
                .verifier(&config)
                .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-            id_token
-                .claims(&verifier, |_: Option<&Nonce>| Ok(()))
+            verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(()))
                .map_err(|retry_err| format!("ID token verification failed after JWKS refresh: {retry_err}"))?;
        }

@@ -1081,7 +1081,7 @@ impl ChecksumHasher for Sha512Hasher {
 /// MD5 hasher for the ADDITIONAL checksum (x-amz-checksum-md5). Separate from the
 /// legacy Content-MD5 / ETag machinery — this only serves the flexible-checksum path.
 pub struct Md5Hasher {
-    hasher: md5::Md5,
+    hasher: rustfs_utils::FastMd5,
 }

 impl Default for Md5Hasher {
@@ -1092,14 +1092,14 @@ impl Default for Md5Hasher {

 impl Md5Hasher {
    pub fn new() -> Self {
-        use md5::Digest as _;
-        Self { hasher: md5::Md5::new() }
+        Self {
+            hasher: rustfs_utils::FastMd5::new(),
+        }
    }
 }

 impl Write for Md5Hasher {
    fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
-        use md5::Digest as _;
        self.hasher.update(buf);
        Ok(buf.len())
    }
@@ -1111,13 +1111,11 @@ impl Write for Md5Hasher {

 impl ChecksumHasher for Md5Hasher {
    fn finalize(&mut self) -> Vec<u8> {
-        use md5::Digest as _;
-        self.hasher.clone().finalize().to_vec()
+        self.hasher.finalize_bytes().to_vec()
    }

    fn reset(&mut self) {
-        use md5::Digest as _;
-        self.hasher = md5::Md5::new();
+        self.hasher.reset();
    }
 }

@@ -15,8 +15,8 @@
 use crate::compress_index::{Index, TryGetIndex};
 use crate::{BadDigest, EtagResolvable, HashReaderDetector, HashReaderMut};
 use hmac::{Hmac, KeyInit, Mac};
-use md5::{Digest, Md5};
 use pin_project_lite::pin_project;
+use rustfs_utils::FastMd5;
 use sha2::Sha256;
 use std::pin::Pin;
 use std::task::{Context, Poll};
@@ -27,11 +27,12 @@ pin_project! {
    pub struct  EtagReader<R> {
        #[pin]
        pub inner: R,
-        pub md5: Md5,
+        pub md5: FastMd5,
        pub finished: bool,
        pub checksum: Option<String>,
        resolved_etag: Option<String>,
        encrypted_etag_mac: Option<Hmac<Sha256>>,
+        delegate_inner_etag: bool,
    }
 }

@@ -39,11 +40,12 @@ impl<R> EtagReader<R> {
    pub fn new(inner: R, checksum: Option<String>) -> Self {
        Self {
            inner,
-            md5: Md5::new(),
+            md5: FastMd5::new(),
            finished: false,
            checksum,
            resolved_etag: None,
            encrypted_etag_mac: None,
+            delegate_inner_etag: false,
        }
    }

@@ -58,6 +60,17 @@ impl<R> EtagReader<R> {
        Ok(reader)
    }

+    /// Reuse the inner reader's plaintext ETag when deriving an encrypted ETag,
+    /// avoiding a redundant second MD5 pass over the same plaintext stream.
+    pub fn with_inner_etag_delegation(mut self, delegate: bool) -> Self
+    where
+        R: EtagResolvable,
+    {
+        self.delegate_inner_etag =
+            delegate && self.encrypted_etag_mac.is_some() && self.checksum.is_none() && self.inner.is_etag_reader();
+        self
+    }
+
    /// Get the final md5 value (etag) as a hex string, only compute once.
    /// Can be called multiple times, always returns the same result after finished.
    pub fn get_etag(&mut self) -> String {
@@ -65,8 +78,8 @@ impl<R> EtagReader<R> {
            return etag.clone();
        }

-        let etag = self.md5.clone().finalize().to_vec();
-        let etag = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower);
+        let digest = self.md5.finalize_bytes();
+        let etag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
        self.resolved_etag = Some(etag.clone());
        etag
    }
@@ -87,30 +100,32 @@ where
        if let Poll::Ready(Ok(())) = &poll {
            let filled = &buf.filled()[orig_filled..];
            if !filled.is_empty() {
-                this.md5.update(filled);
+                if !*this.delegate_inner_etag {
+                    this.md5.update(filled);
+                }
            } else {
                // EOF
                *this.finished = true;
-                let etag = if let Some(etag) = this.resolved_etag.as_ref() {
-                    etag.clone()
-                } else {
-                    let etag = this.md5.clone().finalize().to_vec();
-                    let etag = hex_simd::encode_to_string(etag, hex_simd::AsciiCase::Lower);
-                    *this.resolved_etag = Some(etag.clone());
-                    etag
-                };
+                if let Some(checksum) = this.checksum {
+                    let etag = if let Some(etag) = this.resolved_etag.as_ref() {
+                        etag.clone()
+                    } else {
+                        let digest = this.md5.finalize_bytes();
+                        let etag = hex_simd::encode_to_string(digest, hex_simd::AsciiCase::Lower);
+                        *this.resolved_etag = Some(etag.clone());
+                        etag
+                    };

-                if let Some(checksum) = this.checksum
-                    && *checksum != etag
-                {
-                    error!("Checksum mismatch, expected={:?}, actual={:?}", checksum, etag);
-                    return Poll::Ready(Err(std::io::Error::new(
-                        std::io::ErrorKind::InvalidData,
-                        BadDigest {
-                            expected_md5: checksum.clone(),
-                            calculated_md5: etag,
-                        },
-                    )));
+                    if *checksum != etag {
+                        error!("Checksum mismatch, expected={:?}, actual={:?}", checksum, etag);
+                        return Poll::Ready(Err(std::io::Error::new(
+                            std::io::ErrorKind::InvalidData,
+                            BadDigest {
+                                expected_md5: checksum.clone(),
+                                calculated_md5: etag,
+                            },
+                        )));
+                    }
                }
            }
        }
@@ -118,16 +133,24 @@ where
    }
 }

-impl<R> EtagResolvable for EtagReader<R> {
+impl<R> EtagResolvable for EtagReader<R>
+where
+    R: EtagResolvable,
+{
    fn is_etag_reader(&self) -> bool {
-        true
+        !self.delegate_inner_etag || self.inner.is_etag_reader()
    }
    fn try_resolve_etag(&mut self) -> Option<String> {
        if let Some(mut mac) = self.encrypted_etag_mac.clone() {
            if !self.finished {
                return None;
            }
-            mac.update(self.get_etag().as_bytes());
+            let plaintext_etag = if self.delegate_inner_etag {
+                self.inner.try_resolve_etag()?
+            } else {
+                self.get_etag()
+            };
+            mac.update(plaintext_etag.as_bytes());
            let protected = mac.finalize().into_bytes();
            return Some(hex_simd::encode_to_string(&protected[..16], hex_simd::AsciiCase::Lower));
        }
@@ -167,6 +190,7 @@ where
 #[cfg(test)]
 mod tests {
    use super::*;
+    use md5::{Digest, Md5};
    use rand::RngExt;
    use std::io::Cursor;
    use tokio::io::{AsyncReadExt, BufReader};
@@ -340,4 +364,54 @@ mod tests {
        assert_eq!(digest.expected_md5, wrong_checksum);
        assert_eq!(digest.calculated_md5, calculated_md5);
    }
+
+    #[tokio::test]
+    async fn encrypted_etag_reader_with_inner_delegation_matches_standalone() {
+        let data = b"delegated encrypted etag payload across multiple 64-byte MD5 blocks for verification";
+        let key = [0x42; 32];
+        let nonce = [0x24; 12];
+
+        let mut standalone =
+            EtagReader::new_encrypted(Cursor::new(data.as_slice()), key, nonce).expect("standalone encrypted EtagReader");
+        let mut out_standalone = Vec::new();
+        standalone.read_to_end(&mut out_standalone).await.expect("read standalone");
+        let expected_tag = standalone.try_resolve_etag().expect("standalone etag");
+
+        let inner =
+            crate::HashReader::from_stream(Cursor::new(data.as_slice()), data.len() as i64, data.len() as i64, None, None, false)
+                .expect("inner HashReader");
+        assert!(inner.is_etag_reader(), "HashReader with diskable_md5=false must report is_etag_reader");
+        let mut delegated = EtagReader::new_encrypted(inner, key, nonce)
+            .expect("delegated encrypted EtagReader")
+            .with_inner_etag_delegation(true);
+        let mut out_delegated = Vec::new();
+        delegated.read_to_end(&mut out_delegated).await.expect("read delegated");
+        assert_eq!(out_delegated, data);
+        assert_eq!(
+            delegated.try_resolve_etag().as_deref(),
+            Some(expected_tag.as_str()),
+            "delegated encrypted EtagReader must produce identical HMAC ETag without re-hashing plaintext"
+        );
+
+        let truncated = crate::LimitReader::new(
+            crate::HashReader::from_stream(Cursor::new(data.as_slice()), data.len() as i64, data.len() as i64, None, None, false)
+                .expect("inner HashReader for limit"),
+            16,
+        );
+        assert!(
+            !truncated.is_etag_reader(),
+            "truncating LimitReader must not advertise inner ETag delegation"
+        );
+
+        let mut non_etag_inner = EtagReader::new_encrypted(Cursor::new(data.as_slice()), key, nonce)
+            .expect("encrypted EtagReader over Cursor")
+            .with_inner_etag_delegation(true);
+        let mut out_cursor = Vec::new();
+        non_etag_inner.read_to_end(&mut out_cursor).await.expect("read cursor");
+        assert_eq!(
+            non_etag_inner.try_resolve_etag().as_deref(),
+            Some(expected_tag.as_str()),
+            "delegation request on non-EtagResolvable inner must keep outer MD5 hashing active"
+        );
+    }
 }
@@ -630,6 +630,10 @@ impl AsyncRead for HashReader {
 }

 impl EtagResolvable for HashReader {
+    fn is_etag_reader(&self) -> bool {
+        !self.diskable_md5 && self.inner.is_etag_reader()
+    }
+
    fn try_resolve_etag(&mut self) -> Option<String> {
        if self.diskable_md5 {
            return None;
@@ -201,10 +201,31 @@ where
    boxed_reader(WarpReader::new(reader))
 }

+impl<T> EtagResolvable for std::io::Cursor<T> {}
+
+impl EtagResolvable for &[u8] {}
+
+impl<R> EtagResolvable for tokio::io::BufReader<R>
+where
+    R: tokio::io::AsyncRead + EtagResolvable,
+{
+    fn is_etag_reader(&self) -> bool {
+        self.get_ref().is_etag_reader()
+    }
+
+    fn try_resolve_etag(&mut self) -> Option<String> {
+        self.get_mut().try_resolve_etag()
+    }
+}
+
 impl<T> EtagResolvable for Box<T>
 where
    T: EtagResolvable + ?Sized,
 {
+    fn is_etag_reader(&self) -> bool {
+        self.as_ref().is_etag_reader()
+    }
+
    fn try_resolve_etag(&mut self) -> Option<String> {
        self.as_mut().try_resolve_etag()
    }
@@ -29,6 +29,7 @@ workspace = true

 [features]
 default = []
+fips = ["rustls/fips"]
 hotpath = ["hotpath/hotpath", "hotpath/tokio", "rustfs-common/hotpath", "rustfs-config/hotpath"]
 hotpath-alloc = ["hotpath", "hotpath/hotpath-alloc", "rustfs-common/hotpath-alloc", "rustfs-config/hotpath-alloc"]
 hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-common/hotpath-cpu", "rustfs-config/hotpath-cpu"]
@@ -14,9 +14,300 @@

 use blake2::Blake2b512;
 use highway::{HighwayHash, HighwayHasher, Key};
-use md5::Md5;
 use serde::{Deserialize, Serialize};
 use sha2::{Digest, Sha256};
+use std::hint::black_box;
+
+const MD5_INITIAL_STATE: [u32; 4] = [0x6745_2301, 0xefcd_ab89, 0x98ba_dcfe, 0x1032_5476];
+
+const MD5_RC: [u32; 64] = [
+    0xd76a_a478,
+    0xe8c7_b756,
+    0x2420_70db,
+    0xc1bd_ceee,
+    0xf57c_0faf,
+    0x4787_c62a,
+    0xa830_4613,
+    0xfd46_9501,
+    0x6980_98d8,
+    0x8b44_f7af,
+    0xffff_5bb1,
+    0x895c_d7be,
+    0x6b90_1122,
+    0xfd98_7193,
+    0xa679_438e,
+    0x49b4_0821,
+    0xf61e_2562,
+    0xc040_b340,
+    0x265e_5a51,
+    0xe9b6_c7aa,
+    0xd62f_105d,
+    0x0244_1453,
+    0xd8a1_e681,
+    0xe7d3_fbc8,
+    0x21e1_cde6,
+    0xc337_07d6,
+    0xf4d5_0d87,
+    0x455a_14ed,
+    0xa9e3_e905,
+    0xfcef_a3f8,
+    0x676f_02d9,
+    0x8d2a_4c8a,
+    0xfffa_3942,
+    0x8771_f681,
+    0x6d9d_6122,
+    0xfde5_380c,
+    0xa4be_ea44,
+    0x4bde_cfa9,
+    0xf6bb_4b60,
+    0xbebf_bc70,
+    0x289b_7ec6,
+    0xeaa1_27fa,
+    0xd4ef_3085,
+    0x0488_1d05,
+    0xd9d4_d039,
+    0xe6db_99e5,
+    0x1fa2_7cf8,
+    0xc4ac_5665,
+    0xf429_2244,
+    0x432a_ff97,
+    0xab94_23a7,
+    0xfc93_a039,
+    0x655b_59c3,
+    0x8f0c_cc92,
+    0xffef_f47d,
+    0x8584_5dd1,
+    0x6fa8_7e4f,
+    0xfe2c_e6e0,
+    0xa301_4314,
+    0x4e08_11a1,
+    0xf753_7e82,
+    0xbd3a_f235,
+    0x2ad7_d2bb,
+    0xeb86_d391,
+];
+
+#[inline(always)]
+fn md5_op_f(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    // Keep `w + m + c` off the `x` critical dependency path from the previous round.
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    let f = z ^ (x & (y ^ z));
+    base.wrapping_add(f).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_g(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c)).wrapping_add(y & !z);
+    base.wrapping_add(x & z).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_h(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    base.wrapping_add(x ^ (y ^ z)).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_op_i(w: u32, x: u32, y: u32, z: u32, m: u32, c: u32, s: u32) -> u32 {
+    let base = black_box(w.wrapping_add(m).wrapping_add(c));
+    base.wrapping_add(y ^ (x | !z)).rotate_left(s).wrapping_add(x)
+}
+
+#[inline(always)]
+fn md5_word(words: &[[u8; 4]], idx: usize) -> u32 {
+    u32::from_le_bytes(words[idx])
+}
+
+#[inline]
+pub fn compress_md5_blocks(state: &mut [u32; 4], blocks: &[[u8; 64]]) {
+    let [mut a, mut b, mut c, mut d] = *state;
+    for block in blocks {
+        let words = block.as_chunks::<4>().0;
+        let aa = a;
+        let bb = b;
+        let cc = c;
+        let dd = d;
+
+        a = md5_op_f(a, b, c, d, md5_word(words, 0), MD5_RC[0], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 1), MD5_RC[1], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 2), MD5_RC[2], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 3), MD5_RC[3], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 4), MD5_RC[4], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 5), MD5_RC[5], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 6), MD5_RC[6], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 7), MD5_RC[7], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 8), MD5_RC[8], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 9), MD5_RC[9], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 10), MD5_RC[10], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 11), MD5_RC[11], 22);
+        a = md5_op_f(a, b, c, d, md5_word(words, 12), MD5_RC[12], 7);
+        d = md5_op_f(d, a, b, c, md5_word(words, 13), MD5_RC[13], 12);
+        c = md5_op_f(c, d, a, b, md5_word(words, 14), MD5_RC[14], 17);
+        b = md5_op_f(b, c, d, a, md5_word(words, 15), MD5_RC[15], 22);
+
+        a = md5_op_g(a, b, c, d, md5_word(words, 1), MD5_RC[16], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 6), MD5_RC[17], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 11), MD5_RC[18], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 0), MD5_RC[19], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 5), MD5_RC[20], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 10), MD5_RC[21], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 15), MD5_RC[22], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 4), MD5_RC[23], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 9), MD5_RC[24], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 14), MD5_RC[25], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 3), MD5_RC[26], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 8), MD5_RC[27], 20);
+        a = md5_op_g(a, b, c, d, md5_word(words, 13), MD5_RC[28], 5);
+        d = md5_op_g(d, a, b, c, md5_word(words, 2), MD5_RC[29], 9);
+        c = md5_op_g(c, d, a, b, md5_word(words, 7), MD5_RC[30], 14);
+        b = md5_op_g(b, c, d, a, md5_word(words, 12), MD5_RC[31], 20);
+
+        a = md5_op_h(a, b, c, d, md5_word(words, 5), MD5_RC[32], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 8), MD5_RC[33], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 11), MD5_RC[34], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 14), MD5_RC[35], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 1), MD5_RC[36], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 4), MD5_RC[37], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 7), MD5_RC[38], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 10), MD5_RC[39], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 13), MD5_RC[40], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 0), MD5_RC[41], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 3), MD5_RC[42], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 6), MD5_RC[43], 23);
+        a = md5_op_h(a, b, c, d, md5_word(words, 9), MD5_RC[44], 4);
+        d = md5_op_h(d, a, b, c, md5_word(words, 12), MD5_RC[45], 11);
+        c = md5_op_h(c, d, a, b, md5_word(words, 15), MD5_RC[46], 16);
+        b = md5_op_h(b, c, d, a, md5_word(words, 2), MD5_RC[47], 23);
+
+        a = md5_op_i(a, b, c, d, md5_word(words, 0), MD5_RC[48], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 7), MD5_RC[49], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 14), MD5_RC[50], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 5), MD5_RC[51], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 12), MD5_RC[52], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 3), MD5_RC[53], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 10), MD5_RC[54], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 1), MD5_RC[55], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 8), MD5_RC[56], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 15), MD5_RC[57], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 6), MD5_RC[58], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 13), MD5_RC[59], 21);
+        a = md5_op_i(a, b, c, d, md5_word(words, 4), MD5_RC[60], 6);
+        d = md5_op_i(d, a, b, c, md5_word(words, 11), MD5_RC[61], 10);
+        c = md5_op_i(c, d, a, b, md5_word(words, 2), MD5_RC[62], 15);
+        b = md5_op_i(b, c, d, a, md5_word(words, 9), MD5_RC[63], 21);
+
+        a = a.wrapping_add(aa);
+        b = b.wrapping_add(bb);
+        c = c.wrapping_add(cc);
+        d = d.wrapping_add(dd);
+    }
+    *state = [a, b, c, d];
+}
+
+#[inline]
+fn finalize_md5_state(mut state: [u32; 4], tail: &[u8], total_bytes: u64) -> [u8; 16] {
+    debug_assert!(tail.len() < 64);
+    let mut pad = [0u8; 128];
+    pad[..tail.len()].copy_from_slice(tail);
+    pad[tail.len()] = 0x80;
+    let bit_len = total_bytes.wrapping_mul(8).to_le_bytes();
+    if tail.len() < 56 {
+        pad[56..64].copy_from_slice(&bit_len);
+        compress_md5_blocks(&mut state, pad[..64].as_chunks::<64>().0);
+    } else {
+        pad[120..128].copy_from_slice(&bit_len);
+        compress_md5_blocks(&mut state, pad.as_chunks::<64>().0);
+    }
+    let mut out = [0u8; 16];
+    out[0..4].copy_from_slice(&state[0].to_le_bytes());
+    out[4..8].copy_from_slice(&state[1].to_le_bytes());
+    out[8..12].copy_from_slice(&state[2].to_le_bytes());
+    out[12..16].copy_from_slice(&state[3].to_le_bytes());
+    out
+}
+
+/// Streaming MD5 hasher using critical-path ILP reassociation and direct slice chunking.
+#[derive(Clone, Debug)]
+pub struct FastMd5 {
+    state: [u32; 4],
+    buf: [u8; 64],
+    buf_len: usize,
+    total_bytes: u64,
+}
+
+impl Default for FastMd5 {
+    #[inline]
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl FastMd5 {
+    #[inline]
+    pub const fn new() -> Self {
+        Self {
+            state: MD5_INITIAL_STATE,
+            buf: [0u8; 64],
+            buf_len: 0,
+            total_bytes: 0,
+        }
+    }
+
+    #[inline]
+    pub fn reset(&mut self) {
+        self.state = MD5_INITIAL_STATE;
+        self.buf_len = 0;
+        self.total_bytes = 0;
+    }
+
+    #[inline]
+    pub fn update(&mut self, mut data: &[u8]) {
+        if data.is_empty() {
+            return;
+        }
+        self.total_bytes = self.total_bytes.wrapping_add(data.len() as u64);
+        if self.buf_len > 0 {
+            let need = 64 - self.buf_len;
+            if data.len() < need {
+                self.buf[self.buf_len..self.buf_len + data.len()].copy_from_slice(data);
+                self.buf_len += data.len();
+                return;
+            }
+            self.buf[self.buf_len..64].copy_from_slice(&data[..need]);
+            compress_md5_blocks(&mut self.state, std::slice::from_ref(&self.buf));
+            self.buf_len = 0;
+            data = &data[need..];
+        }
+        let (blocks, remainder) = data.as_chunks::<64>();
+        if !blocks.is_empty() {
+            compress_md5_blocks(&mut self.state, blocks);
+        }
+        if !remainder.is_empty() {
+            self.buf[..remainder.len()].copy_from_slice(remainder);
+            self.buf_len = remainder.len();
+        }
+    }
+
+    #[inline]
+    pub fn finalize_bytes(&self) -> [u8; 16] {
+        finalize_md5_state(self.state, &self.buf[..self.buf_len], self.total_bytes)
+    }
+
+    #[inline]
+    pub fn finalize(self) -> [u8; 16] {
+        self.finalize_bytes()
+    }
+
+    #[inline]
+    pub fn digest(data: &[u8]) -> [u8; 16] {
+        let mut state = MD5_INITIAL_STATE;
+        let (blocks, remainder) = data.as_chunks::<64>();
+        if !blocks.is_empty() {
+            compress_md5_blocks(&mut state, blocks);
+        }
+        finalize_md5_state(state, remainder, data.len() as u64)
+    }
+}

 /// Magic HH-256 key: HH-256 hash of first 100 decimals of π as utf-8 with zero key.
 const MAGIC_HIGHWAY_HASH256_KEY: [u8; 32] = [
@@ -113,7 +404,7 @@ impl HashAlgorithm {
    #[inline]
    pub fn hash_encode(&self, data: &[u8]) -> impl AsRef<[u8]> {
        match self {
-            HashAlgorithm::Md5 => HashEncoded::Md5(Md5::digest(data).into()),
+            HashAlgorithm::Md5 => HashEncoded::Md5(FastMd5::digest(data)),
            HashAlgorithm::HighwayHash256 => {
                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
                hasher.append(data);
@@ -148,11 +439,11 @@ impl HashAlgorithm {
    {
        match self {
            HashAlgorithm::Md5 => {
-                let mut hasher = Md5::new();
+                let mut hasher = FastMd5::new();
                for slice in slices {
                    hasher.update(slice);
                }
-                HashEncoded::Md5(hasher.finalize().into())
+                HashEncoded::Md5(hasher.finalize_bytes())
            }
            HashAlgorithm::HighwayHash256 => {
                let mut hasher = HighwayHasher::new(MAGIC_HIGHWAY_HASH256_PARSED_KEY);
@@ -424,4 +715,24 @@ mod tests {
        let blake_hash2 = HashAlgorithm::BLAKE2b512.hash_encode(data2);
        assert_ne!(blake_hash1.as_ref(), blake_hash2.as_ref());
    }
+
+    #[test]
+    fn fast_md5_matches_reference_across_boundaries_and_chunkings() {
+        use md5::{Digest as _, Md5};
+
+        for len in (0..=260).chain([511, 512, 513, 1024, 4096, 8193]) {
+            let data = generate_compat_test_data(len);
+            let expected: [u8; 16] = Md5::digest(&data).into();
+            assert_eq!(FastMd5::digest(&data), expected, "FastMd5::digest mismatch at len={len}");
+
+            for step in [1usize, 3, 7, 16, 31, 55, 56, 63, 64, 65, 128] {
+                let mut fast = FastMd5::new();
+                for chunk in data.chunks(step) {
+                    fast.update(chunk);
+                }
+                assert_eq!(fast.finalize_bytes(), expected, "FastMd5 streaming mismatch at len={len}, step={step}");
+                assert_eq!(fast.finalize(), expected, "FastMd5 repeated finalize mismatch at len={len}, step={step}");
+            }
+        }
+    }
 }
@@ -694,6 +694,358 @@ pub fn trim_etag(etag: &str) -> String {
    etag.trim_matches('"').to_string()
 }

+/// Exact 64-bit SWAR byte equality mask for `b'.'` (`0x2e`) with zero cross-byte carry.
+/// Bit `7 + 8 * k` is set in the returned mask iff byte `k` of `w` equals `b'.'`.
+#[inline(always)]
+const fn swar_exact_dot_mask(w: u64) -> u64 {
+    let x = w ^ 0x2e2e_2e2e_2e2e_2e2e;
+    let low = (x & 0x7f7f_7f7f_7f7f_7f7f).wrapping_add(0x7f7f_7f7f_7f7f_7f7f);
+    !(low | x) & 0x8080_8080_8080_8080
+}
+
+/// Exact 64-bit SWAR mask for `b'.'` (`0x2e`) or `b'/'` (`0x2f`) across all byte values `0x00..=0xFF`.
+/// Since `0x2e` and `0x2f` differ only in bit 0, `(b & 0xfe) == 0x2e` holds iff `b` is `.` or `/`.
+#[inline(always)]
+const fn swar_exact_dot_or_slash_mask(w: u64) -> u64 {
+    swar_exact_dot_mask(w & 0xfefe_fefe_fefe_fefe)
+}
+
+#[inline(always)]
+const fn is_unix_dot_or_dotdot_at(path: &[u8], dot_idx: usize) -> bool {
+    if dot_idx > 0 && path[dot_idx - 1] != b'/' {
+        return false;
+    }
+    let next = dot_idx + 1;
+    if next == path.len() || path[next] == b'/' {
+        return true;
+    }
+    path[next] == b'.' && (next + 1 == path.len() || path[next + 1] == b'/')
+}
+
+#[inline(always)]
+const fn is_unix_dotdot_at(path: &[u8], dot_idx: usize) -> bool {
+    if dot_idx > 0 && path[dot_idx - 1] != b'/' {
+        return false;
+    }
+    let next = dot_idx + 1;
+    next < path.len() && path[next] == b'.' && (next + 1 == path.len() || path[next + 1] == b'/')
+}
+
+/// Returns `true` iff `path` (interpreted with Unix `/` separators) contains a `..` segment.
+/// Scans 8 bytes at a time using carry-free 64-bit SWAR so ordinary filenames containing dots
+/// (such as `xl.meta`, `part.1`, or `.rustfs.sys`) do not trigger component parsing.
+#[inline]
+pub const fn unix_path_has_dotdot_segment(path: &[u8]) -> bool {
+    let (chunks, rem) = path.as_chunks::<8>();
+    let mut c_idx = 0;
+    while c_idx < chunks.len() {
+        let mut mask = swar_exact_dot_mask(u64::from_le_bytes(chunks[c_idx]));
+        while mask != 0 {
+            let lane = (mask.trailing_zeros() >> 3) as usize;
+            if is_unix_dotdot_at(path, c_idx * 8 + lane) {
+                return true;
+            }
+            mask &= mask - 1;
+        }
+        c_idx += 1;
+    }
+    let base = chunks.len() * 8;
+    let mut r_idx = 0;
+    while r_idx < rem.len() {
+        if rem[r_idx] == b'.' && is_unix_dotdot_at(path, base + r_idx) {
+            return true;
+        }
+        r_idx += 1;
+    }
+    false
+}
+
+#[inline(always)]
+const fn is_invalid_adjacent_dot_slash_pair(path: &[u8], idx: usize) -> bool {
+    let prev = path[idx - 1];
+    let curr = path[idx];
+    if prev == b'/' && curr == b'/' {
+        return true;
+    }
+    if prev == b'/' && curr == b'.' {
+        return is_unix_dot_or_dotdot_at(path, idx);
+    }
+    if prev == b'.' && curr == b'/' {
+        return is_unix_dot_or_dotdot_at(path, idx - 1);
+    }
+    if prev == b'.' && curr == b'.' {
+        return is_unix_dotdot_at(path, idx - 1);
+    }
+    false
+}
+
+/// Returns `true` when `path` is a non-empty Unix path with no repeated slashes (`//`),
+/// no trailing slash (except `"/"`), and no `.` or `..` segments.
+#[inline]
+pub const fn unix_is_canonical_ascii_path(path: &[u8]) -> bool {
+    if path.is_empty() {
+        return false;
+    }
+    if path.len() == 1 {
+        return path[0] != b'.';
+    }
+    let first = path[0];
+    let last = path[path.len() - 1];
+    if last == b'/' || (first == b'.' && is_unix_dot_or_dotdot_at(path, 0)) {
+        return false;
+    }
+    let (chunks, rem) = path.as_chunks::<8>();
+    let mut c_idx = 0;
+    let mut prev_hi = 0u64;
+    while c_idx < chunks.len() {
+        let w = u64::from_le_bytes(chunks[c_idx]);
+        let m = swar_exact_dot_or_slash_mask(w);
+        let mut pair_mask = m & ((m << 8) | prev_hi);
+        while pair_mask != 0 {
+            let lane = (pair_mask.trailing_zeros() >> 3) as usize;
+            let idx = c_idx * 8 + lane;
+            if idx > 0 && is_invalid_adjacent_dot_slash_pair(path, idx) {
+                return false;
+            }
+            pair_mask &= pair_mask - 1;
+        }
+        prev_hi = m >> 56;
+        c_idx += 1;
+    }
+    let base = chunks.len() * 8;
+    let mut r_idx = 0;
+    while r_idx < rem.len() {
+        let b = rem[r_idx];
+        let idx = base + r_idx;
+        if (b == b'/' || b == b'.') && idx > 0 {
+            let prev = path[idx - 1];
+            if (prev == b'/' || prev == b'.') && is_invalid_adjacent_dot_slash_pair(path, idx) {
+                return false;
+            }
+        }
+        r_idx += 1;
+    }
+    true
+}
+
+#[inline(always)]
+const fn trim_trailing_slashes_and_curdir(mut bytes: &[u8]) -> &[u8] {
+    while bytes.len() > 1 {
+        if bytes[bytes.len() - 1] == b'/' {
+            bytes = bytes.split_at(bytes.len() - 1).0;
+        } else if bytes.len() >= 2 && bytes[bytes.len() - 2] == b'/' && bytes[bytes.len() - 1] == b'.' {
+            if bytes.len() == 2 {
+                bytes = b"/";
+                break;
+            }
+            bytes = bytes.split_at(bytes.len() - 2).0;
+        } else {
+            break;
+        }
+    }
+    bytes
+}
+
+#[inline(always)]
+const fn strip_leading_slashes_and_curdir(mut rem: &[u8]) -> &[u8] {
+    while !rem.is_empty() {
+        if rem[0] == b'/' {
+            rem = rem.split_at(1).1;
+        } else if rem.len() == 1 && rem[0] == b'.' {
+            rem = b"";
+        } else if rem.len() >= 2 && rem[0] == b'.' && rem[1] == b'/' {
+            rem = rem.split_at(2).1;
+        } else {
+            break;
+        }
+    }
+    rem
+}
+
+#[inline(always)]
+fn unix_bytes_to_os_str(bytes: &[u8]) -> Option<&std::ffi::OsStr> {
+    #[cfg(unix)]
+    {
+        use std::os::unix::ffi::OsStrExt;
+        Some(std::ffi::OsStr::from_bytes(bytes))
+    }
+    #[cfg(not(unix))]
+    {
+        std::str::from_utf8(bytes).ok().map(std::ffi::OsStr::new)
+    }
+}
+
+#[inline(always)]
+fn unix_bytes_to_path(bytes: &[u8]) -> Option<&Path> {
+    unix_bytes_to_os_str(bytes).map(Path::new)
+}
+
+/// Fast Unix `Path::starts_with` when both `path` and `base` have already been normalized
+/// (i.e. contain no `..` segments or leading `.`). Performs a single slice prefix comparison.
+#[inline]
+pub fn unix_normalized_path_starts_with(path: &Path, base: &Path) -> bool {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(base.as_os_str().as_encoded_bytes());
+    if b.is_empty() {
+        return true;
+    }
+    if b == b"/" {
+        return p.starts_with(b"/");
+    }
+    if p.starts_with(b) {
+        return p.len() == b.len() || p[b.len()] == b'/';
+    }
+    if b.contains(&b'.') || b.windows(2).any(|w| w == b"//") {
+        return path.starts_with(base);
+    }
+    false
+}
+
+/// Fast Unix `Path::strip_prefix` when both `path` and `base` have already been normalized
+/// (i.e. contain no `..` segments or leading `.`). Performs a single slice prefix comparison.
+#[inline]
+pub fn unix_normalized_path_strip_prefix<'a>(path: &'a Path, base: &Path) -> Option<&'a Path> {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(base.as_os_str().as_encoded_bytes());
+    if b.is_empty() {
+        return Some(path);
+    }
+    if b == b"/" {
+        if !p.starts_with(b"/") {
+            return None;
+        }
+        let mut rem = strip_leading_slashes_and_curdir(&p[1..]);
+        if rem.ends_with(b"/") || rem.ends_with(b"/.") {
+            rem = trim_trailing_slashes_and_curdir(rem);
+            if rem == b"/" {
+                rem = b"";
+            }
+        }
+        if let Some(res) = unix_bytes_to_path(rem) {
+            return Some(res);
+        }
+    } else if p.starts_with(b) {
+        if p.len() == b.len() {
+            return Some(Path::new(""));
+        }
+        if p[b.len()] == b'/' {
+            let mut rem = strip_leading_slashes_and_curdir(&p[b.len() + 1..]);
+            if rem.ends_with(b"/") || rem.ends_with(b"/.") {
+                rem = trim_trailing_slashes_and_curdir(rem);
+                if rem == b"/" {
+                    rem = b"";
+                }
+            }
+            if let Some(res) = unix_bytes_to_path(rem) {
+                return Some(res);
+            }
+        }
+        return None;
+    }
+    if b.contains(&b'.') || b.windows(2).any(|w| w == b"//") {
+        return path.strip_prefix(base).ok();
+    }
+    None
+}
+
+/// Fast Unix `Path::parent` that slices at the last `/` via SIMD `memrchr` without scanning
+/// earlier path components.
+#[inline]
+pub fn unix_path_parent(path: &Path) -> Option<&Path> {
+    let p = path.as_os_str().as_encoded_bytes();
+    if p.is_empty() || p == b"/" {
+        return None;
+    }
+    if let Some(idx) = p.iter().rposition(|&b| b == b'/') {
+        if idx + 1 < p.len() {
+            let tail = &p[idx + 1..];
+            if tail != b"." && tail != b".." {
+                if idx == 0 {
+                    return Some(Path::new("/"));
+                }
+                if !(idx == 1 && p[0] == b'.') {
+                    let parent_bytes = trim_trailing_slashes_and_curdir(&p[..idx]);
+                    if let Some(res) = unix_bytes_to_path(parent_bytes) {
+                        return Some(res);
+                    }
+                }
+            }
+        }
+    } else if p != b"." && p != b".." {
+        return Some(Path::new(""));
+    }
+    path.parent()
+}
+
+/// Splits a Unix `Path` into `(parent, file_name)` in a single reverse scan.
+#[inline]
+pub fn unix_split_parent_and_file_name(path: &Path) -> Option<(&Path, &std::ffi::OsStr)> {
+    let p = path.as_os_str().as_encoded_bytes();
+    if p.is_empty() || p == b"/" {
+        return None;
+    }
+    if let Some(idx) = p.iter().rposition(|&b| b == b'/') {
+        if idx + 1 < p.len() {
+            let tail = &p[idx + 1..];
+            if tail != b"." && tail != b".." {
+                if idx == 0 {
+                    if let Some(tail_os) = unix_bytes_to_os_str(tail) {
+                        return Some((Path::new("/"), tail_os));
+                    }
+                } else if !(idx == 1 && p[0] == b'.') {
+                    let parent_bytes = trim_trailing_slashes_and_curdir(&p[..idx]);
+                    if let (Some(parent_p), Some(tail_os)) = (unix_bytes_to_path(parent_bytes), unix_bytes_to_os_str(tail)) {
+                        return Some((parent_p, tail_os));
+                    }
+                }
+            }
+        }
+    } else if p != b"."
+        && p != b".."
+        && let Some(tail_os) = unix_bytes_to_os_str(p)
+    {
+        return Some((Path::new(""), tail_os));
+    }
+    Some((path.parent()?, path.file_name()?))
+}
+
+/// Verifies that `path.parent() == Some(expected_parent)` and returns `path.file_name()` in one
+/// slice prefix check.
+#[inline]
+pub fn unix_direct_child_file_name<'a>(path: &'a Path, expected_parent: &Path) -> Option<&'a std::ffi::OsStr> {
+    let p = path.as_os_str().as_encoded_bytes();
+    let b = trim_trailing_slashes_and_curdir(expected_parent.as_os_str().as_encoded_bytes());
+    if !b.is_empty() && b != b"." && !b.starts_with(b"./") && !unix_path_has_dotdot_segment(b) {
+        if b == b"/" {
+            if p.starts_with(b"/") {
+                let rem = strip_leading_slashes_and_curdir(&p[1..]);
+                if !rem.is_empty()
+                    && rem != b"."
+                    && rem != b".."
+                    && !rem.contains(&b'/')
+                    && let Some(res) = unix_bytes_to_os_str(rem)
+                {
+                    return Some(res);
+                }
+            }
+        } else if p.starts_with(b) && p.len() > b.len() + 1 && p[b.len()] == b'/' {
+            let rem = strip_leading_slashes_and_curdir(&p[b.len() + 1..]);
+            if !rem.is_empty()
+                && rem != b"."
+                && rem != b".."
+                && !rem.contains(&b'/')
+                && let Some(res) = unix_bytes_to_os_str(rem)
+            {
+                return Some(res);
+            }
+        }
+    }
+    if path.parent() != Some(expected_parent) {
+        return None;
+    }
+    path.file_name()
+}
+
 /// Returns `true` if `path` contains a `..` component (parent directory traversal).
 fn contains_parent_dir_component(path: &str) -> bool {
    path.split(['/', '\\']).any(|component| component == "..")
@@ -1211,4 +1563,110 @@ mod tests {
        assert!(normalize_extract_entry_key("file.txt", Some("../bad"), false).is_err());
        assert!(normalize_extract_entry_key("/absolute", None, false).is_err());
    }
+
+    #[test]
+    fn unix_swar_path_helpers_match_std_path_across_boundaries() {
+        for pad in 1..=20 {
+            let prefix = "a".repeat(pad);
+            let clean_cases = [
+                format!("/{prefix}/bucket/object/xl.meta"),
+                format!("/{prefix}/.rustfs.sys/tmp/uuid/part.1"),
+                format!("/{prefix}/dir..name/..hidden/file...ext"),
+                format!("{prefix}/relative/object.tar.gz"),
+            ];
+            for case in &clean_cases {
+                let bytes = case.as_bytes();
+                assert!(!unix_path_has_dotdot_segment(bytes), "clean case flagged dotdot: {case}");
+                assert!(unix_is_canonical_ascii_path(bytes), "canonical case rejected: {case}");
+            }
+
+            let non_canonical_slash_cases = [
+                format!("//{prefix}/bucket/object/xl.meta"),
+                format!("/{prefix}//bucket/object/xl.meta"),
+                format!("/{prefix}/bucket/object/"),
+            ];
+            for case in &non_canonical_slash_cases {
+                assert!(
+                    !unix_is_canonical_ascii_path(case.as_bytes()),
+                    "non-canonical slash case accepted: {case}"
+                );
+            }
+
+            let dot_cases = [
+                format!("/{prefix}/./xl.meta"),
+                format!("/{prefix}/bucket/."),
+                format!("./{prefix}/bucket"),
+                ".".to_string(),
+            ];
+            for case in &dot_cases {
+                let bytes = case.as_bytes();
+                assert!(!unix_path_has_dotdot_segment(bytes), "dot-only case flagged as dotdot: {case}");
+                assert!(!unix_is_canonical_ascii_path(bytes), "dot case accepted as canonical: {case}");
+            }
+
+            let dotdot_cases = [
+                format!("/{prefix}/../xl.meta"),
+                format!("/{prefix}/bucket/.."),
+                format!("../{prefix}/bucket"),
+                "..".to_string(),
+            ];
+            for case in &dotdot_cases {
+                let bytes = case.as_bytes();
+                assert!(unix_path_has_dotdot_segment(bytes), "dotdot case missed by dotdot: {case}");
+                assert!(!unix_is_canonical_ascii_path(bytes), "dotdot case accepted as canonical: {case}");
+            }
+        }
+
+        let base = Path::new("/tmp/rustfs-disk1");
+        let child = Path::new("/tmp/rustfs-disk1/bucket/prefix/xl.meta");
+        let sibling = Path::new("/tmp/rustfs-disk10/bucket/prefix/xl.meta");
+        assert!(unix_normalized_path_starts_with(child, base));
+        assert!(!unix_normalized_path_starts_with(sibling, base));
+        assert_eq!(unix_normalized_path_strip_prefix(child, base), Some(Path::new("bucket/prefix/xl.meta")));
+        assert_eq!(unix_normalized_path_strip_prefix(base, base), Some(Path::new("")));
+        assert_eq!(unix_normalized_path_strip_prefix(sibling, base), None);
+        assert_eq!(unix_path_parent(child), Some(Path::new("/tmp/rustfs-disk1/bucket/prefix")));
+        assert_eq!(unix_path_parent(Path::new("/")), None);
+        assert_eq!(unix_path_parent(Path::new("/rootchild")), Some(Path::new("/")));
+        assert_eq!(unix_path_parent(Path::new("relative")), Some(Path::new("")));
+        assert_eq!(
+            unix_split_parent_and_file_name(child),
+            Some((Path::new("/tmp/rustfs-disk1/bucket/prefix"), std::ffi::OsStr::new("xl.meta")))
+        );
+        assert_eq!(
+            unix_direct_child_file_name(child, Path::new("/tmp/rustfs-disk1/bucket/prefix")),
+            Some(std::ffi::OsStr::new("xl.meta"))
+        );
+        assert_eq!(unix_direct_child_file_name(child, base), None);
+
+        let proc_root = Path::new("/proc/self/fd/12/.");
+        let proc_bucket = Path::new("/proc/self/fd/12/./bucket");
+        let proc_obj = Path::new("/proc/self/fd/12/./bucket/prefix/xl.meta");
+        assert!(unix_normalized_path_starts_with(proc_bucket, proc_root));
+        assert!(unix_normalized_path_starts_with(proc_obj, proc_bucket));
+        assert!(unix_normalized_path_starts_with(proc_obj, proc_root));
+        assert_eq!(
+            unix_normalized_path_strip_prefix(proc_obj, proc_root),
+            Some(Path::new("bucket/prefix/xl.meta"))
+        );
+        assert_eq!(
+            unix_normalized_path_strip_prefix(proc_obj, proc_bucket),
+            Some(Path::new("prefix/xl.meta"))
+        );
+        assert_eq!(unix_path_parent(proc_obj), Some(Path::new("/proc/self/fd/12/./bucket/prefix")));
+        assert_eq!(
+            unix_direct_child_file_name(proc_obj, Path::new("/proc/self/fd/12/./bucket/prefix")),
+            Some(std::ffi::OsStr::new("xl.meta"))
+        );
+
+        assert!(unix_is_canonical_ascii_path("bucket/对象-ключ-é/xl.meta".as_bytes()));
+        assert!(unix_normalized_path_starts_with(
+            Path::new("/data/rustfs/bucket/dir/xl.meta"),
+            Path::new("/data/rustfs/bucket//dir")
+        ));
+        assert_eq!(
+            unix_normalized_path_strip_prefix(Path::new("/a/b/"), Path::new("/a")),
+            Some(Path::new("b"))
+        );
+    }
 }
@@ -1,6 +1,6 @@
 [package]
 name = "zstd"
-version = "0.13.3"
+version = "0.13.4"
 publish = false
 edition.workspace = true
 license.workspace = true
@@ -6,6 +6,6 @@ This private package satisfies the version constraints of Avro, Arrow IPC, Parqu

 The fixed implementation closes RUSTSEC-2026-0391, RUSTSEC-2026-0398, and RUSTSEC-2026-0403. A successful advisory scan alone is insufficient evidence: cargo-deny skips local path packages. The dependency-graph test therefore requires this adapter's sole runtime dependency to be the exact fixed registry implementation and rejects vulnerable registry codec versions. Security tests exercise abandoned stream contexts, malformed legacy frames, and seekable offsets past EOF. Consumer tests retain zstandard support in all four upstream formats.

-Run `cargo nextest run --locked -p zstd@0.13.3` for the focused tests. The seekable feature is enabled by a test-only dependency to cover the optional vulnerable API without enabling it in production.
+Run `cargo nextest run --locked -p zstd@0.13.4` for the focused tests. The seekable feature is enabled by a test-only dependency to cover the optional vulnerable API without enabling it in production.

 Remove this package and the root Cargo patch when all consumers accept the fixed zstd release line directly. The cleanup marker is registered in the architecture compatibility register.
@@ -57,7 +57,7 @@ fn legacy_constraints_resolve_only_to_the_fixed_registry_implementation() {
        let version = Version::parse(package["version"].as_str().expect("package version")).expect("valid package semver");
        if name == "zstd" && package.get("source").is_none() {
            adapters += 1;
-            assert_eq!(version, Version::new(0, 13, 3));
+            assert_eq!(version, Version::new(0, 13, 4));
            let runtime = package["dependencies"].as_array().expect("adapter dependency edges");
            assert!(runtime.iter().any(|value| value.as_str() == Some("zstd 0.14.3")));
        } else {
@@ -36,10 +36,6 @@ allow-git = [
    # Official s3s repository pinned for Iceberg SigV4 path encoding support.
    # owner: GatewayJ  review: 2026-10
    "https://github.com/s3s-project/s3s",
-    # Upstream OIDC/OAuth2 crypto migrations, pinned to reviewed PR commits.
-    # owner: rustfs-maintainers  review: 2026-10
-    "https://github.com/ramosbugs/openidconnect-rs",
-    "https://github.com/ramosbugs/oauth2-rs",
    # Completion ownership patch fixed to rustfs/uring#20.
    # owner: overtrue  review: 2026-10
    "https://github.com/rustfs/uring.git",
@@ -96,4 +92,8 @@ exceptions = [
    # transitively via `bzip2`. Not on a hot path.
    # owner: rustfs-maintainers  review: 2026-07
    { allow = ["bzip2-1.0.6"], crate = "libbz2-rs-sys" },
+    # `aws-lc-fips-sys` declares ISC, Apache-2.0, and the OpenSSL license for
+    # the FIPS-validated AWS-LC module when the optional `fips` feature is enabled.
+    # owner: rustfs-maintainers  review: 2026-10
+    { allow = ["OpenSSL"], crate = "aws-lc-fips-sys" },
 ]
@@ -46,6 +46,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 | [background-controller-contract.md](background-controller-contract.md) | adding a status snapshot or reconcile surface for a background service, or being tempted to fold several services into a generic controller |
 | [background-services-inventory.md](background-services-inventory.md) | you need one audited background service's desired source, current-status inputs, status surface, and declared side effects |
 | [scanner-usage-publication.md](scanner-usage-publication.md) | changing scanner data-usage cache publication, quota-visible usage snapshots, scanner cycle recovery, or the persisted scanner usage artifacts |
+| [scanner-prefix-statistics-contract.md](scanner-prefix-statistics-contract.md) | designing, implementing, or reviewing scanner-derived prefix statistics for the object browser or another authorized consumer (proposed; not runtime implemented) |
 | [scanner-dirty-journal-source-contract.md](scanner-dirty-journal-source-contract.md) | designing storage-verified provenance, completion receipts, repair responsibility or replay for the node-owned scanner dirty-producer journal (proposed; not runtime implemented) |
 | [scanner-usage-authority-decision.md](scanner-usage-authority-decision.md) | deciding whether quota admission depends on scanner data usage, removing scanner publication layers, or designing a scanner storage boundary |
 | [scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md) | changing what makes a scanner cycle `Complete`, `Superseded`, or converged, adding or re-classifying a dirty-usage producer, deciding whether a background writeback may invalidate a cycle, changing scanner worker concurrency or throttling, or activating scoped dirty acknowledgements |
@@ -11,6 +11,8 @@

 ## Open Items

+- `openidconnect-hmac-signature-width` ID token verification guard: openidconnect 4.1.0 converts decoded HMAC signatures to fixed-size arrays, which panic on unexpected lengths. Check HS256, HS384, and HS512 signature widths before delegating all cryptographic, issuer, audience, expiry, and nonce checks to the upstream verifier. Remove the guard after a supported upstream release returns verification errors for arbitrary HMAC signature lengths.
+
 - `zstd-013-fixed-backend` dependency constraint bridge: Avro, Arrow IPC, Parquet, and Kafka still require zstd 0.13. The local adapter re-exports the fixed registry zstd 0.14.3 implementation and its zstd-safe types. Remove the adapter and Cargo patch after every dependency accepts the fixed zstd line without the 0.13 bridge.

 - `8565` historical rebalance replacement recovery: older schema-seven intents can exhaust their retry budget before format repair when retained rebalance metadata describes removed pools. After validating blank targets, terminal historical topology, and ownership, grant one durable handoff and one rebuild attempt. Remove after supported direct upgrades no longer admit snapshots lacking replacement_preformat_handoff_retry_done and all retained affected intents have been resolved.
@@ -0,0 +1,84 @@
+# Scanner Prefix Statistics Contract
+
+**Use this when:** designing, implementing, or reviewing scanner-derived prefix statistics for the object browser or another authorized consumer.
+**Status:** proposed contract; scanner-derived current-object prefix statistics, their indexed persistence, and their query API are not implemented by this document.
+**Source of truth:** `crates/scanner/src/prefix_usage.rs`, `crates/data-usage/src/data_usage.rs`, `crates/scanner/src/scanner_io/io_disk.rs`, `crates/filemeta/src/metacache.rs`, `crates/ecstore/src/object_api/types.rs`, `crates/ecstore/src/store/list_objects.rs`, and `rustfs/src/storage/s3_api/bucket.rs`.
+
+## Purpose and current boundary
+
+A browser needs the total size of current visible objects below a displayed folder prefix and the latest modification time among those objects, including when a prefix contains very many objects. The intended result is an asynchronous scanner observation with explicit freshness and coverage, not an atomic S3 snapshot or an unconditional near-real-time guarantee. A consumer must be able to distinguish a completed observation from an incomplete, stale, or unavailable result. The requested operating scale is hundreds of millions of objects; a freshness target at that scale requires workload-specific measurement.
+
+The existing `bucket_prefix_usage` reads per-set usage caches and exposes subtree counters and cache update time. Its `PrefixUsageSummary` does not contain current-object modification time, and scanner size accounting can include historical versions. Cache path normalization and subtree compaction can lose the literal prefix identity or the child breakdown. Consequently that interface must not be presented as this contract's current-object statistics without a separate implementation and proof.
+
+The current S3 list path filters current delete markers and version-purge entries, and excludes free versions when selecting the visible metadata. Its response uses `ObjectInfo::get_actual_size_or_physical` and `mod_time`; for legacy data, a physical-size fallback can therefore appear where logical size is unknown. The scanner reads local metadata and has its own failure and completeness rules, so equivalent single-object decoding alone does not prove equivalence to a quorum-based listing.
+
+This contract does not change the existing scanner usage publication, quota, repair, or S3 API guarantees. Those guarantees remain owned by [scanner usage publication](scanner-usage-publication.md), [scanner convergence](scanner-per-bucket-convergence.md), and [placement and repair invariants](placement-repair-invariants.md).
+
+## Observable semantics
+
+A result describes the literal requested prefix and the current visible objects covered by that prefix. The modification time is the maximum modification time of those objects, not a property supplied by S3 for the prefix itself and not the time at which a scanner cache was written. Empty coverage has a zero byte count and no object modification time only when the entire requested scope was successfully observed and is known to be empty.
+
+Current-version selection and returned object size must be checked against the ordinary recursive `ListObjectsV2` projection, including noncurrent versions, a current delete marker, tier cleanup free versions, version-purge state, legacy compressed metadata, encryption, directory-marker objects, and absent or invalid timestamps. If the implementation cannot establish a field, it must mark that field unknown or the observation incomplete. A compatibility fallback used by S3 must be identified rather than represented as a proven logical byte count.
+
+Prefix keys are literal S3 names, not filesystem paths. Preserve separators, percent signs, Unicode, and distinct spellings without path cleaning, repeated decoding, or prefix normalization. The API must define its supported scope, including whether it accepts only the delimiter-derived folder prefixes returned by the browser or arbitrary byte prefixes. An unsupported, compacted-away, or absent index entry is unavailable unless independently certified as empty. A trailing-slash key such as `foo/` and a literal key such as `foo__XLDIR__` can map to the same physical encoding; `crates/scanner/src/scanner_folder/prefix_attribution.rs` explicitly treats this suffix as ambiguous. A physical name or suffix decoder alone is not trusted evidence of the original key: require a verified identity source that distinguishes the represented logical key, or report the affected scope as partial or unavailable without assigning the observation to either spelling.
+
+Counts and byte totals must use checked arithmetic. All byte totals must cross the API boundary as decimal strings, including small values, with their encoding specified by the API. Missing values, arithmetic overflow, corrupt metadata, and negative or indeterminate sizes must never become a plausible zero or saturated total.
+
+## Observation lifecycle
+
+The following states describe the externally visible observation, not new runtime types or implemented endpoints. Coverage and freshness remain separate fields even when the state is summarized for a consumer.
+
+| State | Admission and transition |
+|---|---|
+| Unavailable | No verified compatible observation covers the request, or the caller cannot be given one; never return a fabricated zero. |
+| Building | A bounded scan or rebuild is in progress; a separately verified last-known-good observation may remain available with its original coverage and age. |
+| Partial | A verified candidate covers only named ranges or sets, or contains unknown fields; missing, failed, and unsupported portions remain explicit. |
+| Complete | All required sets and requested ranges have compatible verified coverage, required fields, and quorum or equivalent authority for visible-version selection; this certifies an observation over a time interval, not an atomic snapshot. |
+| Stale | A retained verified observation is no longer within the declared freshness target or its scope was invalidated; retain its previous coverage and timestamps. |
+| Failed | A build, update, or verification failed; expose its reason without replacing a verified observation with an unverified candidate. |
+
+A successful build can move from building to complete only after the coverage and publication checks below succeed; otherwise it remains partial or unavailable. A failed build moves to failed while a previous observation, if any, remains separately labelled stale or retains the freshness supported by its evidence. Retry starts another bounded build; cancellation stops that build without changing the last verified observation. A bucket incarnation or layout mismatch invalidates the affected identity rather than relabelling its data for a new bucket or set.
+
+## Coverage and freshness
+
+A result carries its coverage and freshness independently. Internally, its evidence binds the bucket incarnation and requested literal prefix, contributing and expected set identities, each set's observation interval or oldest and newest represented observations, the completeness of its represented scope, and reasons for missing or unknown fields. The response exposes only the authorized, redacted coverage and freshness needed to interpret the result, as required below. A consumer may show a partial observed total only if it is labelled partial and its covered scope is explicit.
+
+A completed traversal proves only the coverage actually certified by that traversal; it does not create a point-in-time snapshot across sets or concurrent writes. A last-known-good shard or reused subtree retains its own oldest observation time. A publication timestamp, latest set update, or cache invalidation time must not be substituted for the time range of the data represented. Traversal completion on one local disk is not proof of an authoritative visible version. Before a scope is complete, require the same cross-disk quorum and version-selection evidence used by S3 listing, or separately verified equivalent evidence; stale replicas, competing delete markers, unavailable disks, and conflicting cross-pool migration observations must remain partial or unavailable until resolved. For each object, replica disagreement or missing authority cannot be settled by choosing whichever metadata the scanner saw first.
+
+Before any valid observation exists, the result is unavailable or building. A bounded traversal may publish an explicitly partial result when its covered ranges and unresolved gaps are known; after all required scope is certified it may publish a complete observation. A later mutation, expired freshness target, incomplete retry, topology change, unreadable set, or failed verification marks the affected result stale, partial, or unavailable as appropriate. A reader must never infer currentness solely from an earlier complete traversal.
+
+The result must expose the distinction between coverage and age even when a consumer requests a single summary. A complete but old result remains complete for its recorded observation and stale for current use; a recently updated partial result remains partial. The deployment must define freshness targets and report measured observation lag rather than claim an unverified deadline.
+
+Freshness certification requires evidence that relevant mutations and generation changes in the represented scope are covered through the claimed observation boundary. A dirty notification or local generation alone is not proof that every producer, peer, pending physical tail, or data-movement path was observed. A missing producer, unknown generation, replay gap, mixed-version peer, lost notification, or unverified journal interval must invalidate the freshness claim and leave the affected observation stale, partial, or unavailable until coverage is re-established. This document does not assert that a complete mutation journal or such certification already exists.
+
+## Persistence and recovery
+
+The index requires a separately versioned persistence contract. Reuse the scanner's admission, pacing, traversal, checkpoint, and coverage mechanisms only where their identities and guarantees match the indexed scope. Do not change the meaning of the existing `.usage-cache.bin` data or quota-visible totals to store the new result.
+
+Partition persistent index data so that reading or updating one scope does not require loading all prefixes in a bucket into memory. A manifest identifies its immutable bounded shards, literal key ranges, format version, bucket incarnation, topology and set identities, scanner generation, observation intervals, coverage, and integrity evidence. The supported prefix query and shard bounds must be explicit; pagination and continuation tokens must bind the requested scope and compatible generation. Validate untrusted cursors for authenticity, expiry, version, bounds, principal and authorized scope, bucket incarnation, prefix, and generation before use; reject altered, cross-scope, stale, or unsupported cursors without leaking indexed data.
+
+Publish a generation only after all referenced shard writes are durable and verified and the active writer's leadership, bucket incarnation, topology, and publication fences remain valid at the commit point. A concrete implementation must establish the atomic publication or conditional replacement primitive and its failure behavior; this document does not assert that a suitable primitive already exists. Readers validate the manifest and shard identity before use and retain a previous verified generation only with its original freshness and coverage labels.
+
+Interrupted work may resume only from a validated checkpoint bound to the same scope, generation, bucket incarnation, layout, and coverage evidence. Otherwise discard the candidate or rebuild it; do not combine incompatible generations or count both a replayed segment and its prior contribution. The release design must handle competing writers, crashes before and after publication, missing shards, failed reads, corruption, leader replacement, storage exhaustion, bucket deletion and recreation, resharding, rollback, and orphan cleanup.
+
+Unknown format versions and legacy records lacking the new proof are unavailable, not empty. During rolling upgrades, an older writer must not erase a newer index or certify new fields as complete; keep independent index ownership and version fencing, and provide a supported rollback path. A stopped or failed index must not block object writes, lifecycle, healing, or existing scanner usage publication.
+
+## Authorization and disclosure
+
+An index query must authenticate and authorize the caller for the actual bucket and literal prefix before reading or returning index data. Its decision must respect the same applicable `ListBucket` policy semantics as the corresponding recursive S3 listing, including prefix conditions, bucket policy, explicit denies, session restrictions, and current credentials. The operation must not use a broader background or administrator identity to disclose information unavailable to the caller.
+
+A parent-prefix aggregate may reveal objects in a restricted descendant. The implementation must establish that the caller can obtain the represented scope through the corresponding listing policy or return no aggregate for that scope. Pagination, cached responses, progress, error details, and last-known-good data are subject to the same authorization and must not expose other buckets, prefixes, or principals. Reauthorize each request; a cached allow decision must not survive a relevant policy or credential change without a proven invalidation mechanism. Ordinary users receive only the coverage and freshness information needed to interpret their authorized result; set identifiers, internal topology, physical paths, and detailed failure provenance require a separately authorized administrative surface. Redaction must also apply to cursors, progress, and error messages.
+
+The existing admin prefix-usage handler uses the shared admin gate before parsing the requested bucket. Its current authorization and response semantics must not be assumed to implement this contract; an integration must establish and test the actual resource and query-condition decision before using that route.
+
+## Resource and lifecycle boundaries
+
+Index construction and refresh are background work. Bound admission, queue depth, per-bucket and per-tenant concurrency, scan work per scheduling quantum, disk and network I/O, memory, shard size, retained generations, and disk usage. Cancellation and restart must stop or resume work without exposing unfinished data as complete. Refresh, compaction, cleanup, and a slow or unavailable set must not starve foreground object operations.
+
+A design must avoid accumulating the object population or an unbounded set of prefix rows in process memory, or issuing a full object-list scan for every displayed row. Query admission must enforce independent hard limits on requested page size, shards visited, bytes read, CPU time, memory, concurrent requests, and request duration; when a budget is exhausted, return a bounded partial or unavailable result with an appropriate continuation only when continuation is safe. It must account for key depth and write amplification, large single prefixes, many sibling prefixes, pagination, topology changes, and backpressure. Resource exhaustion must produce a visible degraded or unavailable state and retain object-store correctness; it must not cause silent data loss or an unbounded retry loop.
+
+## Required evidence
+
+Before the indexed result is enabled, tests must compare its current-object projection with S3 listings for ordinary, versioned, deleted, transitioned, legacy, corrupt, and unusual literal keys. Integration tests must cover scope-limited allow and deny policies, session changes, pagination and hostile cursors, internal-topology redaction, multiple sets, incomplete or corrupt shards, old formats, mixed versions, deletion and recreation, leader changes, cancellation, retries, and concurrent object mutations. Differential and fault tests must cover `foo/` versus literal `foo__XLDIR__`, stale replicas, competing delete markers, unavailable quorum, cross-pool migration conflicts, and gaps in mutation or generation evidence; no such case may silently certify a guessed key or version.
+
+Recovery tests must inject failure around shard writes, manifest publication, checkpoint replay, and cleanup and prove that readers never certify missing or mismatched data. Performance validation must measure scan completion and observation lag, memory and persistent-space growth, I/O load, and foreground workload impact under representative large populations and concurrent writes. Claims for the requested hundreds-of-millions scale and any near-real-time target require measured evidence at the stated scale and workload; a unit test, small successful scan, or unsupported extrapolation does not establish them. Larger scale claims require their own target and evidence and are not implied by this contract.
@@ -1,29 +1,29 @@
 # Cryptographic compliance positioning

-**Use this when:** writing README, CHANGELOG, release notes, marketing, RFP, or security-questionnaire text that touches FIPS, or reasoning about the `rustfs-crypto` `fips` feature and algorithm deprecation.
-**Source of truth:** `scripts/check_fips_wording.sh` (the enforced guard); `crates/crypto/Cargo.toml` (`fips` feature); `crates/crypto/src/encdec/id.rs` (`ID` algorithm bytes); `rustfs/src/startup_runtime_hooks.rs` (`install_default_crypto_provider`).
+**Use this when:** writing README, CHANGELOG, release notes, marketing, RFP, or security-questionnaire text that touches FIPS, or reasoning about the `rustfs` / `rustfs-tls-runtime` / `rustfs-crypto` `fips` features and algorithm deprecation.
+**Source of truth:** `scripts/check_fips_wording.sh` (the enforced guard); `rustfs/Cargo.toml` and `crates/tls-runtime/Cargo.toml` (`fips` feature); `crates/crypto/Cargo.toml` (`fips` feature); `crates/crypto/src/encdec/id.rs` (`ID` algorithm bytes); `rustfs/src/startup_runtime_hooks.rs` (`install_default_crypto_provider`).

 This document records where RustFS stands on cryptographic module validation and what may and may not be said about it, so the question is answered once, from the code. For where master key material lives per backend, see [KMS backend security properties](kms-backend-security.md).

 ## Status: not FIPS 140-3 validated

-**RustFS is not FIPS 140-3 (or 140-2) validated, and no component it links runs as a validated cryptographic module.** There is no CMVP certificate covering RustFS or the libraries it uses in the shipped configuration. This is a deliberate position and not a statement about algorithm strength: the algorithms in use are standard, well-reviewed AEADs. Validation is a property of a specific module build, its documented boundary, and a certificate — none of which RustFS has or pursues.
+**RustFS is not FIPS 140-3 (or 140-2) validated as a product, and in the default build no component it links runs as a validated cryptographic module.** There is no CMVP certificate covering RustFS itself, or the libraries it uses in the default shipped configuration. An opt-in `--features fips` build flag on `rustfs` switches the `rustls` / `aws-lc-rs` provider to `aws-lc-fips-sys` for TLS, X.509, and JWT operations, while object data-path and local KMS operations remain on unvalidated RustCrypto software implementations. Validation is a property of a specific module build, its documented boundary, and a certificate — none of which RustFS as a whole has or pursues.

 ### What the process actually links

 | Layer | Where | Implementation | Primitives | Validated module |
 | --- | --- | --- | --- | --- |
-| TLS (S3 server, internode, outbound clients) | Process-wide default provider installed by `install_default_crypto_provider` in `rustfs/src/startup_runtime_hooks.rs` | `rustls` with the `aws-lc-rs` provider | TLS 1.2/1.3 suites, `prefer-post-quantum` hybrid key exchange | No — the ordinary `aws-lc-rs` build, not the `aws-lc-fips-sys`-backed FIPS variant |
+| TLS (S3 server, internode, outbound clients) | Process-wide default provider installed by `install_default_crypto_provider` in `rustfs/src/startup_runtime_hooks.rs` | `rustls` with the `aws-lc-rs` provider (`default_provider` by default; `default_fips_provider` when built with `--features fips`) | TLS 1.2/1.3 suites, `prefer-post-quantum` hybrid key exchange (restricted to FIPS-approved suites and groups when `--features fips` is enabled) | No in the default build (`aws-lc-sys`); opt-in `--features fips` links `aws-lc-fips-sys` and asserts `provider.fips()` at startup |
 | Object data path AEAD (SSE) | `crates/kms/src/encryption/ciphers.rs`, `crates/rio/src/encrypt_reader.rs`, `crates/rio-v2/src/encrypt_reader.rs` | RustCrypto `aes-gcm`, `chacha20poly1305` | AES-256-GCM, ChaCha20-Poly1305 | No |
 | DEK wrapping | `crates/kms/src/encryption/dek.rs` | RustCrypto `aes-gcm` | AES-256-GCM | No |
 | Local KMS backend master key | `crates/kms/src/backends/local.rs` | RustCrypto `argon2`, `aes-gcm` | Argon2id KDF, AES-256-GCM | No |
-| Config and IAM blobs at rest | `crates/crypto/src/encdec/` (`rustfs-crypto`) | RustCrypto `pbkdf2`/`argon2`, `aes-gcm`, `chacha20poly1305`, `sha2` | see [the `fips` feature](#the-rustfs-crypto-fips-feature-what-it-actually-does) | No |
-| JWT signing and verification | `jsonwebtoken` with the `aws_lc_rs` feature (`crates/crypto`, `crates/iam`, `crates/policy`) | AWS-LC through `aws-lc-rs` | Non-FIPS build | No |
+| Config and IAM blobs at rest | `crates/crypto/src/encdec/` (`rustfs-crypto`) | RustCrypto `pbkdf2`/`argon2`, `aes-gcm`, `chacha20poly1305`, `sha2` | see [the `rustfs-crypto` `fips` feature](#the-rustfs-crypto-fips-feature-what-it-actually-does) | No |
+| JWT signing and verification | `jsonwebtoken` with the `aws_lc_rs` feature (`crates/crypto`, `crates/iam`, `crates/policy`) | AWS-LC through `aws-lc-rs` | Non-FIPS `aws-lc-sys` build by default; unified with `aws-lc-fips-sys` when `rustfs` is built with `--features fips` | No in the default build; links `aws-lc-fips-sys` only when built with `-p rustfs --features fips` |

 Two consequences are commonly assumed the other way:

- **AWS-LC being present does not imply FIPS.** `aws-lc-rs` has a FIPS variant; the workspace does not enable it anywhere.
- **The data path never touches AWS-LC.** Every byte of object plaintext is encrypted by RustCrypto software implementations. Swapping the TLS provider would not change that; see route 1 below.
+- **AWS-LC being present does not imply FIPS in the default build.** `aws-lc-rs` has a FIPS variant (`aws-lc-fips-sys`); default builds do not enable it. Building `rustfs` with `--features fips` enables `rustls/fips`, `tokio-rustls/fips`, and `rustfs-tls-runtime/fips`, which links `aws-lc-fips-sys` and installs `rustls::crypto::default_fips_provider()` (see [below](#the-rustfs-and-rustfs-tls-runtime-fips-feature)).
+- **The object data path never touches AWS-LC.** Every byte of object plaintext is encrypted by RustCrypto software implementations, even when `rustfs` is built with `--features fips`. Enabling the TLS FIPS provider does not change the object encryption data path; see route 1 below.

 ## Terminology red lines for external material

@@ -50,6 +50,15 @@ Boilerplate when the topic cannot be avoided:

 `scripts/check_fips_wording.sh` greps `README.md` and `CHANGELOG.md` for the banned phrases above, and separately rejects any wording under `crates/kms` that describes the Vault KV2 backend as wrapping key material through Vault's Transit engine (`KmsBackend::VaultKv2` stores RustFS-wrapped material in KV v2 and never calls Transit; use `VaultTransit` when cryptographic isolation is the requirement). This document is intentionally outside the scan: it needs the terminology to define the policy.

+## The `rustfs` and `rustfs-tls-runtime` `fips` feature
+
+`rustfs/Cargo.toml` (`fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips"]`) and `crates/tls-runtime/Cargo.toml` (`fips = ["rustls/fips"]`) define an opt-in Cargo feature (disabled by default) for builds that require the `aws-lc-fips-sys` cryptographic module on the `aws-lc-rs` / `rustls` path:
+
+- **What it enables:** Activates `rustls/fips` and `tokio-rustls/fips`, which pull in `aws-lc-rs/fips` (`aws-lc-fips-sys`) and `rustls-webpki/aws-lc-rs-fips`. Because Cargo unifies features across the dependency graph, other `aws-lc-rs` consumers in the binary (`jsonwebtoken`, `rcgen`, `google-cloud-auth`, `async-nats`) also link against `aws-lc-fips-sys`.
+- **Runtime provider installation:** In `rustfs/src/startup_runtime_hooks.rs`, `build_default_crypto_provider()` selects `rustls::crypto::default_fips_provider()` (instead of `rustls::crypto::aws_lc_rs::default_provider()`) and asserts `provider.fips()` on startup, restricting TLS negotiation to FIPS-approved cipher suites (e.g., `TLS_AES_256_GCM_SHA384`, `TLS_AES_128_GCM_SHA256`, `TLS_ECDHE_*_AES_*_GCM_*`) and key exchange groups (excluding `TLS_CHACHA20_POLY1305_SHA256` and `X25519`).
+- **Build requirements:** Compiling `aws-lc-fips-sys` requires CMake, Go, and Clang/`libclang` in the build environment (unlike the default `aws-lc-sys` build).
+- **What it does not change:** It does not alter the object data-path AEAD (`crates/kms`, `crates/rio`, `crates/rio-v2`) or `rustfs-crypto` config/IAM blob encryption, which continue to use pure-Rust `RustCrypto` crates.
+
 ## The `rustfs-crypto` `fips` feature: what it actually does

 `crates/crypto/Cargo.toml` declares `default = ["crypto", "fips"]`, so the feature is on in every normal build. Its entire effect is **which algorithm the write path selects**; the implementation is RustCrypto either way.
@@ -71,7 +80,7 @@ Selection sites are `crates/crypto/src/encdec/encrypt.rs` and `crates/crypto/src

 | Route | Scope | What it buys | Verdict |
 | --- | --- | --- | --- |
-| 1. Adopt the `aws-lc-rs` FIPS variant (`aws-lc-fips-sys`) | The TLS provider swap is the small part. Every AEAD call in the data path (`crates/kms/src/encryption/`, `crates/rio*/src/encrypt_reader.rs`, `crates/kms/src/backends/local.rs`, `crates/crypto/src/encdec/`) would be re-implemented against `aws-lc-rs` primitives; Argon2id (no FIPS status) and ChaCha20-Poly1305 would need read-only compatibility stories. Build needs CMake, a C toolchain, and Go, on a narrower target set — the static musl release already hits AWS-LC's `getentropy` abort on kernels older than 3.17, and the FIPS variant constrains the matrix strictly harder | The accurate claim becomes "cryptographic operations are performed by a FIPS 140-3 validated module", not "RustFS is FIPS validated"; a product-level claim additionally needs a documented boundary, approved-mode enforcement, self-tests, zeroization, and entropy documentation | Heavy; re-opens an open platform-support problem. Justified only by a named customer or regulatory commitment |
+| 1. Adopt the `aws-lc-rs` FIPS variant (`aws-lc-fips-sys`) across the data path | The TLS/X.509/JWT provider swap is wired via the opt-in `-p rustfs --features fips` build flag. Extending `aws-lc-fips-sys` to the data path would require re-implementing every AEAD/KDF call in `crates/kms/src/encryption/`, `crates/rio*/src/encrypt_reader.rs`, `crates/kms/src/backends/local.rs`, and `crates/crypto/src/encdec/` against `aws-lc-rs` primitives; Argon2id (no FIPS status) and ChaCha20-Poly1305 would need read-only compatibility stories. Build needs CMake, a C toolchain, Clang/`libclang`, and Go, on a narrower target set — the static musl release already hits AWS-LC's `getentropy` abort on kernels older than 3.17, and the FIPS variant constrains the matrix strictly harder | The accurate claim becomes "cryptographic operations are performed by a FIPS 140-3 validated module", not "RustFS is FIPS validated"; a product-level claim additionally needs a documented boundary, approved-mode enforcement, self-tests, zeroization, and entropy documentation | Data-path rewrite remains heavy and re-opens platform-support constraints; TLS/X.509/JWT is opt-in via `--features fips` |
 | 2. Let an externally validated KMS carry key operations | Mostly built: the `VaultTransit` backend never lets key-encryption key material leave Vault, so master key generation, wrapping, unwrapping, and rotation happen inside whatever module Vault's seal/HSM is validated against. Remaining work is configuration guidance and a supported-deployment statement | A defensible partial answer to "where do keys live and who validated that". The object data path stays RustCrypto and TLS stays non-FIPS AWS-LC: "key management operations are performed by an externally validated module; the object data path is not validated" | Nearest partial step, no code rewrite. Point customers here when the requirement is key custody rather than a certificate covering the storage layer |
 | 3. Make no validation claim (current default) | This document plus the grep guard | Nothing further | The current decision. FIPS 140-3 validation is not a roadmap target; adjacent items (PKCS#11, KMIP, BYOK, signing keys) are deferred for lack of demand and because HSM-dependent paths cannot be exercised in CI |

@@ -273,9 +273,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.156.0"
+version = "1.157.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1072c9aa20159b9b417115342d38c4c53108fc98e5359fcb86b7d4575747eb59"
+checksum = "6ea9794c741520d8baec9462bcd0ad12164d4f5f78d008a8da9d6a0b7222caa8"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1837,18 +1837,18 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b39e941010d86063ff233bf8325269bf68097fb24ba67000fbddc0a410e8056e"
+checksum = "d9632c1fd213e63c1b88d8f3e59543da4999b17f859cc1e1d67cf8f2207e8935"
 dependencies = [
 "hotpath-macros",
 ]

 [[package]]
 name = "hotpath-macros"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a24f5179ff680132003306354efd0ac8cde12e87b1a788719c836e91af35cb58"
+checksum = "5d92fdfec803932e5764ea93c201cf7c6c2d8cff1871da44f64f5094c9c3ef8a"

 [[package]]
 name = "http"
@@ -2418,9 +2418,9 @@ dependencies = [

 [[package]]
 name = "log"
-version = "0.4.34"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c"

 [[package]]
 name = "lru"
@@ -4134,8 +4134,7 @@ dependencies = [
 [[package]]
 name = "rustfs-uring"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7deda770b7981ee3bfaa202ecd18625fc794c962b8d58cd4f436952b7e66dcf"
+source = "git+https://github.com/rustfs/uring.git?rev=eb8fc160e71d804427de1e905f5c226d580f71bf#eb8fc160e71d804427de1e905f5c226d580f71bf"
 dependencies = [
 "io-uring",
 "libc",
@@ -58,6 +58,7 @@ required-features = ["swift"]

 [features]
 default = ["ftps", "webdav", "gcs"]
+fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips"]
 gcs = ["rustfs-ecstore/gcs", "dep:google-cloud-auth"]
 metrics-gpu = ["rustfs-obs/gpu"]
 ftps = ["rustfs-protocols/ftps"]
@@ -14,7 +14,10 @@

 use crate::license::license_status;
 use crate::startup_runtime_sources;
+#[cfg(not(feature = "fips"))]
 use rustls::crypto::aws_lc_rs::default_provider;
+#[cfg(feature = "fips")]
+use rustls::crypto::default_fips_provider as default_provider;
 use std::future::Future;
 use std::io::Result;
 use tracing::{debug, info, warn};
@@ -105,8 +108,18 @@ where
    shutdown();
 }

+fn build_default_crypto_provider() -> rustls::crypto::CryptoProvider {
+    let provider = default_provider();
+    #[cfg(feature = "fips")]
+    assert!(
+        provider.fips(),
+        "FIPS crypto provider requested via `fips` feature, but AWS-LC FIPS mode is not active"
+    );
+    provider
+}
+
 pub(crate) fn install_default_crypto_provider() {
-    if default_provider().install_default().is_err() {
+    if build_default_crypto_provider().install_default().is_err() {
        debug!(
            target: "rustfs::main",
            event = EVENT_CRYPTO_PROVIDER_STATE,
@@ -155,7 +168,7 @@ pub(crate) async fn init_embedded_runtime_hooks(obs_endpoint: String) -> Result<
 }

 fn install_embedded_default_crypto_provider() {
-    if let Err(err) = default_provider().install_default() {
+    if let Err(err) = build_default_crypto_provider().install_default() {
        debug!(
            component = LOG_COMPONENT_EMBEDDED,
            subsystem = LOG_SUBSYSTEM_EMBEDDED,
@@ -169,12 +182,18 @@ fn install_embedded_default_crypto_provider() {

 #[cfg(test)]
 mod tests {
-    use super::{init_profiling_runtime_with, shutdown_profiling_runtime_with};
+    use super::{build_default_crypto_provider, init_profiling_runtime_with, shutdown_profiling_runtime_with};
    use std::sync::{
        Arc,
        atomic::{AtomicBool, Ordering},
    };

+    #[test]
+    fn default_crypto_provider_matches_fips_feature() {
+        let provider = build_default_crypto_provider();
+        assert_eq!(provider.fips(), cfg!(feature = "fips"));
+    }
+
    #[tokio::test]
    async fn init_profiling_runtime_invokes_registered_hook() {
        let called = Arc::new(AtomicBool::new(false));
@@ -540,7 +540,7 @@ struct CleanupObservation {
    elapsed: Duration,
 }

-async fn observe_cleanup(replies: [CleanupReply; 2]) -> CleanupObservation {
+async fn observe_cleanup(replies: [CleanupReply; 2], measurement_delay: Duration) -> CleanupObservation {
    use tokio::io::{AsyncReadExt as _, AsyncWriteExt as _};

    const PRIVATE_BODY: &str = "private-cleanup-response-body";
@@ -583,6 +583,12 @@ async fn observe_cleanup(replies: [CleanupReply; 2]) -> CleanupObservation {
                .to_owned();
            let reply = if index < 2 {
                assert!(request_line.starts_with("PUT "));
+                if index == 1 && !measurement_delay.is_zero() {
+                    // Model descheduling during the measured PUT without a wall-clock sleep.
+                    tokio::time::pause();
+                    tokio::time::advance(measurement_delay).await;
+                    tokio::time::resume();
+                }
                CleanupReply::Status(200)
            } else {
                assert!(request_line.starts_with("DELETE "));
@@ -616,6 +622,8 @@ async fn observe_cleanup(replies: [CleanupReply; 2]) -> CleanupObservation {
    )
    .expect("cleanup probe");
    let mut request = request(ObjectOperation::PutObject);
+    // Status fixtures must reach cleanup independently of short measurement deadlines.
+    request.duration = REAL_SERVER_MEASUREMENT_BUDGET;
    request.traffic_bytes = 4;
    let log = CleanupLog::default();
    let writer = log.clone();
@@ -707,7 +715,7 @@ async fn cleanup_diagnostics_identify_each_failed_http_stage_without_private_con
        ([CleanupReply::Status(204), CleanupReply::Status(403)], "bucket_delete", 403),
        ([CleanupReply::Status(204), CleanupReply::Status(404)], "bucket_delete", 404),
    ] {
-        let observed = observe_cleanup(replies).await;
+        let observed = observe_cleanup(replies, Duration::ZERO).await;
        assert_cleanup_failure(&observed, stage, "CleanupFailed", Some(status));
        assert_eq!(observed.deletes.len(), if stage == "object_delete" { 1 } else { 2 });
    }
@@ -720,7 +728,7 @@ async fn cleanup_diagnostics_identify_each_disconnected_transport_stage() {
        ([CleanupReply::Disconnect, CleanupReply::Status(204)], "object_delete"),
        ([CleanupReply::Status(204), CleanupReply::Disconnect], "bucket_delete"),
    ] {
-        let observed = observe_cleanup(replies).await;
+        let observed = observe_cleanup(replies, Duration::ZERO).await;
        assert_cleanup_failure(&observed, stage, "ProtocolFailure", None);
    }
 }
@@ -728,13 +736,21 @@ async fn cleanup_diagnostics_identify_each_disconnected_transport_stage() {
 #[tokio::test]
 async fn cleanup_accepts_missing_object_and_still_deletes_the_bucket() {
    let _guard = TEST_LOCK.lock().await;
-    let observed = observe_cleanup([CleanupReply::Status(404), CleanupReply::Status(204)]).await;
-    assert_eq!(observed.measurement.result.outcome(), ObjectOutcome::Succeeded);
-    assert_eq!(observed.measurement.target.reason_code, ObjectTargetReasonCode::Complete);
-    assert_eq!(observed.deletes.len(), 2);
-    assert!(observed.deletes[0].ends_with("/synthetic-object HTTP/1.1"));
-    assert!(observed.deletes[1].ends_with("/ HTTP/1.1"));
-    assert!(!observed.log.contains("connect_object_cleanup_failed"));
+    for measurement_delay in [Duration::ZERO, Duration::from_secs(2)] {
+        let observed = observe_cleanup([CleanupReply::Status(404), CleanupReply::Status(204)], measurement_delay).await;
+        assert_eq!(
+            observed.measurement.result.outcome(),
+            ObjectOutcome::Succeeded,
+            "target result: {:?}; cleanup diagnostics: {}",
+            observed.measurement.target,
+            observed.log
+        );
+        assert_eq!(observed.measurement.target.reason_code, ObjectTargetReasonCode::Complete);
+        assert_eq!(observed.deletes.len(), 2);
+        assert!(observed.deletes[0].ends_with("/synthetic-object HTTP/1.1"));
+        assert!(observed.deletes[1].ends_with("/ HTTP/1.1"));
+        assert!(!observed.log.contains("connect_object_cleanup_failed"));
+    }
 }

 #[tokio::test]
@@ -750,7 +766,7 @@ async fn cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_dead
            "bucket_delete",
        ),
    ] {
-        let observed = observe_cleanup(replies).await;
+        let observed = observe_cleanup(replies, Duration::ZERO).await;
        assert_cleanup_failure(&observed, stage, "TimedOut", None);
        assert!(observed.elapsed >= Duration::from_secs(5));
        assert!(observed.log.contains("remaining_ms=0"), "{}", observed.log);
@@ -8,6 +8,7 @@ KEYCLOAK_IMAGE="${KEYCLOAK_IMAGE:-quay.io/keycloak/keycloak@sha256:6a7217a100bd3
 WORK_DIR="${RUNNER_TEMP:-/tmp}/rustfs-keycloak-live-${$}"
 KEYCLOAK_CONTAINER="rustfs-keycloak-live-${$}"
 RUSTFS_PID=""
+DOCKER=(docker)
 mkdir -p "${WORK_DIR}"

 cleanup() {
@@ -17,10 +18,10 @@ cleanup() {
    wait "${RUSTFS_PID}" 2>/dev/null || true
  fi
  if [[ "${status}" -ne 0 ]]; then
-    docker logs "${KEYCLOAK_CONTAINER}" >"${WORK_DIR}/keycloak.log" 2>&1 || true
+    "${DOCKER[@]}" logs "${KEYCLOAK_CONTAINER}" >"${WORK_DIR}/keycloak.log" 2>&1 || true
    echo "live-gate logs retained in ${WORK_DIR}" >&2
  fi
-  docker rm -f "${KEYCLOAK_CONTAINER}" >/dev/null 2>&1 || true
+  "${DOCKER[@]}" rm -f "${KEYCLOAK_CONTAINER}" >/dev/null 2>&1 || true
  if [[ "${status}" -eq 0 ]]; then
    rm -rf "${WORK_DIR}"
  fi
@@ -33,6 +34,10 @@ for command in curl docker python3 aws; do
    exit 1
  }
 done
+if ! docker info >/dev/null 2>&1; then
+  DOCKER=(sudo -n docker)
+  "${DOCKER[@]}" info >/dev/null
+fi
 [[ -x "${RUSTFS_BINARY}" ]] || {
  echo "RustFS binary is not executable: ${RUSTFS_BINARY}" >&2
  exit 1
@@ -59,7 +64,7 @@ ISSUER="${KEYCLOAK_ORIGIN}/realms/rustfs-ci"
 DISCOVERY_URL="${ISSUER}/.well-known/openid-configuration"
 RUSTFS_ORIGIN="http://127.0.0.1:${RUSTFS_PORT}"

-docker run --detach --rm \
+"${DOCKER[@]}" run --detach --rm \
  --name "${KEYCLOAK_CONTAINER}" \
  --memory 1g \
  --publish "127.0.0.1:${KEYCLOAK_PORT}:8080" \
@@ -74,8 +79,8 @@ for _ in $(seq 1 120); do
  if curl --noproxy '*' -fsS "${DISCOVERY_URL}" >"${WORK_DIR}/discovery.json" 2>/dev/null; then
    break
  fi
-  if [[ "$(docker inspect -f '{{.State.Running}}' "${KEYCLOAK_CONTAINER}" 2>/dev/null || true)" != "true" ]]; then
-    docker logs "${KEYCLOAK_CONTAINER}" >&2 || true
+  if [[ "$("${DOCKER[@]}" inspect -f '{{.State.Running}}' "${KEYCLOAK_CONTAINER}" 2>/dev/null || true)" != "true" ]]; then
+    "${DOCKER[@]}" logs "${KEYCLOAK_CONTAINER}" >&2 || true
    echo "Keycloak exited before discovery became ready" >&2
    exit 1
  fi
@@ -261,6 +261,47 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
                if missing:
                    self.assertIn(f"missing required STS field: {missing}", result.stderr)

+    def test_oidc_docker_access_and_cleanup_use_the_same_privileges(self):
+        tools = self.directory / "docker-tools"
+        tools.mkdir()
+        (tools / "docker").write_text(
+            '#!/bin/bash\n'
+            'printf "%s %s\\n" "${DOCKER_PRIVILEGED:-no}" "$*" >> "$DOCKER_CALLS"\n'
+            'case "$1" in\n'
+            '  info) [ "$DOCKER_MODE" = direct ] || '
+            '{ [ "$DOCKER_MODE" = sudo ] && [ "${DOCKER_PRIVILEGED:-no}" = yes ]; } ;;\n'
+            '  run) exit 43 ;;\n'
+            '  *) exit 0 ;;\n'
+            'esac\n'
+        )
+        (tools / "sudo").write_text(
+            '#!/bin/bash\n[ "$1" = -n ] || exit 92\nshift\n'
+            'export DOCKER_PRIVILEGED=yes\nexec "$@"\n'
+        )
+        for name in ("docker", "sudo"):
+            (tools / name).chmod(0o755)
+        (tools / "python3").symlink_to(sys.executable)
+        for name in ("aws", "curl"):
+            (tools / name).symlink_to("/usr/bin/true")
+        for mode in ("direct", "sudo", "unavailable"):
+            with self.subTest(mode=mode):
+                calls = self.directory / f"docker-{mode}.log"
+                result = subprocess.run(
+                    ["bash", str(ROOT / "scripts/test/oidc_keycloak_live.sh"), "/usr/bin/true"],
+                    env={**os.environ, "PATH": f"{tools}:/usr/bin:/bin", "RUNNER_TEMP": str(self.directory),
+                         "DOCKER_MODE": mode, "DOCKER_CALLS": str(calls)},
+                    capture_output=True, text=True, timeout=10,
+                )
+                log = calls.read_text()
+                if mode == "unavailable":
+                    self.assertNotEqual(result.returncode, 0)
+                    self.assertNotIn(" run ", log)
+                else:
+                    self.assertEqual(result.returncode, 43, result.stderr)
+                    privilege = "yes" if mode == "sudo" else "no"
+                    for command in ("run", "logs", "rm -f"):
+                        self.assertIn(f"{privilege} {command} ", log)
+
    def test_existing_evidence_directory_is_rejected(self) -> None:
        for suffix in ("", "-scratch"):
            self.setUp()
@@ -22,7 +22,7 @@ class RegistryTests(unittest.TestCase):
    def test_all_workflow_responsibilities_and_suite_exits_are_wired(self):
        self.assertTrue(registry.check_wiring(self.registry))
        self.assertEqual(len(self.registry["suites"]), 12)
-        self.assertEqual(sum(item["origin"] == "baseline" for item in self.registry["workflows"]), 60)
+        self.assertEqual(sum(item["origin"] == "baseline" for item in self.registry["workflows"]), 61)
        self.assertEqual(len(self.registry["workflows"]), len(list((registry.ROOT / ".github/workflows").glob("*.yml"))))

    def test_manual_and_periodic_plan_choose_identical_responsibilities(self):
Author	SHA1	Message	Date
Hauser Zhi	ed4099ebe1	chore(deps): update crates and fix storage CI regressions (#9111 ) * chore(deps): refresh crates and adopt openidconnect 4.1 Use published OIDC/OAuth2 releases and reject malformed HMAC signature lengths before upstream verification. Refresh workspace and fuzz locks, and advance the local zstd compatibility patch to 0.13.4 while retaining the fixed 0.14.3 codec backend. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * chore(deps): refresh compatible lockfile resolutions Resolve Prost build and derive dependencies with itertools 0.14 and prefer windows-sys 0.61.2 within the existing supported version ranges. Retain all package versions, sources and checksums. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(connect): separate cleanup checks from short probe windows Use the functional measurement budget for the cleanup HTTP fixture so setup and measured I/O can finish before checking delete status handling. Exercise both normal scheduling and an in-flight PUT clock advance, and include target classification in unexpected-failure diagnostics. Retain the cleanup deadline, outer timeout and delete-stage assertions. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(ecstore): isolate stalled multipart disk timeouts Apply the short injected timeout only to the deliberately stalled disk. Healthy disks retain their normal operation budgets, while logical lock, physical owner, immediate CopyObject and final shard assertions remain. Keep overrides scoped across cloned and reconnected handles, preserve shorter caller deadlines and explicit zero timeouts, and report drive health when the real CopyObject assertion fails. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(rebalance): retry metadata write lock quorum admission Recognize typed write-lock quorum failures for rebalance.bin within the existing bounded metadata admission retry policy. Each attempt must acquire quorum before loading or saving metadata; data writes are not replayed by this retry. Cover recovered and persistent quorum failures, cancellation after a failed attempt, and rejection of unrelated resource or read-lock errors. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 06:49:24 +00:00
Boyi (Elliot) Li	3a1fac16e3	feat(tls): add opt-in fips feature for aws-lc-rs and rustls (#9084 ) Add an opt-in `fips` Cargo feature to `rustfs` and `rustfs-tls-runtime` that enables `rustls/fips` and `tokio-rustls/fips` (activating `aws-lc-rs/fips` / `aws-lc-fips-sys` and `rustls-webpki/aws-lc-rs-fips`). When compiled with `--features fips`, `startup_runtime_hooks` installs `rustls::crypto::default_fips_provider()` as the process-wide default crypto provider and asserts `provider.fips()` at startup. Also lock `aws-lc-fips-sys v0.14.4` in `Cargo.lock`, allow the `OpenSSL` license on `aws-lc-fips-sys` in `deny.toml`, and update `docs/operations/kms-cryptographic-compliance.md` to document the scope of `-p rustfs --features fips` vs. `rustfs-crypto`. Co-authored-by: hector <42570491+majinghe@users.noreply.github.com>	2026-11-02 04:27:05 +00:00
Hauser Zhi	b00e85f1a4	test(filemeta): cover visible version after tier cleanup (#9115 ) Characterize how scanner metadata decoding and object listing handle a leading free version and a later delete marker. Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 03:59:27 +00:00
Hauser Zhi	ecbddc188a	docs(scanner): define prefix statistics contract (#9116 ) Define the scope, authority, coverage, persistence, and resource guarantees needed before scanner-derived current-object statistics can be enabled. Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 02:43:28 +00:00
Chris An	2d63931bba	fix(test): use available Docker privileges in OIDC live gate (#9109 )	2026-11-02 09:03:15 +08:00
Ramakrishna Chilaka	d429c6f792	perf(ecstore,rio,utils): accelerate MD5 ETag hashing and SWAR disk path checks (#9110 )	2026-11-02 01:02:26 +08:00
hector	e494dc2b4b	ci(validation): sweep the migrated canonical validation cases nightly (#9107 ) * ci(validation): sweep the migrated canonical validation cases nightly The canonical release-validation cases migrated into auto-testing by rustfs/auto-testing#113 (66 implemented, all non-heal) currently run only through the manual validation flow; the last fleet executions stopped on 2026-10-27 and nothing schedules them. Add a scheduled validation sweep that closes the gap: - nightly at 19:30 UTC (after the functional chain settles), rotating windows of the implemented catalog cases so the whole catalog is covered roughly weekly, with a bounded batch per night - workflow_dispatch for a single case (--case), one module (--module), or an explicit batch (--limit), mirroring how the lanes are testable standalone - fleet admission through functional_environment enter/finish with the same quarantine discipline as the lanes, inter-case node cleanup, per-case logs + reports retained as a sweep artifact - a markdown sweep report uploaded to rustfs/dashboard under functional-reports/validation/ so the cases appear next to the lane reports, plus an alert-on-failure job for scheduled runs - heal.* cases stay on the interactive validation flow: many need the reviewed native/tool collectors built by separate validation-tools runs before fleet execution The registry gains the workflow responsibility (monitor: workflow, max-age 60h so the watchdog notices a dead sweep) and the wiring test baseline count moves with it. Verified locally: check_wiring, check_test_wiring, test_validation_registry (9/9), test_functional_chain + test_functional_chain_health + test_security_workflow (99/99). * ci(validation): group the sweep report by catalog module The validation cases cover distinct areas (pool lifecycle, ILM expiration, bucket/site replication, kms, data services, identity, s3 compatibility), so a flat case table hides the category breakdown. Render one section per catalog module with its own case table and a Subtotal line, and keep the single grand - Total:/- PASS:/- FAIL: block for the dashboard overview parser. Subtotals deliberately use a different bullet prefix so the overview regexes cannot mistake a module subtotal for the grand total. * ci(validation): sweep the eight planned cases whose executors exist The planned catalog entries are not all unimplemented: all eight (kms.sse-kms-round-trip, tiering.transition-and-restore, events.delivery, audit.request-correlation, iam.policy-enforcement, oidc.login-and-session, s3.basic-object-round-trip and scanner.usage-convergence) have dedicated executors routed in run_validation_case.sh and live PASS evidence recorded in the migration map. Widen the sweep filter to include them: the rotating nightly batch now covers 74 cases (66 implemented + these 8), and the fresh evidence each sweep produces feeds their acceptance review. heal.* procedure-ready cases remain excluded - their executors need the reviewed native/tool collectors built by separate validation-tools runs before fleet execution. --------- Co-authored-by: Hauser Zhi <housemecn@gmail.com>	2026-11-01 21:53:00 +08:00
Hauser Zhi	2b7c927ee8	perf(s3): streamline conditional multipart preflight (#9099 ) * perf(s3): streamline conditional multipart preflight Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(e2e): isolate mc mirror fixture in full suite Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(connect): isolate cleanup deadline fixtures Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(ecstore): avoid lock cleanup refcount race Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): defer decommission lookup lock loss Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-01 18:59:57 +08:00