perf(ecstore,rio,utils): accelerate MD5 ETag hashing and SWAR disk path checks (#9110)

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

.config/nextest.toml

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

.github/scheduled-validations.json

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

.github/workflows/rustfs-validation-sweep.yml

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

crates/ecstore/src/core/pools.rs

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

crates/ecstore/src/data_movement/mod.rs

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

crates/ecstore/src/disk/local.rs

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

crates/ecstore/src/disk/os.rs

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

crates/ecstore/src/io_support/rio.rs

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

crates/ecstore/src/set_disk/mod.rs

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

crates/ecstore/src/set_disk/ops/multipart.rs

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

crates/ecstore/src/set_disk/ops/object.rs

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

crates/ecstore/src/store/multipart.rs

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

crates/rio/src/checksum.rs

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

crates/rio/src/etag_reader.rs

@@ -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"
+        );
+    }
 }

crates/rio/src/hash_reader.rs

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

crates/rio/src/lib.rs

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

crates/utils/src/hash.rs

@@ -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}");
+            }
+        }
+    }
 }

crates/utils/src/path.rs

@@ -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"))
+        );
+    }
 }

flake.lock

@@ -2,11 +2,11 @@
   "nodes": {
     "nixpkgs": {
       "locked": {
-        "lastModified": 1792812153,
-        "narHash": "sha256-T2+XBfNbf9pecLUWS4+RtuVNZ3W8K5taG5QdV71mWlk=",
+        "lastModified": 1793466162,
+        "narHash": "sha256-pt+peAEoOK3lLLdtbnQ1AGIBTZbdB6jrkgmurfcQwig=",
         "owner": "NixOS",
         "repo": "nixpkgs",
-        "rev": "674a2028661e550a97400c1cc992b9cca9a668a1",
+        "rev": "a0788123ada57ec92a79ab2581cda6d3c6cca75f",
         "type": "github"
       },
       "original": {
@@ -29,11 +29,11 @@
         ]
       },
       "locked": {
-        "lastModified": 1792825530,
-        "narHash": "sha256-aphXA2IFNd3q48++TKmfuuvDDjkrgGdJDnDIhij9684=",
+        "lastModified": 1793434469,
+        "narHash": "sha256-+sgVNI2c67di+XIeMaz5KSei4T65kX68s5Nl0E5Gzbc=",
         "owner": "oxalica",
         "repo": "rust-overlay",
-        "rev": "5b9f1198b59bcb31c5c3250b332e7d77590e6306",
+        "rev": "6ecf534a7c16e7c969f26838fdd8c93597ae93f3",
         "type": "github"
       },
       "original": {

scripts/test_validation_registry.py

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