perf(filemeta): grow metacache read chunks geometrically (#9133 )

Large metacache elements (an xl.meta grown past 16 MiB with its version history) were accumulated in fixed 64 KiB reads, paying hundreds of read rounds per entry on the listing path. Double the per-round chunk up to a 4 MiB bound so each reservation stays incremental and bounded (rustfs/rustfs#2715 corrupt-length-prefix safety is unchanged) while wide listings and oversized entries decode in far fewer rounds. Add regression coverage for rustfs/rustfs#9120: a listing whose aggregate stream exceeds the per-value 16 MiB bound, and an oversized entry listed among ordinary neighbors with peek/skip interleaving.
fix(metrics): reconcile peer-health gauge on successful RPCs and probes (#9134 )
2026-11-03 01:23:53 +00:00 · 2026-11-03 06:07:41 +08:00 · 2026-11-02 19:07:37 +00:00 · 2026-11-02 16:58:42 +00:00 · 2026-11-02 14:05:07 +00:00 · 2026-11-02 23:05:07 +08:00
58 changed files with 2654 additions and 394 deletions
@@ -1,2 +1,2 @@
-sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
-sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
+sha256-linux=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
+sha256-darwin=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
@@ -827,7 +827,7 @@ slow-timeout = { period = "120s", terminate-after = 12, grace-period = "10s" }
 # ---------------------------------------------------------------------------
 # e2e-distributed profile — 4-node 4-disk Actions suite
 # ---------------------------------------------------------------------------
-# Storage-sensitive PR / nightly / dispatch lane owned by
+# Nightly and manual-dispatch lane owned by
 # .github/workflows/e2e-distributed.yml.
 # Each case starts four rustfs processes (and for site replication, two
 # clusters). Upgrade cases also require RUSTFS_UPGRADE_SOURCE_BINARY.
@@ -32,6 +32,8 @@ updates:
      - "overtrue"
      - "majinghe"
    ignore:
+      - dependency-name: "pulsar"
+        versions: [ "6.9.x" ]
      - dependency-name: "object_store"
        versions: [ "0.13.x" ]
      - dependency-name: "libunftp"
@@ -769,12 +769,10 @@
      "workflow": ".github/workflows/e2e-distributed.yml",
      "name": "e2e-distributed",
      "roles": [
-        "pr",
        "periodic",
        "manual"
      ],
      "events": [
-        "pull_request",
        "workflow_dispatch",
        "schedule"
      ],
@@ -849,12 +847,11 @@
      "workflow": ".github/workflows/e2e-upgrade.yml",
      "name": "Upgrade Compatibility",
      "roles": [
-        "pr",
        "periodic",
-        "manual"
+        "manual",
+        "release"
      ],
      "events": [
-        "pull_request",
        "push",
        "schedule",
        "workflow_dispatch"
@@ -113,7 +113,7 @@ jobs:
            printf '%s\n' 'mode=full' >> "$GITHUB_OUTPUT"
            echo "Base CI policy unavailable; running the full matrix."
          fi
-          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=2$' "$GITHUB_OUTPUT"; then
+          if ! grep -q '^plan=' "$GITHUB_OUTPUT" || ! grep -q '^plan_contract=3$' "$GITHUB_OUTPUT"; then
            printf '%s\n' 'mode=full' 'connect=true' 'full_validation=true' 'rehearse_shards=false' 'packages=' 'plan=' >> "$GITHUB_OUTPUT"
            echo "Base policy does not cover this workflow contract; retaining the full matrix."
          fi
@@ -497,6 +497,63 @@ jobs:
      - name: Check offline enrollment E2E root boundary
        run: ./scripts/check_offline_enrollment_e2e.sh

+  fips-provider:
+    name: FIPS provider boundaries
+    if: needs.classify-changes.outputs.mode == 'full' && contains(fromJSON('["schedule", "workflow_dispatch"]'), github.event_name)
+    needs: [ quick-checks, classify-changes ]
+    # This builds the RustFS library test target, which exceeds hosted-runner disk capacity.
+    runs-on: sm-standard-4
+    timeout-minutes: 90
+    strategy:
+      fail-fast: false
+      matrix:
+        mode: [default, fips]
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      CARGO_BUILD_JOBS: "2"
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-fips-provider
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+
+      - name: Install FIPS build tools
+        if: matrix.mode == 'fips'
+        run: sudo apt-get install -y cmake golang clang libclang-dev
+
+      - name: Verify provider installation
+        shell: bash
+        run: |
+          args=()
+          if [[ '${{ matrix.mode }}' == 'fips' ]]; then
+            args+=(--features fips)
+          fi
+          cargo nextest run --locked -p rustfs-tls-runtime --lib "${args[@]}" --no-tests fail -E 'test(provider_policy_isolated)'
+          cargo nextest run --locked -p rustfs-s3-client --lib "${args[@]}" --no-tests fail -E 'test(overflowing_closed_range_is_rejected_before_network_io)'
+          cargo nextest run --locked -p rustfs-targets --lib "${args[@]}" --no-tests fail -E 'test(validate_pulsar)'
+          cargo nextest run --locked -p rustfs-targets --lib "${args[@]}" --no-tests fail -E 'test(webhook_provider_policy_isolated)'
+          if [[ '${{ matrix.mode }}' == 'fips' ]]; then
+            cargo nextest run --locked -p rustfs-protocols --lib --features ftps,webdav,fips --no-tests fail -E 'test(protocols_reject_incompatible_host_provider)'
+            cargo nextest run --locked -p rustfs-targets --lib --features fips --no-tests fail -E 'test(fips_postgres_rejects_incompatible_host_provider)'
+            cargo nextest run --locked -p rustfs-s3-client --lib --features fips --no-tests fail -E 'test(fips_transition_rejects_incompatible_host_provider)'
+          fi
+          tests=(provider_installation_isolated prepared_generation_shares_crypto_and_serves_sni_with_mtls)
+          if [[ '${{ matrix.mode }}' == 'fips' ]]; then
+            tests+=(fips_provider_builds_fips_tls_configs fips_provider_tls_handshake_accepts_and_rejects_algorithms)
+          fi
+          for test_name in "${tests[@]}"; do
+            cargo nextest run --locked -p rustfs --lib "${args[@]}" --no-tests fail -E "test($test_name)"
+          done
+          cargo clippy --locked -p rustfs --lib "${args[@]}" -- -D warnings
+
  test-and-lint-rio-v2:
    name: Test and Lint (rio-v2)
    # Specialized boundary, hardware and alternate-feature lanes run only on
@@ -1397,6 +1454,7 @@ jobs:
      - workspace-shard-evidence
      - core-contract
      - connect-feature-boundaries
+      - fips-provider
      - test-and-lint-rio-v2
      - test-and-lint-protocols
      - build-rustfs-debug-binary
@@ -1424,6 +1482,7 @@ jobs:
    needs:
      - classify-changes
      - connect-feature-boundaries
+      - fips-provider
      - required-checks
      - typos
      - quick-checks
@@ -17,8 +17,8 @@
 # Each selected test starts a real localhost cluster via
 # `RustFSTestClusterEnvironment` (4 processes; 4 drives per node unless the
 # case is a two-site 4-node 1-drive pair or a 4-node upgrade). Membership is
-# `[profile.e2e-distributed]` in `.config/nextest.toml`. Storage-sensitive PRs,
-# nightly runs, and manual dispatches all execute the same fail-closed suite.
+# `[profile.e2e-distributed]` in `.config/nextest.toml`. The nightly schedule
+# and manual dispatches execute the same fail-closed suite.
 # Upgrade cases download the same pinned previous release as e2e-upgrade.yml.
 #
 # Isolated pool filesystems: expand/decommission/rebalance cases require
@@ -33,32 +33,6 @@
 name: e2e-distributed

 on:
-  pull_request:
-    paths:
-      - "Cargo.lock"
-      - "Cargo.toml"
-      - ".config/nextest.toml"
-      - ".config/e2e-distributed-selection.txt"
-      - ".github/workflows/e2e-distributed.yml"
-      - "crates/audit/**"
-      - "crates/common/**"
-      - "crates/config/**"
-      - "crates/e2e_test/**"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/heal/**"
-      - "crates/iam/**"
-      - "crates/lock/**"
-      - "crates/madmin/**"
-      - "crates/notify/**"
-      - "crates/replication/**"
-      - "crates/s3-client/**"
-      - "crates/s3-ops/**"
-      - "crates/s3-types/**"
-      - "crates/scanner/**"
-      - "crates/storage-api/**"
-      - "crates/utils/**"
-      - "rustfs/**"
  workflow_dispatch:
    inputs:
      filter:
@@ -12,30 +12,19 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.

+# Single-node upgrade and rollback lane.
+#
+# The weekly Monday schedule, version tags, and workflow_dispatch run this
+# suite. Distributed upgrade cases stay on the nightly e2e-distributed lane.
+
 name: Upgrade Compatibility

 on:
-  pull_request:
-    paths:
-      - ".github/workflows/e2e-upgrade.yml"
-      - "crates/e2e_test/src/common.rs"
-      - "crates/e2e_test/src/distributed/**"
-      - "crates/e2e_test/src/fake_s3_target/**"
-      - "crates/e2e_test/src/lib.rs"
-      - "crates/e2e_test/src/replication_extension_test.rs"
-      - "crates/e2e_test/src/rpc_fault_proxy.rs"
-      - "crates/e2e_test/src/upgrade_compatibility_test.rs"
-      - "crates/ecstore/**"
-      - "crates/filemeta/**"
-      - "crates/heal/**"
-      - "crates/kms/**"
-      - "crates/storage-api/**"
-      - "rustfs/**"
-      - "Cargo.lock"
  push:
    tags:
      - "[0-9]*.[0-9]*.[0-9]*"
  schedule:
+    # 03:17 UTC every Monday.
    - cron: "17 3 * * 1"
  workflow_dispatch:

@@ -44,7 +33,7 @@ permissions:

 concurrency:
  group: ${{ github.workflow }}-${{ github.ref }}
-  cancel-in-progress: ${{ github.event_name == 'pull_request' }}
+  cancel-in-progress: false

 env:
  CARGO_TERM_COLOR: always
@@ -346,12 +346,12 @@ jobs:
          path: ${{ steps.publish.outputs.candidate_file }}
          if-no-files-found: error

-      # Publish the deb/rpm pair to the auto-testing repo's `assets` branch so
-      # engineers can download and install the nightly directly. The branch is
-      # a single-commit orphan rewritten on every build, which keeps the repo
-      # small while the latest files stay reachable at stable raw URLs.
      # Publish the deb/rpm pair as assets of the rolling `nightly` release on
      # rustfs/auto-testing (see scripts/release/publish_nightly_assets.sh).
+      # The dispatch below runs only after this upload and the immutable R2
+      # candidate both succeed. It does not replace rustfs-functional-chain.yml,
+      # which still starts from this workflow's workflow_run completion.
+      # kms-vault-lane and kms-vault-ha-failover stay parallel with this job.
      - name: Publish packages to auto-testing release assets
        env:
          ASSETS_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
@@ -360,6 +360,90 @@ jobs:
          DEB_DATE: ${{ steps.deb.outputs.deb_date }}
          BUILD_REF: ${{ env.NIGHTLY_BUILD_REF }}
        run: bash scripts/release/publish_nightly_assets.sh
+
+      # Cross-repo repository_dispatch cannot use GITHUB_TOKEN. The secret is a
+      # PAT or GitHub App token for rustfs/auto-testing: classic `repo` scope,
+      # or fine-grained Contents write (the permission GitHub checks for
+      # POST /repos/{owner}/{repo}/dispatches). A dispatch failure fails this
+      # step, and therefore this job, so the scheduled failure alert fires.
+      # GitHub allows at most 10 top-level client_payload properties, so the
+      # schema 2 candidate object is nested under `candidate` and the release
+      # fields sit beside it.
+      - name: Dispatch nightly candidate to auto-testing
+        if: ${{ steps.publish.outputs.candidate_file != '' }}
+        env:
+          DISPATCH_TOKEN: ${{ secrets.AUTO_TESTING_DISPATCH_TOKEN }}
+          CANDIDATE_FILE: ${{ steps.publish.outputs.candidate_file }}
+          DEB_DATE: ${{ steps.deb.outputs.deb_date }}
+          SOURCE_REPO: ${{ github.repository }}
+        shell: bash
+        run: |
+          set -euo pipefail
+          : "${DISPATCH_TOKEN:?AUTO_TESTING_DISPATCH_TOKEN is required to dispatch rustfs/auto-testing}"
+          : "${CANDIDATE_FILE:?nightly candidate manifest is required}"
+          : "${DEB_DATE:?DEB_DATE is required}"
+          [[ "${DEB_DATE}" =~ ^[0-9]{4}-[0-9]{2}-[0-9]{2}$ ]]
+          [[ "${SOURCE_REPO}" == "rustfs/rustfs" ]]
+
+          payload="${RUNNER_TEMP}/nightly-candidate-dispatch-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}.json"
+          jq -n --slurpfile candidate "${CANDIDATE_FILE}" \
+            --arg event_type "nightly-candidate-ready" \
+            --arg source_repo "${SOURCE_REPO}" \
+            --arg latest_alias_url "https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb" \
+            --arg auto_testing_release "https://github.com/rustfs/auto-testing/releases/tag/nightly" \
+            --arg channel "nightly" \
+            --arg deb_date "${DEB_DATE}" \
+            --arg nightly_workflow ".github/workflows/nightly-gnu.yml" \
+            '{
+              event_type: $event_type,
+              source_repo: $source_repo,
+              latest_alias_url: $latest_alias_url,
+              auto_testing_release: $auto_testing_release,
+              channel: $channel,
+              deb_date: $deb_date,
+              nightly_workflow: $nightly_workflow,
+              candidate: $candidate[0]
+            }' > "${payload}"
+          if ! jq -e '
+            .event_type == "nightly-candidate-ready"
+            and .source_repo == "rustfs/rustfs"
+            and .channel == "nightly"
+            and .deb_date == env.DEB_DATE
+            and .nightly_workflow == ".github/workflows/nightly-gnu.yml"
+            and .latest_alias_url == "https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb"
+            and .auto_testing_release == "https://github.com/rustfs/auto-testing/releases/tag/nightly"
+            and .candidate.schema == 2
+            and ((.candidate | keys | sort) == ["build_run_attempt","build_run_id","package_sha256","package_url","schema","source_ref","source_sha","workflow_sha"])
+            and ((keys | sort) == ["auto_testing_release","candidate","channel","deb_date","event_type","latest_alias_url","nightly_workflow","source_repo"])
+          ' "${payload}" >/dev/null; then
+            echo "nightly dispatch payload does not match the nightly-candidate-ready contract" >&2
+            exit 1
+          fi
+
+          body="${RUNNER_TEMP}/nightly-candidate-dispatch-body-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}.json"
+          jq -n --slurpfile client_payload "${payload}" \
+            '{event_type: "nightly-candidate-ready", client_payload: $client_payload[0]}' > "${body}"
+          response="${RUNNER_TEMP}/nightly-candidate-dispatch-response-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}.txt"
+          status="$(curl -sS --retry 3 --connect-timeout 15 --max-time 60 \
+            -o "${response}" -w '%{http_code}' \
+            -X POST \
+            -H "Authorization: Bearer ${DISPATCH_TOKEN}" \
+            -H "Accept: application/vnd.github+json" \
+            -H "X-GitHub-Api-Version: 2022-11-28" \
+            -H "Content-Type: application/json" \
+            --data-binary @"${body}" \
+            https://api.github.com/repos/rustfs/auto-testing/dispatches)" || {
+            echo "repository_dispatch request to rustfs/auto-testing failed before an HTTP status" >&2
+            exit 1
+          }
+          if [[ "${status}" != "204" ]]; then
+            echo "repository_dispatch to rustfs/auto-testing failed: HTTP ${status}" >&2
+            if [[ -s "${response}" ]]; then
+              cat "${response}" >&2
+            fi
+            exit 1
+          fi
+          echo "dispatched nightly-candidate-ready to rustfs/auto-testing (deb_date=${DEB_DATE}, run=${GITHUB_RUN_ID}, attempt=${GITHUB_RUN_ATTEMPT})"
  # Live-Vault lane for the rustfs-kms suite (rustfs/backlog#1774).
  #
  # RUSTFS_KMS_VAULT_TOKEN is the single switch that adds the Vault KV2 and
@@ -370,17 +370,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f04e0a89cc175eaf3740282537851e6fd9d7276f97730b9d0ea2859c11a85df"
 dependencies = [
 "arrow-arith",
- "arrow-array",
- "arrow-buffer",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
 "arrow-cast",
 "arrow-csv",
- "arrow-data",
- "arrow-ipc",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
 "arrow-json",
 "arrow-ord",
 "arrow-row",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "arrow-string",
 ]

@@ -390,10 +390,10 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "157943c720919e8f5b6731407ce5cd4c9a5ef47a75bf10edaf24b7081b5763ef"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
 "chrono",
 "num-traits",
 ]
@@ -405,9 +405,9 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "72b5b1682924f11ecf6929689c16cbc69efbb6dd141b06ad98ee1cb10d858c43"
 dependencies = [
 "ahash",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
 "chrono",
 "chrono-tz",
 "half",
@@ -417,6 +417,24 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "arrow-array"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0f3d22a41d3df78b71df0b685af219ae3a94188ab98d6fb29c958eb55c3247f4"
+dependencies = [
+ "ahash",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
+ "chrono",
+ "half",
+ "hashbrown 0.17.1",
+ "num-complex",
+ "num-integer",
+ "num-traits",
+]
+
 [[package]]
 name = "arrow-buffer"
 version = "59.3.2"
@@ -429,18 +447,30 @@ dependencies = [
 "num-traits",
 ]

+[[package]]
+name = "arrow-buffer"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7bd46a530bc1a8e763102fc36c1dc0f51f3ffa150c78e7f5658ccb54b1458ade"
+dependencies = [
+ "bytes",
+ "half",
+ "num-bigint 0.5.1",
+ "num-traits",
+]
+
 [[package]]
 name = "arrow-cast"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f23c3379feede42135ef27ca60e67f6e643cd4343ae74e4c8b16c83f55afd85"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
 "arrow-ord",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "atoi 2.0.0",
 "base64 0.23.1",
 "chrono",
@@ -451,15 +481,26 @@ dependencies = [
 "ryu",
 ]

+[[package]]
+name = "arrow-cmp"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1913e57a64fbf2c4502fa6d54b547d6a1ae59e3146cbc15321d47468c937abd1"
+dependencies = [
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-schema 60.1.0",
+]
+
 [[package]]
 name = "arrow-csv"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "85b1984628b091b8319b3c305fce21d04f9e9b2578bb733b0d0108462ae1d698"
 dependencies = [
- "arrow-array",
+ "arrow-array 59.3.2",
 "arrow-cast",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "chrono",
 "csv",
 "csv-core",
@@ -472,8 +513,21 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb0a031c7ff5e2bdb443117054fb466af1d4df5b7f0ecbe197713dc8b113b1ba"
 dependencies = [
- "arrow-buffer",
- "arrow-schema",
+ "arrow-buffer 59.3.2",
+ "arrow-schema 59.3.2",
+ "half",
+ "num-integer",
+ "num-traits",
+]
+
+[[package]]
+name = "arrow-data"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "415b7fbc4cf56b080bb5677a4ae74211b5c79e571ed37a1d547633ec07f3b43b"
+dependencies = [
+ "arrow-buffer 60.1.0",
+ "arrow-schema 60.1.0",
 "half",
 "num-integer",
 "num-traits",
@@ -485,28 +539,43 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f5dd1afadda2eb821aaf851bb4c0adef60c30cc0f1a29bd02fb99dd07fb9f58d"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "flatbuffers",
 "lz4_flex",
 "zstd 0.13.4",
 ]

+[[package]]
+name = "arrow-ipc"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "45f3e766bccf5a0cb57a5de52a4cba450dea5171547c2cb5939b486d08887cba"
+dependencies = [
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
+ "arrow-select 60.1.0",
+ "flatbuffers",
+ "zstd 0.14.3",
+]
+
 [[package]]
 name = "arrow-json"
 version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2bc4d56a0dad84cc51c5812ccff48b642047a1f3be80349cd84318983e79edb"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
 "arrow-cast",
 "arrow-ord",
- "arrow-schema",
- "arrow-select",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "chrono",
 "half",
 "indexmap 2.14.2",
@@ -526,11 +595,11 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "561b8209faa2cd489764b7fd07d7cb9e8d274d78048c38de59f77c41140cb834"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 ]

 [[package]]
@@ -539,10 +608,10 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0d8c68b87a569f92b90c46cb0a09d86bd9609ddc9b1bf1fc5a503c8c6907c14e"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
 "half",
 ]

@@ -556,6 +625,12 @@ dependencies = [
 "serde_json",
 ]

+[[package]]
+name = "arrow-schema"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "de2c8e714b466404cdc1189e89f892426efea614d938e16c15518c340cd43ba3"
+
 [[package]]
 name = "arrow-select"
 version = "59.3.2"
@@ -563,10 +638,25 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c844141958d805b47d1208e0d12aac53fcf388ce8011639fe82aa492adde0bb4"
 dependencies = [
 "ahash",
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "num-traits",
+]
+
+[[package]]
+name = "arrow-select"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a5af3f6656c5f2c6944f93d9a357a7f204c93d5f349385a44cf46c4a26fbcf30"
+dependencies = [
+ "ahash",
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-cmp",
+ "arrow-data 60.1.0",
+ "arrow-schema 60.1.0",
 "num-traits",
 ]

@@ -576,11 +666,11 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bc130892b1a9552da77971924b390e06cf375d1f7e4344a7b7b40ca4426fa470"
 dependencies = [
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "memchr",
 "num-traits",
 "regex",
@@ -669,9 +759,9 @@ dependencies = [

 [[package]]
 name = "async-compression"
-version = "0.4.51"
+version = "0.4.52"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9d5755640120b24961f8231e05892e4a4f7e0d04e8aeca161508c9c397946970"
+checksum = "6ad47a740fb906fffdd891e8b6d5fde2511600f7bfaebd68939fa08a14999459"
 dependencies = [
 "compression-codecs",
 "compression-core",
@@ -1993,9 +2083,9 @@ dependencies = [

 [[package]]
 name = "cc"
-version = "1.10.0"
+version = "1.11.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "86e055fa6e3e40ad6187ba25b6855eeb635d5271f8e0f0e1e9a5826b8d75bea0"
+checksum = "375989decec1a0fa424442bed520fa12ede43a48475ce8c7417d2c69b32b5f6b"
 dependencies = [
 "find-msvc-tools",
 "jobserver",
@@ -2251,9 +2341,9 @@ dependencies = [

 [[package]]
 name = "compression-codecs"
-version = "0.4.46"
+version = "0.4.47"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cae36683101b2899f48bd31f8db9839437356188e3b1ff23096e73dab09ba8fe"
+checksum = "84f840cff84d10085260e460023934d0af699e71212b09b2a41452e8bd9cda43"
 dependencies = [
 "brotli 9.1.1",
 "bzip2",
@@ -2499,9 +2589,9 @@ dependencies = [

 [[package]]
 name = "crc32fast"
-version = "1.5.3"
+version = "1.5.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0ba106e844045ddc02c20768ac4b0b7cb45f2cc2ecbd833ac9f5dacb183c2055"
+checksum = "5f97d3416d5337d15c4b0d78c8c91df922ca143f57ec20e2d1c188333047ef04"
 dependencies = [
 "cfg-if",
 ]
@@ -2891,7 +2981,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4cd5317db77ba6bca25c761a8091b9d294a9f4188e4f9dcd0243e6a57de862ac"
 dependencies = [
 "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "async-trait",
 "chrono",
 "datafusion-catalog",
@@ -2925,7 +3015,7 @@ dependencies = [
 "log",
 "object_store",
 "parking_lot 0.12.5",
- "parquet",
+ "parquet 59.3.2",
 "sqlparser",
 "tempfile",
 "tokio",
@@ -2989,8 +3079,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d26626e12b224b83351239465dcd00f72d47b02fb5e5b65a643e2dc8ed37f687"
 dependencies = [
 "arrow",
- "arrow-ipc",
- "arrow-schema",
+ "arrow-ipc 59.3.2",
+ "arrow-schema 59.3.2",
 "chrono",
 "foldhash 0.2.0",
 "half",
@@ -3001,7 +3091,7 @@ dependencies = [
 "log",
 "num-traits",
 "object_store",
- "parquet",
+ "parquet 59.3.2",
 "recursive",
 "sqlparser",
 "tokio",
@@ -3057,7 +3147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "65b1b041688cad707fa103c3e7f4c6d89b3134ebf67da0129070dad3381851eb"
 dependencies = [
 "arrow",
- "arrow-ipc",
+ "arrow-ipc 59.3.2",
 "async-trait",
 "bytes",
 "datafusion-common",
@@ -3127,7 +3217,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c521ae91b6d899d7190812881f45fe9463c222c11d90149f5332a2e56555904"
 dependencies = [
 "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "async-trait",
 "bytes",
 "datafusion-common",
@@ -3148,7 +3238,7 @@ dependencies = [
 "log",
 "object_store",
 "parking_lot 0.12.5",
- "parquet",
+ "parquet 59.3.2",
 "tokio",
 ]

@@ -3165,7 +3255,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cfdd5ac0769d1e74a885450e49fde8f63ca7ab41eb479a068564aaea80b837"
 dependencies = [
 "arrow",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
 "async-trait",
 "bytes",
 "dashmap",
@@ -3191,7 +3281,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "86560fd7df4dda924f5474b588c56e7ae8e96dcf158c7327c9cfe9daff448ec5"
 dependencies = [
 "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "async-trait",
 "chrono",
 "datafusion-common",
@@ -3227,7 +3317,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "939ae2771da8a3fdc23ab4de610ae8fef9515476626b49d96d0d5c782b105771"
 dependencies = [
 "arrow",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
 "base64 0.23.1",
 "chrono",
 "chrono-tz",
@@ -3387,7 +3477,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f002cef849e10c0597a217ec7891c2796c0f8b42d739f5d24abc24496373406"
 dependencies = [
 "arrow",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "datafusion-common",
 "datafusion-expr",
 "datafusion-expr-common",
@@ -3462,10 +3552,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b1451d6817e5ac0dd21819b19949d1c420ab51d748f55dd9a96ecb5418a24c97"
 dependencies = [
 "arrow",
- "arrow-data",
- "arrow-ipc",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
 "arrow-ord",
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "async-trait",
 "bytes",
 "datafusion-common",
@@ -3513,7 +3603,7 @@ version = "55.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "62dc5ce37a5ce1ff5362949776829aded2529a11dbaa13d24b0ffa6a059e106a"
 dependencies = [
- "arrow-schema",
+ "arrow-schema 59.3.2",
 "async-trait",
 "datafusion-common",
 "datafusion-execution",
@@ -7591,12 +7681,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8eb260d566f2fbdabe356d26aaa5d13403aac29b5b4e44463b8f9bd7b2b12680"
 dependencies = [
 "ahash",
- "arrow-array",
- "arrow-buffer",
- "arrow-data",
- "arrow-ipc",
- "arrow-schema",
- "arrow-select",
+ "arrow-array 59.3.2",
+ "arrow-buffer 59.3.2",
+ "arrow-data 59.3.2",
+ "arrow-ipc 59.3.2",
+ "arrow-schema 59.3.2",
+ "arrow-select 59.3.2",
 "base64 0.23.1",
 "brotli 8.0.4",
 "bytes",
@@ -7618,6 +7708,32 @@ dependencies = [
 "zstd 0.13.4",
 ]

+[[package]]
+name = "parquet"
+version = "60.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "806eaa0f1c0436dcb43bee8e6a82bf2efff90521a60172f8281cb9ef577b43a4"
+dependencies = [
+ "ahash",
+ "arrow-array 60.1.0",
+ "arrow-buffer 60.1.0",
+ "arrow-data 60.1.0",
+ "arrow-ipc 60.1.0",
+ "arrow-schema 60.1.0",
+ "arrow-select 60.1.0",
+ "base64 0.23.1",
+ "bytes",
+ "chrono",
+ "half",
+ "hashbrown 0.17.1",
+ "num-bigint 0.5.1",
+ "num-integer",
+ "num-traits",
+ "seq-macro",
+ "twox-hash",
+ "zstd 0.14.3",
+]
+
 [[package]]
 name = "password-hash"
 version = "0.6.1"
@@ -8976,9 +9092,9 @@ dependencies = [

 [[package]]
 name = "reflink-copy"
-version = "0.1.31"
+version = "0.1.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3dd31f458a66eeda05a9e837fc5f0effbaacbe8565868ecae5b37bf0544c19f5"
+checksum = "1ad64f61357bcf16105a0f353f9783733409c8bd889bc326ad857959605ea927"
 dependencies = [
 "cfg-if",
 "libc",
@@ -10214,18 +10330,18 @@ dependencies = [

 [[package]]
 name = "rustfs-mimalloc"
-version = "0.7.2"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8301a578772fc9823d68e083e42c3b414613d588a0eac594c1ca6bc0bbe0653f"
+checksum = "216ecfc9341b5d6e2506d47ce57c9232ae1b8020bb10264ea05286fb234286e7"
 dependencies = [
 "rustfs-mimalloc-sys",
 ]

 [[package]]
 name = "rustfs-mimalloc-sys"
-version = "0.7.2"
+version = "0.7.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6eb4082853043cb29556b4b776b05f28c3146706f43c53869dbf1ef262e00e91"
+checksum = "fb16ad68f96883407a75cdb88252b43dc2ea592428f0fc54ff7885756de4b814"
 dependencies = [
 "cc",
 ]
@@ -12420,18 +12536,18 @@ dependencies = [

 [[package]]
 name = "test-case"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "124953e7f67cb0b2fcfb87e899e4ae5a64fb68e2e160767933cc67b646fb0042"
+checksum = "ae6e80608d2037173d7d59a09089d249e43352c182f6dde6b00d138bcb7cfdbe"
 dependencies = [
 "test-case-macros",
 ]

 [[package]]
 name = "test-case-core"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bd097615b407247e102046ba34426bb594923ab3554cf0f3cdfed050f1a5a3e8"
+checksum = "29b8e1f65819001927b8cdcbb5560f2428212f2e5ab5684e3c6815a4d80e72db"
 dependencies = [
 "cfg-if",
 "proc-macro2",
@@ -12441,9 +12557,9 @@ dependencies = [

 [[package]]
 name = "test-case-macros"
-version = "3.4.0"
+version = "3.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f4ce35ece947bccac166e1ded639133827a953f69a48ba134d17f91108333082"
+checksum = "802d6165a14df1598b7290a8c546df402940d3cfbe82d15c0fd35de176771693"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -14082,9 +14198,9 @@ dependencies = [

 [[package]]
 name = "zip"
-version = "9.2.1"
+version = "9.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "848dad21639597378c3be94c59fd4de683f0f6d8a5387575896e1245737d4f37"
+checksum = "e408c6cc4699dd017d53dc080cd51c2cea2b5354af57a1479bd265ed42911487"
 dependencies = [
 "aes 0.9.3",
 "bzip2",
@@ -14095,6 +14211,7 @@ dependencies = [
 "getrandom 0.4.3",
 "hmac 0.13.0",
 "indexmap 2.14.2",
+ "libc",
 "lzma-rust2",
 "memchr",
 "nix 0.31.3",
@@ -14138,12 +14255,12 @@ name = "zstd"
 version = "0.13.4"
 dependencies = [
 "apache-avro",
- "arrow-array",
- "arrow-ipc",
- "arrow-schema",
+ "arrow-array 60.1.0",
+ "arrow-ipc 60.1.0",
+ "arrow-schema 60.1.0",
 "bytes",
 "kafka-protocol",
- "parquet",
+ "parquet 60.1.0",
 "semver",
 "toml",
 "zstd 0.14.3",
@@ -147,7 +147,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.2" }
 async-channel = "2.5.0"
 async_zip = { default-features = false, version = "0.0.19" }
 mysql_async = { default-features = false, version = "0.37.1" }
-async-compression = { version = "0.4.51" }
+async-compression = { version = "0.4.52" }
 async-trait = "0.1.92"
 async-nats = { version = "0.51.0", default-features = false }
 axum = "0.8.9"
@@ -337,7 +337,7 @@ strum = { version = "0.28.0" }
 sysinfo = "0.40.1"
 temp-env = "0.3.6"
 tempfile = "3.27.0"
-test-case = "3.4.0"
+test-case = "3.4.1"
 thiserror = "2.0.21"
 tracing = { version = "0.1.44" }
 tracing-appender = "0.2.5"
@@ -356,7 +356,7 @@ winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
 xxhash-rust = { version = "0.8.19" }
-zip = "9.2.1"
+zip = "9.2.2"
 zstd = "0.14.3"

 # Observability and Metrics
@@ -384,7 +384,7 @@ russh-sftp = "3.0.1"
 dav-server = "0.11.0"

 # Performance Analysis and Memory Profiling
-rustfs-mimalloc = { version = "0.7.2" }
+rustfs-mimalloc = { version = "0.7.3" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
 hotpath = { version = "0.30.3", default-features = false }
 # Snapshot testing for output format regression detection
@@ -26,7 +26,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
 | **protocols** | [`src/protocols/`](src/protocols) | FTPS, WebDAV, SFTP compliance. Fixed ports, own guide: [`src/protocols/README.md`](src/protocols/README.md) |
 | **reliant** | [`src/reliant/`](src/reliant) | Tests that reuse an **externally started** server (SQL/select, conditional writes, lifecycle, deleted-object reads, node-interact). Run via [`scripts/run_e2e_tests.sh`](../../scripts/run_e2e_tests.sh); see [`src/reliant/README.md`](src/reliant/README.md) |
 | **cluster** | `cluster_concurrency_test`, `stale_multipart_cleanup_cluster_test`, `namespace_lock_quorum_test`, `admin_timeout_regression_test`, `object_lambda_test`, `replication_extension_test`, `tier_stats_cluster_test` | Multi-node scenarios via `RustFSTestClusterEnvironment` |
-| **distributed 4×4** | [`src/distributed/`](src/distributed) | Storage-sensitive PR and nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
+| **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
 | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
 | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |

@@ -194,9 +194,9 @@ the same profile for membership and execution with one nightly worker.
 | `s3s-e2e` black-box | `e2e-tests` + `e2e-tests-rio-v2` jobs | **Active** (external conformance tool) |
 | ILM / lifecycle (ignored) | `test-ilm-integration-serial` lane, `-j1` | **Active** (backlog#1148 ilm-1) |
 | KMS suite | `e2e-full` job, merge queue + main | **Active** |
-| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, storage-sensitive PRs + release tags + weekly | **Active** |
+| Direct and mixed-version rolling upgrades from pinned previous release | `e2e-upgrade.yml`, release tags + weekly schedule + manual dispatch | **Active** |
 | Cluster faults (`e2e-nightly` profile) | consolidated nightly workflow | **Active** (backlog#1149 ci-7) |
-| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (storage-sensitive PR / nightly / dispatch) |
+| Distributed 4-node 4-disk (`e2e-distributed` profile) | `.github/workflows/e2e-distributed.yml` | **Active** (nightly / dispatch) |
 | Protocols (FTPS/WebDAV/SFTP) | consolidated nightly workflow, serial | **Active** (backlog#1149 ci-7) |
 | Replication (fast subset) | `e2e-smoke` profile, `e2e-tests` job, every PR | **Active** (backlog#1147 repl-1) |
 | Replication (slow + multi-node) | `e2e-repl-nightly` profile, consolidated nightly workflow | **Active** (backlog#1147 repl-1) |
@@ -12,11 +12,249 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.

-use super::harness::{DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, unique_bucket};
+use super::harness::{DistCluster, DistLayout, TestResult, payload_for, s3_probe_client, sha256_hex, unique_bucket};
 use crate::common::init_logging;
+use aws_sdk_s3::{
+    Client,
+    error::{ProvideErrorMetadata, SdkError},
+    primitives::ByteStream,
+};
+use std::any::TypeId;
 use std::collections::BTreeSet;
+use std::fmt::Debug;
+use std::future::Future;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::Barrier;
+use tokio::time::Instant;
+
+const LOCK_ADMISSION_BUDGET: Duration = Duration::from_secs(45);
+
+async fn after_lock_contention<T, E, F, Fut>(operation: &str, mut call: F) -> TestResult<Result<T, SdkError<E>>>
+where
+    E: Debug + ProvideErrorMetadata + 'static,
+    F: FnMut() -> Fut,
+    Fut: Future<Output = Result<T, SdkError<E>>>,
+{
+    let started = Instant::now();
+    let mut attempts = 0;
+    // The workload uses unique unversioned keys. Repeat only explicitly
+    // advertised lock backpressure; every step still needs its real result.
+    let outcome = tokio::time::timeout(LOCK_ADMISSION_BUDGET, async {
+        loop {
+            attempts += 1;
+            let result = call().await;
+            let retry = result.as_ref().err().is_some_and(|error| {
+                let Some(response) = error.raw_response() else {
+                    return false;
+                };
+                let Some(service_error) = error.as_service_error() else {
+                    return false;
+                };
+                // HEAD has no XML error body, so its SDK service error can
+                // lack a code even when the server supplies retry headers.
+                let namespace_code = service_error.code() == Some("ServiceUnavailable")
+                    || (service_error.code().is_none()
+                        && TypeId::of::<E>() == TypeId::of::<aws_sdk_s3::operation::head_object::HeadObjectError>());
+                response.status().as_u16() == 503
+                    && namespace_code
+                    && response.headers().get("x-rustfs-error-cause") == Some("namespace_lock_timeout")
+                    && response.headers().get("retry-after") == Some("1")
+            });
+            if !retry {
+                return result;
+            }
+            tokio::time::sleep(Duration::from_secs(1)).await;
+        }
+    })
+    .await;
+    if attempts > 1 || !matches!(&outcome, Ok(Ok(_))) {
+        eprintln!(
+            "mixed workload request: operation={operation}, attempts={attempts}, elapsed={:?}",
+            started.elapsed()
+        );
+    }
+    outcome.map_err(|_| {
+        format!("{operation} exceeded its {LOCK_ADMISSION_BUDGET:?} request budget after {attempts} attempts").into()
+    })
+}
+
+async fn assert_workload_bytes(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult {
+    let output = after_lock_contention(&format!("GET {key}"), || client.get_object().bucket(bucket).key(key).send()).await??;
+    let actual = output.body.collect().await?.into_bytes();
+    if actual.as_ref() != expected {
+        return Err(format!(
+            "object {key} bytes mismatch: expected {} bytes sha256={} got {} bytes sha256={}",
+            expected.len(),
+            sha256_hex(expected),
+            actual.len(),
+            sha256_hex(&actual)
+        )
+        .into());
+    }
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use aws_smithy_http_client::test_util::infallible_client_fn;
+    use aws_smithy_types::body::SdkBody;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+
+    type Reply = (u16, &'static str, Option<&'static str>, Option<&'static str>, &'static str);
+
+    fn client(respond: impl Fn(usize) -> Reply + Send + Sync + 'static) -> (Client, Arc<AtomicUsize>) {
+        let requests = Arc::new(AtomicUsize::new(0));
+        let seen = Arc::clone(&requests);
+        let transport = infallible_client_fn(move |_| {
+            let (status, code, cause, retry_after, body) = respond(seen.fetch_add(1, Ordering::SeqCst));
+            let body = if code.is_empty() {
+                body.to_string()
+            } else {
+                format!("<Error><Code>{code}</Code></Error>")
+            };
+            let mut response = http::Response::builder()
+                .status(status)
+                .header("content-type", "application/xml")
+                .header("content-length", body.len());
+            if let Some(cause) = cause {
+                response = response.header("x-rustfs-error-cause", cause);
+            }
+            if let Some(retry_after) = retry_after {
+                response = response.header("retry-after", retry_after);
+            }
+            response.body(SdkBody::from(body)).expect("scripted S3 response")
+        });
+        let config = crate::common::build_test_s3_config(
+            "http://contention.example.invalid",
+            "synthetic-access",
+            "synthetic-secret",
+            None,
+            "contention-test",
+        );
+        let client = Client::from_conf(config.to_builder().http_client(transport).build());
+        (s3_probe_client(&client), requests)
+    }
+
+    fn contention() -> Reply {
+        (503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("1"), "")
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn advertised_contention_recovers_after_standard_sdk_attempts_are_exhausted() -> TestResult {
+        let (client, requests) = client(|attempt| {
+            if attempt < 3 {
+                contention()
+            } else {
+                (200, "", None, None, "")
+            }
+        });
+        let standard = Client::from_conf(
+            client
+                .config()
+                .to_builder()
+                .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(3))
+                .build(),
+        );
+        standard
+            .put_object()
+            .bucket("bucket")
+            .key("key")
+            .body(ByteStream::from_static(b"body"))
+            .send()
+            .await
+            .expect_err("three advertised lock responses exhaust the standard SDK policy");
+        assert_eq!(requests.load(Ordering::SeqCst), 3);
+        requests.store(0, Ordering::SeqCst);
+        let started = Instant::now();
+        after_lock_contention("PUT recovery", || {
+            client
+                .put_object()
+                .bucket("bucket")
+                .key("key")
+                .body(ByteStream::from_static(b"body"))
+                .send()
+        })
+        .await??;
+        assert_eq!(requests.load(Ordering::SeqCst), 4);
+        assert!(started.elapsed() >= Duration::from_secs(3), "honor each Retry-After response");
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn bodyless_head_contention_uses_explicit_retry_headers() -> TestResult {
+        let (client, requests) = client(|attempt| {
+            if attempt == 0 {
+                (503, "", Some("namespace_lock_timeout"), Some("1"), "")
+            } else {
+                (200, "", None, None, "")
+            }
+        });
+        after_lock_contention("HEAD recovery", || client.head_object().bucket("bucket").key("key").send()).await??;
+        assert_eq!(requests.load(Ordering::SeqCst), 2);
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn unrelated_or_malformed_responses_are_not_retried() -> TestResult {
+        for reply in [
+            (500, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
+            (403, "AccessDenied", None, None, ""),
+            (404, "NoSuchKey", None, None, ""),
+            (503, "ServiceUnavailable", Some("storage_write_quorum"), None, ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_unavailable"), Some("1"), ""),
+            (503, "InternalError", Some("namespace_lock_timeout"), Some("1"), ""),
+            (503, "", Some("namespace_lock_timeout"), Some("1"), ""),
+            (503, "ServiceUnavailable", None, Some("1"), ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_timeout"), None, ""),
+            (503, "ServiceUnavailable", Some("namespace_lock_timeout"), Some("2"), ""),
+        ] {
+            let (client, requests) = client(move |_| reply);
+            let result = after_lock_contention("PUT fatal response", || {
+                client
+                    .put_object()
+                    .bucket("bucket")
+                    .key("key")
+                    .body(ByteStream::from_static(b"body"))
+                    .send()
+            })
+            .await?;
+            assert!(result.is_err(), "fatal response cannot become success: {reply:?}");
+            assert_eq!(requests.load(Ordering::SeqCst), 1, "fatal response must remain visible: {reply:?}");
+        }
+        Ok(())
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn persistent_lock_contention_exhausts_the_request_budget() {
+        let (client, requests) = client(|_| contention());
+        let started = Instant::now();
+        let result = after_lock_contention("PUT unavailable", || {
+            client
+                .put_object()
+                .bucket("bucket")
+                .key("key")
+                .body(ByteStream::from_static(b"body"))
+                .send()
+        })
+        .await;
+        assert!(result.is_err(), "persistent contention cannot acknowledge an operation");
+        assert!(started.elapsed() >= LOCK_ADMISSION_BUDGET);
+        assert!(requests.load(Ordering::SeqCst) > 1);
+        assert!(requests.load(Ordering::SeqCst) <= LOCK_ADMISSION_BUDGET.as_secs() as usize + 1);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn acknowledged_wrong_bytes_fail_without_retry() {
+        let (client, requests) = client(|_| (200, "", None, None, "wrong bytes"));
+        let error = assert_workload_bytes(&client, "bucket", "key", b"expected bytes")
+            .await
+            .expect_err("a successful but incorrect read is a consistency failure");
+        assert!(error.to_string().contains("bytes mismatch"));
+        assert_eq!(requests.load(Ordering::SeqCst), 1);
+    }
+}

 #[tokio::test]
 async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node() -> TestResult {
@@ -24,7 +262,9 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
    let dist = DistCluster::start(DistLayout::FourByFour).await?;
    let bucket = unique_bucket("conc");
    dist.create_bucket(&bucket).await?;
-    let clients = Arc::new(dist.clients()?);
+    // This consistency probe owns its admission policy; unrelated responses
+    // must not be hidden by the SDK's general-purpose retries.
+    let clients = Arc::new(dist.clients()?.iter().map(s3_probe_client).collect::<Vec<_>>());
    let worker_count = 24;
    let rounds = 4;
    let barrier = Arc::new(Barrier::new(worker_count));
@@ -44,31 +284,46 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
                let key = format!("source/worker-{idx:02}-round-{round}.bin");
                let copy_key = format!("retained/worker-{idx:02}-round-{round}.bin");
                let body = payload_for(&key, 64 * 1024);
-                put_object(writer, &bucket, &key, body.clone()).await?;
+                let context = format!("worker={idx} round={round}");
+                after_lock_contention(&format!("{context} PUT source"), || {
+                    writer
+                        .put_object()
+                        .bucket(&bucket)
+                        .key(&key)
+                        .body(ByteStream::from(body.clone()))
+                        .send()
+                })
+                .await??;

-                let head = reader.head_object().bucket(&bucket).key(&key).send().await?;
+                let head = after_lock_contention(&format!("{context} HEAD source"), || {
+                    reader.head_object().bucket(&bucket).key(&key).send()
+                })
+                .await??;
                if head.content_length() != Some(body.len() as i64) {
                    return Err(format!("HEAD returned the wrong size for {key}: {head:?}").into());
                }
-                assert_object_bytes(reader, &bucket, &key, &body).await?;
+                assert_workload_bytes(reader, &bucket, &key, &body).await?;

-                copier
-                    .copy_object()
-                    .bucket(&bucket)
-                    .key(&copy_key)
-                    .copy_source(format!("{bucket}/{key}"))
-                    .send()
-                    .await?;
-                assert_object_bytes(writer, &bucket, &copy_key, &body).await?;
+                after_lock_contention(&format!("{context} COPY retained"), || {
+                    copier
+                        .copy_object()
+                        .bucket(&bucket)
+                        .key(&copy_key)
+                        .copy_source(format!("{bucket}/{key}"))
+                        .send()
+                })
+                .await??;
+                assert_workload_bytes(writer, &bucket, &copy_key, &body).await?;

-                writer.delete_object().bucket(&bucket).key(&key).send().await?;
-                let missing = reader
-                    .head_object()
-                    .bucket(&bucket)
-                    .key(&key)
-                    .send()
-                    .await
-                    .expect_err("deleted source key must not remain visible");
+                after_lock_contention(&format!("{context} DELETE source"), || {
+                    writer.delete_object().bucket(&bucket).key(&key).send()
+                })
+                .await??;
+                let missing = after_lock_contention(&format!("{context} HEAD deleted source"), || {
+                    reader.head_object().bucket(&bucket).key(&key).send()
+                })
+                .await?
+                .expect_err("deleted source key must not remain visible");
                if missing.raw_response().map(|response| response.status().as_u16()) != Some(404) {
                    return Err(format!("deleted source {key} returned an unexpected result: {missing:?}").into());
                }
@@ -82,14 +337,22 @@ async fn four_node_high_concurrency_mixed_workload_is_consistent_on_every_node()
    for handle in handles {
        inventory.extend(handle.await??);
    }
+    assert_eq!(
+        inventory.len(),
+        worker_count * rounds,
+        "every worker round must retain its acknowledged copy"
+    );

    let expected_keys: BTreeSet<_> = inventory.iter().map(|(key, _)| key.as_str()).collect();
    for (node_idx, client) in clients.iter().enumerate() {
-        let listed = client.list_objects_v2().bucket(&bucket).prefix("retained/").send().await?;
+        let listed = after_lock_contention(&format!("node={node_idx} LIST retained"), || {
+            client.list_objects_v2().bucket(&bucket).prefix("retained/").send()
+        })
+        .await??;
        let listed_keys: BTreeSet<_> = listed.contents().iter().filter_map(|object| object.key()).collect();
        assert_eq!(listed_keys, expected_keys, "node {node_idx} returned a divergent retained-key listing");
        for (key, body) in &inventory {
-            assert_object_bytes(client, &bucket, key, body)
+            assert_workload_bytes(client, &bucket, key, body)
                .await
                .map_err(|error| format!("node {node_idx} failed to read {key}: {error}"))?;
        }
@@ -31,6 +31,7 @@ workspace = true

 [features]
 default = []
+fips = ["rustfs-s3-client/fips"]
 diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"]
 # Sampled wall/CPU diagnostics only; does not change the read backend.
 diagnostic-local-batch = ["rustix/time"]
@@ -1637,6 +1637,7 @@ impl RemoteDisk {
        let initial_probe_ok = Self::perform_connectivity_check(&addr).await.is_ok();
        if initial_probe_ok {
            health.record_operation_success(&endpoint, "connectivity_probe_success");
+            rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&addr);
        } else if health.mark_failure(&endpoint, "connectivity_probe_failed") {
            warn!(
                event = EVENT_REMOTE_DISK_HEALTH,
@@ -1708,6 +1709,7 @@ impl RemoteDisk {
                    // Perform basic connectivity check
                    if Self::perform_connectivity_check(&addr).await.is_ok() {
                        health.record_operation_success(&endpoint, "connectivity_probe_success");
+                        rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&addr);
                    } else if health.mark_failure(&endpoint, "connectivity_probe_failed") {
                        warn!(
                            event = EVENT_REMOTE_DISK_HEALTH,
@@ -1778,6 +1780,7 @@ impl RemoteDisk {
                }
                _ = interval.tick() => {
                    if Self::perform_recovery_probe(&addr, &endpoint).await.is_ok() {
+                        rustfs_io_metrics::internode_metrics::record_peer_reachable(&addr);
                        let became_online = health.mark_recovery_success(&endpoint, "disk_info_probe_success");
                        debug!(
                            event = EVENT_REMOTE_DISK_HEALTH,
@@ -2009,6 +2012,7 @@ impl RemoteDisk {
            let operation_result = operation().await;
            if operation_result.is_ok() {
                self.health.record_operation_success(&self.endpoint, "operation_success");
+                rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&self.addr);
            }
            self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action)
                .await;
@@ -2023,6 +2027,7 @@ impl RemoteDisk {
                // Log success; the waiting guard balances every exit path.
                if operation_result.is_ok() {
                    self.health.record_operation_success(&self.endpoint, "operation_success");
+                    rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&self.addr);
                }
                self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action)
                    .await;
@@ -445,6 +445,10 @@ impl RemoteClient {
                }
            }
        });
+        // A successful lock RPC over a cached or fresh channel proves the peer is
+        // reachable; reconcile the internode peer-health gauge so it does not stay
+        // stale when no fresh dial fires (issue #9119).
+        rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&self.addr);
    }

    fn admit_late_release(&self) -> bool {
@@ -672,6 +676,7 @@ impl RemoteClient {
                    with_lock_peer_health(&addr, |health| {
                        health.last_success = Some(Instant::now());
                    });
+                    rustfs_io_metrics::internode_metrics::record_peer_reachable_if_offline(&addr);
                    rustfs_io_metrics::lock_metrics::record_remote_lock_rpc_late_completion(op, "success");
                    if let Some(late) = late {
                        late(response).await;
@@ -3012,7 +3012,7 @@ fn should_fail_local_inline_rollback_hardlink(dst_path: &Path) -> bool {
        .remove(dst_path)
 }

-#[cfg(test)]
+#[cfg(all(test, target_os = "linux"))]
 pub(super) fn remove_dst_base_before_commit_std(
    dst_path: &str,
    guard: os::RenameCommitGuard,
@@ -3290,7 +3290,7 @@ fn should_fail_local_inline_rollback_hardlink(_dst_path: &Path) -> bool {
    false
 }

-#[cfg(not(test))]
+#[cfg(all(not(test), target_os = "linux"))]
 pub(super) fn remove_dst_base_before_commit_std(
    _dst_path: &str,
    guard: os::RenameCommitGuard,
@@ -1936,6 +1936,7 @@ impl NativeMultipartPublication {
        *self.validated_prepared.lock() = Some(validated);
        Ok(())
    }
+    #[cfg(target_os = "linux")]
    pub(super) fn prepare_and_install_head_on_thread(
        &self,
        new_metadata: &[u8],
@@ -1677,8 +1677,230 @@ const MAX_GLOBAL_FILE_SYNCS: usize = 512;
 #[cfg(test)]
 const TEST_GLOBAL_FILE_SYNCS: usize = 64;

+const PATH_LOCK_SHARDS: usize = 64;
+const PATH_LOCK_INITIAL_GC_THRESHOLD: usize = 64;
+
+#[cfg(unix)]
+#[derive(Clone, Copy, Default)]
+struct Xxh3SliceHasher(u64);
+
+#[cfg(unix)]
+impl std::hash::Hasher for Xxh3SliceHasher {
+    #[inline]
+    fn write(&mut self, bytes: &[u8]) {
+        self.0 = xxhash_rust::xxh3::xxh3_64(bytes);
+    }
+
+    #[inline]
+    fn write_usize(&mut self, _i: usize) {}
+
+    #[inline]
+    fn finish(&self) -> u64 {
+        self.0
+    }
+}
+
+#[cfg(unix)]
+type Xxh3SliceBuildHasher = std::hash::BuildHasherDefault<Xxh3SliceHasher>;
+
+#[cfg(unix)]
+#[inline]
+fn is_canonical_unix_key_bytes(bytes: &[u8]) -> bool {
+    if bytes.starts_with(b"./") {
+        return !bytes[1..].windows(2).any(|w| w == b"//") && !bytes[1..].windows(3).any(|w| w == b"/./");
+    }
+    const SLASH_MASK: u64 = 0x2f2f_2f2f_2f2f_2f2f;
+    const LOW_BITS: u64 = 0x0101_0101_0101_0101;
+    const HIGH_BITS: u64 = 0x8080_8080_8080_8080;
+
+    let mut i = 0usize;
+    let len = bytes.len();
+    let mut prev_s = 0u64;
+    while i + 8 <= len {
+        let mut buf = [0u8; 8];
+        buf.copy_from_slice(&bytes[i..i + 8]);
+        let w = u64::from_le_bytes(buf);
+        let x = w ^ SLASH_MASK;
+        let s = (x.wrapping_sub(LOW_BITS)) & !x & HIGH_BITS;
+        let s_prev1 = (s << 8) | (prev_s >> 56);
+        let s_prev2 = (s << 16) | (prev_s >> 48);
+        if (s & (s_prev1 | s_prev2)) != 0 {
+            let start = i.saturating_sub(2);
+            let window = &bytes[start..i + 8];
+            if window.windows(2).any(|pair| pair == b"//") || window.windows(3).any(|triple| triple == b"/./") {
+                return false;
+            }
+        }
+        prev_s = s;
+        i += 8;
+    }
+    let tail_start = i.saturating_sub(2);
+    let tail = &bytes[tail_start..];
+    !tail.windows(2).any(|pair| pair == b"//") && !tail.windows(3).any(|triple| triple == b"/./")
+}
+
+#[cfg(unix)]
+#[cold]
+fn canonical_unix_path_key_slow(path: &Path) -> std::borrow::Cow<'static, [u8]> {
+    use std::os::unix::ffi::OsStrExt;
+    let mut out = Vec::with_capacity(path.as_os_str().len());
+    for comp in path.components() {
+        match comp {
+            Component::RootDir => {
+                out.clear();
+                out.push(b'/');
+            }
+            Component::CurDir => {
+                if out.is_empty() {
+                    out.push(b'.');
+                }
+            }
+            Component::ParentDir => {
+                if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
+                    out.push(b'/');
+                }
+                out.extend_from_slice(b"..");
+            }
+            Component::Normal(part) => {
+                if out.len() > 1 || (out.len() == 1 && out[0] != b'/') {
+                    out.push(b'/');
+                }
+                out.extend_from_slice(part.as_bytes());
+            }
+            Component::Prefix(_) => {}
+        }
+    }
+    std::borrow::Cow::Owned(out)
+}
+
+#[cfg(unix)]
+#[inline]
+fn canonical_unix_path_key(path: &Path) -> std::borrow::Cow<'_, [u8]> {
+    use std::os::unix::ffi::OsStrExt;
+    let mut bytes = path.as_os_str().as_bytes();
+    while bytes.len() > 1 {
+        if bytes[bytes.len() - 1] == b'/' {
+            bytes = &bytes[..bytes.len() - 1];
+        } else if bytes.len() >= 2 && bytes[bytes.len() - 2] == b'/' && bytes[bytes.len() - 1] == b'.' {
+            bytes = if bytes.len() == 2 { b"/" } else { &bytes[..bytes.len() - 2] };
+        } else {
+            break;
+        }
+    }
+    if is_canonical_unix_key_bytes(bytes) {
+        return std::borrow::Cow::Borrowed(bytes);
+    }
+    canonical_unix_path_key_slow(path)
+}
+
+#[cfg(unix)]
+#[inline]
+fn unix_canonical_key_starts_with(path_key: &[u8], root_key: &[u8]) -> bool {
+    if root_key.is_empty() {
+        return true;
+    }
+    if root_key == b"/" {
+        return path_key.starts_with(b"/");
+    }
+    path_key.starts_with(root_key) && (path_key.len() == root_key.len() || path_key.get(root_key.len()) == Some(&b'/'))
+}
+
+struct PathLockShard<L> {
+    #[cfg(unix)]
+    entries: HashMap<Box<[u8]>, Weak<L>, Xxh3SliceBuildHasher>,
+    #[cfg(not(unix))]
+    entries: HashMap<PathBuf, Weak<L>>,
+    next_gc_threshold: usize,
+}
+
+impl<L> Default for PathLockShard<L> {
+    fn default() -> Self {
+        Self {
+            entries: HashMap::default(),
+            next_gc_threshold: PATH_LOCK_INITIAL_GC_THRESHOLD,
+        }
+    }
+}
+
+struct ShardedPathLockRegistry<L> {
+    shards: [Mutex<PathLockShard<L>>; PATH_LOCK_SHARDS],
+}
+
+impl<L> ShardedPathLockRegistry<L> {
+    fn new() -> Self {
+        Self {
+            shards: std::array::from_fn(|_| Mutex::new(PathLockShard::default())),
+        }
+    }
+
+    #[cfg(unix)]
+    #[inline]
+    fn get_or_create_by_key(&self, key: &[u8], init: impl FnOnce() -> L) -> Arc<L> {
+        let hash = xxhash_rust::xxh3::xxh3_64(key);
+        let shard_idx = ((hash >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
+        let mut shard = self.shards[shard_idx].lock();
+        if let Some(existing) = shard.entries.get(key).and_then(Weak::upgrade) {
+            return existing;
+        }
+        let created = Arc::new(init());
+        let weak = Arc::downgrade(&created);
+        if let Some(slot) = shard.entries.get_mut(key) {
+            *slot = weak;
+            return created;
+        }
+        if shard.entries.len() >= shard.next_gc_threshold {
+            shard.entries.retain(|_, entry| entry.strong_count() > 0);
+            shard.next_gc_threshold = shard
+                .entries
+                .len()
+                .saturating_add(1)
+                .saturating_mul(2)
+                .max(PATH_LOCK_INITIAL_GC_THRESHOLD);
+        }
+        shard.entries.insert(key.into(), weak);
+        created
+    }
+
+    #[inline]
+    fn get_or_create(&self, path: &Path, init: impl FnOnce() -> L) -> Arc<L> {
+        #[cfg(unix)]
+        {
+            let key = canonical_unix_path_key(path);
+            self.get_or_create_by_key(key.as_ref(), init)
+        }
+        #[cfg(not(unix))]
+        {
+            use std::hash::{Hash, Hasher};
+            let mut hasher = std::collections::hash_map::DefaultHasher::new();
+            path.hash(&mut hasher);
+            let shard_idx = ((hasher.finish() >> 32) as usize) & (PATH_LOCK_SHARDS - 1);
+            let mut shard = self.shards[shard_idx].lock();
+            if let Some(existing) = shard.entries.get(path).and_then(Weak::upgrade) {
+                return existing;
+            }
+            let created = Arc::new(init());
+            let weak = Arc::downgrade(&created);
+            if let Some(slot) = shard.entries.get_mut(path) {
+                *slot = weak;
+                return created;
+            }
+            if shard.entries.len() >= shard.next_gc_threshold {
+                shard.entries.retain(|_, entry| entry.strong_count() > 0);
+                shard.next_gc_threshold = shard
+                    .entries
+                    .len()
+                    .saturating_add(1)
+                    .saturating_mul(2)
+                    .max(PATH_LOCK_INITIAL_GC_THRESHOLD);
+            }
+            shard.entries.insert(path.to_path_buf(), weak);
+            created
+        }
+    }
+}
+
 static FILE_SYNC_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(global_file_sync_limit()));
-static DISK_FILE_SYNC_LIMITERS: LazyLock<Mutex<HashMap<PathBuf, Weak<Semaphore>>>> = LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_FILE_SYNC_LIMITERS: LazyLock<ShardedPathLockRegistry<Semaphore>> = LazyLock::new(ShardedPathLockRegistry::new);

 /// Dedicated tokio runtime for fsync/fdatasync blocking operations. When
 /// configured with >1 threads, isolates device-bound fsync from the main
@@ -1742,12 +1964,10 @@ fn fsync_spawn_blocking<T: Send + 'static>(f: impl FnOnce() -> T + Send + 'stati
        f()
    })
 }
-static DISK_VOLUME_MUTATION_LOCKS: LazyLock<Mutex<HashMap<PathBuf, Weak<RwLock<()>>>>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_VOLUME_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<RwLock<()>>> = LazyLock::new(ShardedPathLockRegistry::new);
 type NamespaceMutationLock = AsyncMutex<()>;
-type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
-static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
+    LazyLock::new(ShardedPathLockRegistry::new);

 fn default_global_file_sync_limit(cpu_count: usize, max_blocking_threads: usize) -> usize {
    let cpu_scaled = cpu_count
@@ -1771,15 +1991,7 @@ fn global_file_sync_limit() -> usize {

 /// Reuse a disk's limiter across reconnects while detached sync calls still hold it.
 pub(crate) fn disk_file_sync_limiter(root: &Path) -> Arc<Semaphore> {
-    let mut limiters = DISK_FILE_SYNC_LIMITERS.lock();
-    limiters.retain(|_, limiter| limiter.strong_count() > 0);
-    if let Some(limiter) = limiters.get(root).and_then(Weak::upgrade) {
-        return limiter;
-    }
-
-    let limiter = Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS));
-    limiters.insert(root.to_path_buf(), Arc::downgrade(&limiter));
-    limiter
+    DISK_FILE_SYNC_LIMITERS.get_or_create(root, || Semaphore::new(MAX_PARALLEL_FILE_SYNCS))
 }

 // Keep warm draining workers per disk/lane and spin briefly in user space
@@ -2147,45 +2359,131 @@ pub(crate) fn enter_blocking_test_hook() -> BlockingTestHookGuard {
    BlockingTestHookGuard { queue }
 }

-static DISK_BLOCKING_QUEUES: LazyLock<Mutex<HashMap<PathBuf, Weak<DiskBlockingQueue>>>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+struct RegisteredDiskBlockingQueue {
+    #[cfg(not(unix))]
+    root: PathBuf,
+    #[cfg(unix)]
+    canonical_root: Box<[u8]>,
+    queue: Weak<DiskBlockingQueue>,
+}
+
+static DISK_BLOCKING_QUEUES: LazyLock<parking_lot::RwLock<Vec<RegisteredDiskBlockingQueue>>> =
+    LazyLock::new(|| parking_lot::RwLock::new(Vec::new()));
 static FALLBACK_BLOCKING_QUEUES: LazyLock<[Arc<DiskBlockingQueue>; FALLBACK_NAMESPACE_QUEUE_SHARDS]> =
    LazyLock::new(|| std::array::from_fn(|_| DiskBlockingQueue::new()));

 pub(crate) fn register_disk_blocking_queue(root: &Path, io_root: &Path) -> Arc<DiskBlockingQueue> {
-    let mut queues = DISK_BLOCKING_QUEUES.lock();
-    queues.retain(|_, q| q.strong_count() > 0);
-    let queue = queues.get(root).and_then(Weak::upgrade).unwrap_or_else(|| {
-        let queue = DiskBlockingQueue::new();
-        queues.insert(root.to_path_buf(), Arc::downgrade(&queue));
-        queue
-    });
-    if io_root != root {
-        queues.insert(io_root.to_path_buf(), Arc::downgrade(&queue));
+    #[cfg(unix)]
+    let root_key = canonical_unix_path_key(root);
+    #[cfg(unix)]
+    let io_root_key = canonical_unix_path_key(io_root);
+
+    let mut queues = DISK_BLOCKING_QUEUES.write();
+    queues.retain(|entry| entry.queue.strong_count() > 0);
+    let queue = queues
+        .iter()
+        .find_map(|entry| {
+            #[cfg(unix)]
+            let matches = entry.canonical_root.as_ref() == root_key.as_ref();
+            #[cfg(not(unix))]
+            let matches = entry.root == root;
+            matches.then(|| entry.queue.upgrade()).flatten()
+        })
+        .unwrap_or_else(|| {
+            let queue = DiskBlockingQueue::new();
+            queues.push(RegisteredDiskBlockingQueue {
+                #[cfg(not(unix))]
+                root: root.to_path_buf(),
+                #[cfg(unix)]
+                canonical_root: root_key.as_ref().into(),
+                queue: Arc::downgrade(&queue),
+            });
+            queue
+        });
+    #[cfg(unix)]
+    let io_matches_root = io_root_key.as_ref() == root_key.as_ref();
+    #[cfg(not(unix))]
+    let io_matches_root = io_root == root;
+    if !io_matches_root {
+        let weak = Arc::downgrade(&queue);
+        if let Some(existing) = queues.iter_mut().find(|entry| {
+            #[cfg(unix)]
+            {
+                entry.canonical_root.as_ref() == io_root_key.as_ref()
+            }
+            #[cfg(not(unix))]
+            {
+                entry.root == io_root
+            }
+        }) {
+            existing.queue = weak;
+        } else {
+            queues.push(RegisteredDiskBlockingQueue {
+                #[cfg(not(unix))]
+                root: io_root.to_path_buf(),
+                #[cfg(unix)]
+                canonical_root: io_root_key.as_ref().into(),
+                queue: weak,
+            });
+        }
    }
    queue
 }

 fn disk_blocking_queue_for_root(root: &Path) -> Arc<DiskBlockingQueue> {
-    register_disk_blocking_queue(root, root)
-}
-
-fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
+    #[cfg(unix)]
+    let root_key = canonical_unix_path_key(root);
    {
-        let queues = DISK_BLOCKING_QUEUES.lock();
-        for (root, weak) in queues.iter() {
-            if path.starts_with(root)
-                && let Some(queue) = weak.upgrade()
-            {
+        let queues = DISK_BLOCKING_QUEUES.read();
+        for entry in queues.iter() {
+            #[cfg(unix)]
+            let matches = entry.canonical_root.as_ref() == root_key.as_ref();
+            #[cfg(not(unix))]
+            let matches = entry.root == root;
+            if matches && let Some(queue) = entry.queue.upgrade() {
                return queue;
            }
        }
    }
-    use std::hash::{Hash, Hasher};
-    let mut hasher = std::collections::hash_map::DefaultHasher::new();
-    path.hash(&mut hasher);
-    let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
-    Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    register_disk_blocking_queue(root, root)
+}
+
+fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
+    #[cfg(unix)]
+    {
+        let path_key = canonical_unix_path_key(path);
+        {
+            let queues = DISK_BLOCKING_QUEUES.read();
+            for entry in queues.iter() {
+                if unix_canonical_key_starts_with(path_key.as_ref(), entry.canonical_root.as_ref())
+                    && let Some(queue) = entry.queue.upgrade()
+                {
+                    return queue;
+                }
+            }
+        }
+        let hash = xxhash_rust::xxh3::xxh3_64(path_key.as_ref());
+        let idx = usize::try_from(hash % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
+        Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    }
+    #[cfg(not(unix))]
+    {
+        {
+            let queues = DISK_BLOCKING_QUEUES.read();
+            for entry in queues.iter() {
+                if path.starts_with(&entry.root)
+                    && let Some(queue) = entry.queue.upgrade()
+                {
+                    return queue;
+                }
+            }
+        }
+        use std::hash::{Hash, Hasher};
+        let mut hasher = std::collections::hash_map::DefaultHasher::new();
+        path.hash(&mut hasher);
+        let idx = usize::try_from(hasher.finish() % (FALLBACK_NAMESPACE_QUEUE_SHARDS as u64)).unwrap_or(0);
+        Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    }
 }

 /// Serialize a bucket's local metadata commits with physical bucket removal.
@@ -2194,32 +2492,46 @@ fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
 /// [`LocalDisk`](super::local::LocalDisk) instances share the same lock while
 /// disconnected disks do not keep the registry alive.
 pub(crate) fn disk_volume_mutation_lock(root: &Path, volume: &str) -> Arc<RwLock<()>> {
-    let key = root.join(volume);
-    let mut locks = DISK_VOLUME_MUTATION_LOCKS.lock();
-    locks.retain(|_, lock| lock.strong_count() > 0);
-    if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
-        return lock;
+    #[cfg(unix)]
+    {
+        let vol_bytes = volume.as_bytes();
+        if !vol_bytes.is_empty()
+            && vol_bytes[0] != b'/'
+            && vol_bytes != b"."
+            && !vol_bytes.starts_with(b"./")
+            && !vol_bytes.ends_with(b"/")
+            && !vol_bytes.ends_with(b"/.")
+            && is_canonical_unix_key_bytes(vol_bytes)
+            && let std::borrow::Cow::Borrowed(root_bytes) = canonical_unix_path_key(root)
+            && !root_bytes.is_empty()
+            && root_bytes != b"/"
+        {
+            let total_len = root_bytes.len() + 1 + vol_bytes.len();
+            if total_len <= 512 {
+                let mut stack_key = [0u8; 512];
+                stack_key[..root_bytes.len()].copy_from_slice(root_bytes);
+                stack_key[root_bytes.len()] = b'/';
+                stack_key[root_bytes.len() + 1..total_len].copy_from_slice(vol_bytes);
+                return DISK_VOLUME_MUTATION_LOCKS.get_or_create_by_key(&stack_key[..total_len], || RwLock::new(()));
+            }
+        }
    }
-
-    let lock = Arc::new(RwLock::new(()));
-    locks.insert(key, Arc::downgrade(&lock));
-    lock
+    let key = root.join(volume);
+    DISK_VOLUME_MUTATION_LOCKS.get_or_create(&key, || RwLock::new(()))
 }

 fn disk_namespace_mutation_lock(path: &Path) -> Arc<NamespaceMutationLock> {
-    let mut locks = DISK_NAMESPACE_MUTATION_LOCKS.lock();
-    locks.retain(|_, lock| lock.strong_count() > 0);
-    if let Some(lock) = locks.get(path).and_then(Weak::upgrade) {
-        return lock;
-    }
-
-    let lock = Arc::new(AsyncMutex::new(()));
-    locks.insert(path.to_path_buf(), Arc::downgrade(&lock));
-    lock
+    DISK_NAMESPACE_MUTATION_LOCKS.get_or_create(path, || AsyncMutex::new(()))
 }

-static DISK_METADATA_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+#[cfg(unix)]
+#[inline]
+fn disk_namespace_mutation_lock_by_key(key: &[u8]) -> Arc<NamespaceMutationLock> {
+    DISK_NAMESPACE_MUTATION_LOCKS.get_or_create_by_key(key, || AsyncMutex::new(()))
+}
+
+static DISK_METADATA_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
+    LazyLock::new(ShardedPathLockRegistry::new);

 /// Serializes the complete xl.meta read/modify/commit transaction. This domain
 /// precedes namespace/volume publication locks, whose narrower syscall leases
@@ -2233,17 +2545,7 @@ pub(crate) async fn acquire_metadata_mutation_lease(
    object: &Path,
    owner: Option<Arc<dyn Send + Sync>>,
 ) -> Arc<MetadataMutationLease> {
-    let lock = {
-        let mut locks = DISK_METADATA_MUTATION_LOCKS.lock();
-        locks.retain(|_, lock| lock.strong_count() > 0);
-        if let Some(lock) = locks.get(object).and_then(Weak::upgrade) {
-            lock
-        } else {
-            let lock = Arc::new(AsyncMutex::new(()));
-            locks.insert(object.to_path_buf(), Arc::downgrade(&lock));
-            lock
-        }
-    };
+    let lock = DISK_METADATA_MUTATION_LOCKS.get_or_create(object, || AsyncMutex::new(()));
    Arc::new(MetadataMutationLease {
        _guard: lock.lock_owned().await,
        _owner: owner,
@@ -2306,17 +2608,45 @@ pub(crate) async fn acquire_rename_data_mutation_lease_with_owner(
 ) -> Arc<NamespaceMutationLease> {
    // Metadata ownership precedes namespace paths in lexical order, then volume
    // ownership. Deduplicate identical keys so TMP-to-TMP publication cannot self-lock.
-    let destination_lock = disk_namespace_mutation_lock(destination_object);
-    let (namespace_guard, staging_guard) = match staging_prefix {
-        Some(source) if source < destination_object => {
-            let staging = disk_namespace_mutation_lock(source).lock_owned().await;
-            (destination_lock.lock_owned().await, Some(staging))
+    #[cfg(unix)]
+    let (namespace_guard, staging_guard) = {
+        let dst_key = canonical_unix_path_key(destination_object);
+        let destination_lock = disk_namespace_mutation_lock_by_key(dst_key.as_ref());
+        match staging_prefix {
+            Some(source) => {
+                let src_key = canonical_unix_path_key(source);
+                match src_key.as_ref().cmp(dst_key.as_ref()) {
+                    std::cmp::Ordering::Less => {
+                        let staging = disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await;
+                        (destination_lock.lock_owned().await, Some(staging))
+                    }
+                    std::cmp::Ordering::Greater => {
+                        let destination = destination_lock.lock_owned().await;
+                        (
+                            destination,
+                            Some(disk_namespace_mutation_lock_by_key(src_key.as_ref()).lock_owned().await),
+                        )
+                    }
+                    std::cmp::Ordering::Equal => (destination_lock.lock_owned().await, None),
+                }
+            }
+            None => (destination_lock.lock_owned().await, None),
        }
-        Some(source) if source > destination_object => {
-            let destination = destination_lock.lock_owned().await;
-            (destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
+    };
+    #[cfg(not(unix))]
+    let (namespace_guard, staging_guard) = {
+        let destination_lock = disk_namespace_mutation_lock(destination_object);
+        match staging_prefix {
+            Some(source) if source < destination_object => {
+                let staging = disk_namespace_mutation_lock(source).lock_owned().await;
+                (destination_lock.lock_owned().await, Some(staging))
+            }
+            Some(source) if source > destination_object => {
+                let destination = destination_lock.lock_owned().await;
+                (destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
+            }
+            _ => (destination_lock.lock_owned().await, None),
        }
-        _ => (destination_lock.lock_owned().await, None),
    };
    let volume_guard = disk_volume_mutation_lock(root, volume).read_owned().await;
    let blocking_queue = disk_blocking_queue_for_root(root);
@@ -11992,6 +12322,121 @@ mod tests {
        );
        assert!(preserved_final.exists(), "final file must remain when intermediate removal fails");
    }
+
+    #[cfg(unix)]
+    #[test]
+    fn canonical_unix_path_key_preserves_std_path_equality_and_prefix_semantics() {
+        let samples = [
+            "",
+            "/",
+            "///",
+            "/.",
+            "/./.",
+            ".",
+            "./",
+            "./.",
+            "./a",
+            "././a/",
+            ".//a/.",
+            "a",
+            "a/",
+            "a/.",
+            "a/b/c",
+            "a/b/c/",
+            "a//b/./c/.",
+            "/tmp/d0/bucket/obj",
+            "/tmp/d0/bucket/obj/",
+            "/tmp/d0//bucket/./obj/.",
+            "/tmp/d0/.rustfs.sys/multipart/sha/upload",
+            "/proc/self/fd/12/.",
+            "/proc/self/fd/12",
+            "/proc/self/fd/12/./bucket/obj",
+            "/proc/self/fd/12/bucket/obj",
+            "/tmp/d1",
+            "/tmp/d10",
+            "a/../b",
+            "a/../b/",
+            "/utf8/\u{00af}/segment/",
+        ];
+        for s in samples {
+            let p = Path::new(s);
+            let fast = canonical_unix_path_key(p);
+            let slow = canonical_unix_path_key_slow(p);
+            assert_eq!(fast.as_ref(), slow.as_ref(), "fast and slow key mismatch for {s:?}");
+        }
+        for a in samples {
+            for b in samples {
+                let pa = Path::new(a);
+                let pb = Path::new(b);
+                let ka = canonical_unix_path_key(pa);
+                let kb = canonical_unix_path_key(pb);
+                assert_eq!(
+                    pa == pb,
+                    ka.as_ref() == kb.as_ref(),
+                    "Path::eq vs canonical key mismatch for {a:?} and {b:?}"
+                );
+                assert_eq!(
+                    pa.starts_with(pb),
+                    unix_canonical_key_starts_with(ka.as_ref(), kb.as_ref()),
+                    "Path::starts_with vs canonical key prefix mismatch for {a:?} and {b:?}"
+                );
+            }
+        }
+
+        let dir = tempdir().expect("tempdir");
+        let p_clean = dir.path().join("bucket/object");
+        let p_trailing = PathBuf::from(format!("{}/", p_clean.display()));
+        let p_redundant = PathBuf::from(format!("{}//bucket/./object/.", dir.path().display()));
+        let lock1 = disk_namespace_mutation_lock(&p_clean);
+        let lock2 = disk_namespace_mutation_lock(&p_trailing);
+        let lock3 = disk_namespace_mutation_lock(&p_redundant);
+        assert!(Arc::ptr_eq(&lock1, &lock2));
+        assert!(Arc::ptr_eq(&lock1, &lock3));
+
+        let vlock1 = disk_volume_mutation_lock(dir.path(), "bucket");
+        let vlock2 = disk_volume_mutation_lock(&PathBuf::from(format!("{}/.", dir.path().display())), "bucket/");
+        assert!(Arc::ptr_eq(&vlock1, &vlock2));
+
+        let io_root = dir.path().join("io_root/.");
+        let q_reg = register_disk_blocking_queue(dir.path(), &io_root);
+        let q_root = disk_blocking_queue_for_root(dir.path());
+        let q_path = disk_blocking_queue_for_path(&dir.path().join("io_root/bucket/key"));
+        assert!(Arc::ptr_eq(&q_reg, &q_root));
+        assert!(Arc::ptr_eq(&q_reg, &q_path));
+    }
+
+    #[test]
+    fn sharded_path_lock_registry_reuses_slots_and_bounds_dead_weak_entries() {
+        let registry: ShardedPathLockRegistry<AsyncMutex<()>> = ShardedPathLockRegistry::new();
+        let target = Path::new("/tmp/sharded-lock-test/persistent-key");
+        let held = registry.get_or_create(target, || AsyncMutex::new(()));
+        let held_again = registry.get_or_create(target, || AsyncMutex::new(()));
+        assert!(Arc::ptr_eq(&held, &held_again), "live lock must be shared");
+
+        // Re-acquiring a key after drop must reuse the existing slot in-place.
+        let transient = Path::new("/tmp/sharded-lock-test/reused-slot");
+        drop(registry.get_or_create(transient, || AsyncMutex::new(())));
+        let reused = registry.get_or_create(transient, || AsyncMutex::new(()));
+        drop(reused);
+
+        // Flood transient keys across shards and verify `held` is never evicted
+        // and total dead entries remain bounded by the amortized GC threshold.
+        for i in 0..(PATH_LOCK_SHARDS * PATH_LOCK_INITIAL_GC_THRESHOLD * 3) {
+            let p = PathBuf::from(format!("/tmp/sharded-lock-test/ephemeral-{i}"));
+            drop(registry.get_or_create(&p, || AsyncMutex::new(())));
+        }
+        let still_held = registry.get_or_create(target, || AsyncMutex::new(()));
+        assert!(Arc::ptr_eq(&held, &still_held), "GC must never evict a live lock");
+        for shard in &registry.shards {
+            let guard = shard.lock();
+            assert!(
+                guard.entries.len() <= guard.next_gc_threshold,
+                "shard entries ({}) exceeded GC threshold ({})",
+                guard.entries.len(),
+                guard.next_gc_threshold
+            );
+        }
+    }
 }

 #[cfg(all(test, target_os = "linux"))]
@@ -925,6 +925,13 @@ impl MetaCacheEntriesSorted {

 const METACACHE_STREAM_VERSION: u8 = 2;
 const METACACHE_READ_CHUNK_SIZE: usize = 64 << 10;
+/// Upper bound on one incremental reservation while accumulating a single
+/// metacache element. Read rounds double from [`METACACHE_READ_CHUNK_SIZE`]
+/// up to this bound so a legitimately large xl.meta does not pay one small
+/// read per chunk, while a corrupt length prefix still cannot force a
+/// reservation larger than the payload that has already arrived (see
+/// rustfs/rustfs#2715 and rustfs/rustfs#9120).
+const METACACHE_READ_MAX_CHUNK_SIZE: usize = 4 << 20;
 /// msgpack `false`: the entry-stream terminator, and the only byte a legacy
 /// reader accepts after it (see `MetacacheWriter::close_with_limit_reached`).
 const METACACHE_TERMINATOR: u8 = 0xc2;
@@ -1157,13 +1164,14 @@ impl<R: AsyncRead + Unpin> MetacacheReader<R> {
        // A whole xl.meta can legitimately exceed the limit on its individual
        // msgpack values. Grow only as payload arrives, so a corrupt length
        // prefix cannot cause a speculative allocation of the advertised size.
+        let mut chunk = METACACHE_READ_CHUNK_SIZE;
        while self.buf.len() < read_size {
            let start = self.buf.len();
            let extra = if lookahead {
                let min_chunk = METACACHE_READ_CHUNK_SIZE.saturating_sub(start);
-                (read_size - start).clamp(min_chunk, METACACHE_READ_CHUNK_SIZE)
+                (read_size - start).clamp(min_chunk, chunk)
            } else {
-                (read_size - start).min(METACACHE_READ_CHUNK_SIZE)
+                (read_size - start).min(chunk)
            };
            if let Err(e) = self.buf.try_reserve(extra) {
                let err = Error::other(format!("metacache stream: buffer allocation of {extra} bytes failed: {e}"));
@@ -1194,6 +1202,7 @@ impl<R: AsyncRead + Unpin> MetacacheReader<R> {
                    return Err(err);
                }
            }
+            chunk = chunk.saturating_mul(2).min(METACACHE_READ_MAX_CHUNK_SIZE);
        }

        self.pos = read_size;
@@ -2155,6 +2164,76 @@ mod tests {
        }
    }

+    /// rustfs/rustfs#9120: a listing stream made of many ordinary entries must
+    /// decode completely even when the aggregate stream exceeds the per-value
+    /// `MAX_MSGP_ELEMENT_SIZE` bound that release 1.0.4 applied to every
+    /// element.
+    #[tokio::test]
+    async fn test_reader_wide_listing_aggregate_exceeds_per_value_limit() {
+        // ~2.9 KiB of metadata per entry matches the per-entry cost observed in
+        // the #9120 production report; 6,000 entries push the stream past 16 MiB.
+        const ENTRIES: usize = 6_000;
+        let metadata = vec![0xab; 2_880];
+        let entries: Vec<MetaCacheEntry> = (0..ENTRIES)
+            .map(|i| MetaCacheEntry {
+                name: format!("bigdir/object-{i:05}"),
+                metadata: metadata.clone(),
+                ..Default::default()
+            })
+            .collect();
+        let mut output = Cursor::new(Vec::new());
+        let mut writer = MetacacheWriter::new(&mut output);
+        writer.write(&entries).await.expect("wide listing must encode");
+        writer.close().await.expect("stream must close");
+        let bytes = output.into_inner();
+        assert!(
+            bytes.len() > 16 << 20,
+            "fixture must exceed the per-value limit, got {} bytes",
+            bytes.len()
+        );
+        let mut reader = MetacacheReader::new(Cursor::new(bytes));
+        let decoded = reader.read_all().await.expect("wide listing must decode");
+        assert_eq!(decoded, entries);
+    }
+
+    /// The #9120 failure shape seen in production: one entry whose metadata is
+    /// an xl.meta that grew past 16 MiB with its version history, listed among
+    /// ordinary neighbors and skipped past without losing the framing.
+    #[tokio::test]
+    async fn test_reader_oversized_entry_among_ordinary_neighbors_stays_listable() {
+        let small = |i: u8| MetaCacheEntry {
+            name: format!("bigdir/object-{i}"),
+            metadata: vec![i; 64],
+            ..Default::default()
+        };
+        let oversized = MetaCacheEntry {
+            name: "bigdir/huge-versions".to_string(),
+            metadata: vec![0xcd; (16 << 20) + 12_345],
+            ..Default::default()
+        };
+        let entries = vec![small(0), small(1), oversized.clone(), small(2), small(3)];
+        let mut output = Cursor::new(Vec::new());
+        let mut writer = MetacacheWriter::new(&mut output);
+        writer.write(&entries).await.expect("mixed listing must encode");
+        writer.close().await.expect("stream must close");
+
+        let mut reader = MetacacheReader::new(Cursor::new(output.into_inner()));
+        assert_eq!(
+            reader.peek().await.expect("first entry").expect("present"),
+            entries[0],
+            "an ordinary entry before the oversized one must decode"
+        );
+        reader.skip(1).await.expect("skip the first entry");
+        assert_eq!(reader.peek().await.expect("second entry").expect("present"), entries[1]);
+        assert_eq!(
+            reader.peek().await.expect("oversized entry").expect("present"),
+            oversized,
+            "an entry past the per-value limit must round-trip whole"
+        );
+        reader.skip(1).await.expect("skipping the oversized entry must stay aligned");
+        assert_eq!(reader.read_all().await.expect("tail must decode"), entries[3..]);
+    }
+
    #[test]
    fn file_info_versions_with_free_versions_includes_persisted_tier_cleanup_records() {
        let version_id = Uuid::new_v4();
@@ -1411,6 +1411,28 @@ pub fn record_peer_reachable(addr: &str) {
    publish_offline_gauge(&peers);
 }

+/// Record a successful interaction with a cluster peer, but only when the peer is currently
+/// marked offline. This is the fast recovery path for cached-channel RPCs, recovery probes,
+/// and lock-path successes that do not go through a fresh dial.
+///
+/// Fast path (peer unknown or already online): a shared read lock, no write, no gauge
+/// republish — the overwhelmingly common case on a healthy cluster. Slow path (offline →
+/// online transition): upgrades to the write lock and delegates to [`record_peer_reachable`].
+pub fn record_peer_reachable_if_offline(addr: &str) {
+    let key = normalize_peer_key(addr);
+    // Fast path: read lock to check whether the peer is currently offline.
+    {
+        let peers = CLUSTER_PEER_HEALTH.read().unwrap_or_else(|poisoned| poisoned.into_inner());
+        match peers.get(key) {
+            None => return,
+            Some(entry) if entry.online => return,
+            Some(_) => {} // offline: fall through to the write path
+        }
+    }
+    // Slow path: the peer is offline, so flip it back online and republish the gauge.
+    record_peer_reachable(addr);
+}
+
 /// Record a failed interaction with a cluster peer (dial failure or RPC-triggered eviction). After
 /// `failure_threshold` (>= 1) consecutive failures the peer flips offline.
 pub fn record_peer_unreachable(addr: &str, failure_threshold: u32) {
@@ -2188,6 +2210,34 @@ mod tests {
        assert_eq!(CLUSTER_SERVERS_OFFLINE_TOTAL, "rustfs_cluster_servers_offline_total");
    }

+    #[test]
+    fn record_peer_reachable_if_offline_recovers_stale_offline_state() {
+        let addr = "http://cluster-peer-recover-if-offline-test:9000";
+
+        // Mark the peer offline.
+        record_peer_unreachable(addr, 1);
+        assert!(cluster_peer_is_offline(addr));
+
+        // A successful cached-channel RPC (no fresh dial) clears the offline state.
+        record_peer_reachable_if_offline(addr);
+        assert!(!cluster_peer_is_offline(addr));
+        assert_eq!(cluster_peer_observed_online_status(addr), Some(true));
+    }
+
+    #[test]
+    fn record_peer_reachable_if_offline_no_op_for_online_peer() {
+        let addr = "http://cluster-peer-recover-noop-test:9000";
+
+        record_peer_reachable(addr);
+        // Should not panic or change state for an already-online peer.
+        record_peer_reachable_if_offline(addr);
+        assert_eq!(cluster_peer_observed_online_status(addr), Some(true));
+
+        // Should also be a no-op for unknown peers.
+        record_peer_reachable_if_offline("http://cluster-peer-recover-unknown-test:9000");
+        assert_eq!(cluster_peer_observed_online_status("http://cluster-peer-recover-unknown-test:9000"), None);
+    }
+
    #[test]
    fn cluster_peer_should_bypass_is_self_healing() {
        let addr = "http://cluster-peer-bypass-selfheal-test:9000";
@@ -30,6 +30,7 @@ workspace = true

 [features]
 default = []
+fips = ["rustfs-tls-runtime?/fips"]
 hotpath = [
    "hotpath/hotpath",
    "hotpath/tokio",
@@ -110,6 +110,9 @@ where
            ftps_required = self.config.ftps_required,
            "FTPS server starting"
        );
+        if self.config.tls_enabled && self.config.cert_dir.is_some() {
+            rustfs_tls_runtime::ensure_crypto_provider().map_err(|err| FtpsInitError::InvalidConfig(err.to_string()))?;
+        }
        let (reload_shutdown_tx, reload_shutdown_rx) = watch::channel(false);

        let storage_clone = self.storage.clone();
@@ -191,8 +194,6 @@ where
                );

                // Build ServerConfig with SNI support
-                let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-
                let server_config = rustls::ServerConfig::builder()
                    .with_no_client_auth()
                    .with_cert_resolver(resolver);
@@ -43,3 +43,70 @@ pub use webdav::{config::WebDavConfig, server::WebDavServer};

 #[cfg(feature = "sftp")]
 pub use sftp::{SftpConfig, SftpInitError, SftpServer};
+
+#[cfg(all(test, feature = "fips", feature = "ftps", feature = "webdav"))]
+mod fips_tests {
+    use crate::common::dummy_storage::DummyBackend;
+    use crate::{FtpsConfig, FtpsServer, WebDavConfig, WebDavServer};
+
+    #[test]
+    fn protocols_reject_incompatible_host_provider() {
+        const CHILD: &str = "RUSTFS_TEST_PROTOCOL_FIPS_PROVIDER";
+        if std::env::var_os(CHILD).is_some() {
+            let mut provider = rustfs_tls_runtime::default_crypto_provider();
+            provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+            provider.install_default().expect("install incompatible host provider");
+            let cert_dir = tempfile::tempdir().expect("certificate directory");
+            let cert_dir = cert_dir.path().to_string_lossy().into_owned();
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("test runtime");
+            runtime.block_on(async {
+                let ftps_config = FtpsConfig {
+                    cert_dir: Some(cert_dir.clone()),
+                    ..Default::default()
+                };
+                let ftps = FtpsServer::new(ftps_config, DummyBackend::default())
+                    .await
+                    .expect("FTPS config");
+                let (_sender, receiver) = tokio::sync::broadcast::channel(1);
+                let error = ftps
+                    .start(receiver)
+                    .await
+                    .expect_err("FTPS must reject the incompatible provider");
+                assert!(error.to_string().contains("process-wide rustls crypto provider"));
+
+                let webdav_config = WebDavConfig {
+                    bind_addr: ([127, 0, 0, 1], 0).into(),
+                    cert_dir: Some(cert_dir),
+                    ..Default::default()
+                };
+                let webdav = WebDavServer::new(webdav_config, DummyBackend::default())
+                    .await
+                    .expect("WebDAV config");
+                let (_sender, receiver) = tokio::sync::broadcast::channel(1);
+                let error = webdav
+                    .start(receiver)
+                    .await
+                    .expect_err("WebDAV must reject the incompatible provider");
+                assert!(error.to_string().contains("process-wide rustls crypto provider"));
+            });
+            return;
+        }
+        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .arg("protocols_reject_incompatible_host_provider")
+            .arg("--nocapture")
+            .env(CHILD, "1")
+            .env("NO_PROXY", "localhost,127.0.0.1,::1")
+            .env("no_proxy", "localhost,127.0.0.1,::1")
+            .output()
+            .expect("start isolated protocol TLS test");
+        assert!(
+            output.status.success(),
+            "stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+}
@@ -118,6 +118,10 @@ where
            "WebDAV server starting"
        );

+        if self.config.tls_enabled && self.config.cert_dir.is_some() {
+            rustfs_tls_runtime::ensure_crypto_provider().map_err(|err| WebDavInitError::Tls(err.to_string()))?;
+        }
+
        let listener = TcpListener::bind(self.config.bind_addr).await?;
        self.serve(listener, shutdown_rx).await
    }
@@ -155,8 +159,6 @@ where
                    reload_shutdown_rx.clone(),
                );

-                let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-
                let server_config = ServerConfig::builder().with_no_client_auth().with_cert_resolver(resolver);

                Some(TlsAcceptor::from(Arc::new(server_config)))
@@ -25,6 +25,10 @@ keywords = ["s3", "client", "tiering", "rustfs", "Minio"]
 categories = ["web-programming", "development-tools", "network-programming"]
 documentation = "https://docs.rs/rustfs-s3-client/latest/rustfs_s3_client/"

+[features]
+default = []
+fips = ["rustfs-tls-runtime/fips"]
+
 [dependencies]
 rustfs-checksums.workspace = true
 rustfs-config.workspace = true
@@ -81,7 +81,7 @@ use time::Duration;
 use time::OffsetDateTime;
 use tokio::io::BufReader;
 use tokio::io::{AsyncRead, AsyncReadExt};
-use tracing::{debug, error, trace, warn};
+use tracing::{error, trace, warn};
 use url::{Url, form_urlencoded};
 use uuid::Uuid;

@@ -448,16 +448,7 @@ impl TransitionClient {
        tier_type: &str,
        timeouts: TransitionClientTimeouts,
    ) -> Result<TransitionClient, std::io::Error> {
-        if rustls::crypto::CryptoProvider::get_default().is_none() {
-            // No default provider is set yet; try to install aws-lc-rs.
-            // `install_default` can only fail if another thread races us and installs a provider
-            // between our check and this call, which is still safe to ignore.
-            if rustls::crypto::aws_lc_rs::default_provider().install_default().is_err() {
-                debug!("rustls crypto provider was installed concurrently, skipping aws-lc-rs install");
-            }
-        } else {
-            debug!("rustls crypto provider already installed, skipping aws-lc-rs install");
-        }
+        rustfs_tls_runtime::ensure_crypto_provider().map_err(std::io::Error::other)?;

        let endpoint_url = get_endpoint_url(endpoint, opts.secure)?;
        let timeouts = timeouts.validate()?;
@@ -1862,21 +1853,36 @@ mod tests {
        assert!(outcome.is_ok(), "TLS config creation should not panic");
    }

-    /// Installing the rustls crypto provider when one is already set must not panic or return
-    /// an error that surfaces to callers (the race-safe `get_default` check guards the install).
+    #[cfg(feature = "fips")]
    #[test]
-    fn provider_install_is_idempotent() {
-        // Install once (may already be set by another test in this binary — that's fine).
-        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-
-        // A second install attempt on an already-set provider must not panic.
-        let outcome = std::panic::catch_unwind(|| {
-            if rustls::crypto::CryptoProvider::get_default().is_none() {
-                let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
-            }
-            // If a default is already present, the branch above is simply skipped.
-        });
-        assert!(outcome.is_ok(), "provider install guard must not panic when a provider is already set");
+    fn fips_transition_rejects_incompatible_host_provider() {
+        const CHILD: &str = "RUSTFS_TEST_TRANSITION_FIPS_PROVIDER";
+        if std::env::var_os(CHILD).is_some() {
+            let mut provider = rustfs_tls_runtime::default_crypto_provider();
+            provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+            provider.install_default().expect("install incompatible host provider");
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("test runtime");
+            let error = runtime
+                .block_on(super::TransitionClient::new("127.0.0.1:0", super::Options::default(), ""))
+                .expect_err("transition client must reject the incompatible provider");
+            assert!(error.to_string().contains("process-wide rustls crypto provider"));
+            return;
+        }
+        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .arg("fips_transition_rejects_incompatible_host_provider")
+            .arg("--nocapture")
+            .env(CHILD, "1")
+            .output()
+            .expect("start isolated transition TLS test");
+        assert!(
+            output.status.success(),
+            "stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
    }

    #[test]
@@ -640,12 +640,12 @@ fn http_range_spec_from_get_range(range: &GetRange) -> HTTPRangeSpec {
        GetRange::Bounded(range) => http_range_spec_from_range(range.clone()),
        GetRange::Offset(start) => HTTPRangeSpec {
            is_suffix_length: false,
-            start: *start as i64,
+            start: i64::try_from(*start).unwrap_or(i64::MAX),
            end: -1,
        },
        GetRange::Suffix(length) => HTTPRangeSpec {
            is_suffix_length: true,
-            start: *length as i64,
+            start: i64::try_from(*length).unwrap_or(i64::MAX),
            end: -1,
        },
    }
@@ -654,15 +654,15 @@ fn http_range_spec_from_get_range(range: &GetRange) -> HTTPRangeSpec {
 fn http_range_spec_from_range(range: Range<u64>) -> HTTPRangeSpec {
    HTTPRangeSpec {
        is_suffix_length: false,
-        start: range.start as i64,
-        end: range.end.saturating_sub(1) as i64,
+        start: i64::try_from(range.start).unwrap_or(i64::MAX),
+        end: i64::try_from(range.end.saturating_sub(1)).unwrap_or(i64::MAX),
    }
 }

 fn http_range_spec_from_start(start: u64) -> HTTPRangeSpec {
    HTTPRangeSpec {
        is_suffix_length: false,
-        start: start as i64,
+        start: i64::try_from(start).unwrap_or(i64::MAX),
        end: -1,
    }
 }
@@ -2476,6 +2476,48 @@ mod test {
        assert_eq!(range.end, 19);
    }

+    #[test]
+    fn test_get_range_conversion_for_suffix_ranges() {
+        // Normal suffix: last 100 bytes.
+        let range = http_range_spec_from_get_range(&GetRange::Suffix(100));
+        assert!(range.is_suffix_length);
+        assert_eq!(range.start, 100);
+        assert_eq!(range.end, -1);
+
+        // Overflowing suffix: `u64::MAX` previously wrapped to `-1` via `as i64`
+        // and was misread as "last 1 byte". It now clamps to `i64::MAX`, which
+        // `get_length` treats as "the whole object".
+        let range = http_range_spec_from_get_range(&GetRange::Suffix(u64::MAX));
+        assert!(range.is_suffix_length);
+        assert_eq!(range.start, i64::MAX);
+        assert_eq!(range.end, -1);
+    }
+
+    #[test]
+    fn test_get_range_conversion_for_offset_overflow() {
+        // Normal offset.
+        let range = http_range_spec_from_get_range(&GetRange::Offset(42));
+        assert!(!range.is_suffix_length);
+        assert_eq!(range.start, 42);
+        assert_eq!(range.end, -1);
+
+        // Overflowing offset clamps to `i64::MAX`, which is then rejected as
+        // out-of-range (416) by `get_offset_length`.
+        let range = http_range_spec_from_get_range(&GetRange::Offset(u64::MAX));
+        assert!(!range.is_suffix_length);
+        assert_eq!(range.start, i64::MAX);
+        assert_eq!(range.end, -1);
+    }
+
+    #[test]
+    fn test_get_range_conversion_for_bounded_overflow() {
+        // Overflowing bounded range clamps start/end to `i64::MAX`.
+        let range = http_range_spec_from_get_range(&GetRange::Bounded(u64::MAX - 10..u64::MAX));
+        assert!(!range.is_suffix_length);
+        assert_eq!(range.start, i64::MAX);
+        assert_eq!(range.end, i64::MAX);
+    }
+
    #[tokio::test]
    async fn test_scan_range_output_can_convert_field_delimiter() {
        let chunks = stream::iter(vec![Ok::<_, std::io::Error>(Bytes::from_static(b"a&&1\nb&&2\n"))]);
@@ -13,6 +13,7 @@ documentation = "https://docs.rs/rustfs-targets/latest/rustfs_targets/"

 [features]
 default = []
+fips = ["rustfs-tls-runtime/fips"]
 # Exposes the builder-style `testkit::MockTarget` to downstream test suites. Never enable this
 # from a production `[dependencies]` entry; in-crate unit tests get the module via cfg(test).
 test-support = []
@@ -162,7 +162,7 @@ pub async fn check_mysql_server_available(args: &crate::target::mysql::MySqlArgs
        .db_name(Some(dsn.database.clone()));

    if dsn.tls {
-        ensure_rustls_provider_installed();
+        ensure_rustls_provider_installed()?;
        let mut ssl_opts = SslOpts::default();
        if !args.tls_ca.is_empty() {
            ssl_opts = ssl_opts.with_root_certs(vec![PathBuf::from(args.tls_ca.clone()).into()]);
@@ -293,6 +293,10 @@ fn map_lapin_error(err: lapin::Error, context: &str) -> TargetError {

 pub async fn connect_amqp(args: &AMQPArgs) -> Result<AMQPConnection, TargetError> {
    args.validate()?;
+    #[cfg(feature = "fips")]
+    if args.url.scheme() == "amqps" {
+        super::ensure_rustls_provider_installed()?;
+    }
    tokio::time::timeout(Duration::from_secs(5), async {
        let url = connection_url(args)?;
        // Reconnect explicitly so every new channel enables publisher confirms below.
@@ -27,7 +27,7 @@ use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
 use std::thread_local;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
-use tracing::{debug, warn};
+use tracing::warn;

 pub mod amqp;
 pub mod kafka;
@@ -947,16 +947,12 @@ pub(crate) fn encode_failed_entry(

 /// Ensures a rustls crypto provider is installed before any TLS operation.
 ///
-/// Multiple target modules (MySQL, Redis, Postgres, MQTT) need this because
-/// each may be the first to perform a TLS handshake. Idempotent: if a
-/// provider is already registered, returns immediately.
-pub(crate) fn ensure_rustls_provider_installed() {
-    if rustls::crypto::CryptoProvider::get_default().is_some() {
-        return;
-    }
-    if let Err(err) = rustls::crypto::aws_lc_rs::default_provider().install_default() {
-        debug!("rustls provider already installed or unavailable: {err:?}");
-    }
+/// Target modules may be the first to construct a TLS client. In a FIPS build,
+/// an incompatible provider installed by the host is rejected.
+pub(crate) fn ensure_rustls_provider_installed() -> Result<(), TargetError> {
+    rustfs_tls_runtime::ensure_crypto_provider()
+        .map(|_| ())
+        .map_err(|err| TargetError::Configuration(err.to_string()))
 }

 #[cfg(test)]
@@ -241,7 +241,7 @@ fn build_root_store(ca_path: &str, trust_leaf_as_ca: bool) -> Result<rustls::Roo
 }

 fn build_mqtt_tls_transport(broker: &Url, tls: &MQTTTlsConfig) -> Result<Transport, TargetError> {
-    super::ensure_rustls_provider_installed();
+    super::ensure_rustls_provider_installed()?;

    let client_config = match tls
        .policy
@@ -881,7 +881,7 @@ async fn build_mysql_pool_from_args(args: &MySqlArgs) -> Result<(Pool, bool), Ta
        .db_name(Some(dsn.database.clone()));

    if dsn.tls {
-        super::ensure_rustls_provider_installed();
+        super::ensure_rustls_provider_installed()?;
        let mut ssl_opts = SslOpts::default();
        if !args.tls_ca.is_empty() {
            let _ =
@@ -204,6 +204,8 @@ fn nats_sends_credentials_without_tls(args: &NATSArgs) -> bool {

 pub async fn connect_nats(args: &NATSArgs) -> Result<async_nats::Client, TargetError> {
    args.validate()?;
+    #[cfg(feature = "fips")]
+    super::ensure_rustls_provider_installed()?;

    let mut options = async_nats::ConnectOptions::new().require_tls(args.tls_required);

@@ -462,7 +462,7 @@ pub fn table_probe_sql(schema: &str, table: &str) -> String {
 /// When `tls_client_cert` and `tls_client_key` are both set the connection
 /// uses mTLS authentication; otherwise no client cert is sent.
 pub fn build_tls_config(args: &PostgresArgs) -> Result<rustls::ClientConfig, TargetError> {
-    super::ensure_rustls_provider_installed();
+    super::ensure_rustls_provider_installed()?;

    let mut root_store = rustls::RootCertStore::empty();

@@ -939,6 +939,32 @@ mod tests {
        }
    }

+    #[cfg(feature = "fips")]
+    #[test]
+    fn fips_postgres_rejects_incompatible_host_provider() {
+        const CHILD: &str = "RUSTFS_TEST_POSTGRES_FIPS_PROVIDER";
+        if std::env::var_os(CHILD).is_some() {
+            let mut provider = rustfs_tls_runtime::default_crypto_provider();
+            provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+            provider.install_default().expect("install incompatible host provider");
+            let error = build_tls_config(&base_args()).expect_err("Postgres must reject the incompatible provider");
+            assert!(error.to_string().contains("process-wide rustls crypto provider"));
+            return;
+        }
+        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .arg("fips_postgres_rejects_incompatible_host_provider")
+            .arg("--nocapture")
+            .env(CHILD, "1")
+            .output()
+            .expect("start isolated Postgres TLS test");
+        assert!(
+            output.status.success(),
+            "stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
    #[test]
    fn validate_disabled_skips_all_checks() {
        let args = PostgresArgs {
@@ -90,7 +90,7 @@ impl PulsarArgs {
            return Ok(());
        }

-        validate_pulsar_broker(&self.broker)?;
+        let parsed = validate_pulsar_broker(&self.broker)?;

        if self.topic.trim().is_empty() {
            return Err(TargetError::Configuration("Pulsar topic cannot be empty".to_string()));
@@ -116,8 +116,6 @@ impl PulsarArgs {
            return Err(TargetError::Configuration("Pulsar queue directory must be an absolute path".to_string()));
        }

-        let parsed = Url::parse(&self.broker)
-            .map_err(|e| TargetError::Configuration(format!("Invalid Pulsar broker URL: {e} (value: '{}')", self.broker)))?;
        let tls_enabled = parsed.scheme() == "pulsar+ssl";
        // A CA bundle is TLS trust material a plaintext `pulsar://` broker
        // silently ignores, so treat it as a genuine misconfiguration. The
@@ -173,6 +171,10 @@ fn pulsar_producer_name(target_id: &TargetID, target_type: TargetType, nonce: Uu

 pub async fn connect_pulsar(args: &PulsarArgs) -> Result<Pulsar<TokioExecutor>, TargetError> {
    args.validate()?;
+    #[cfg(feature = "fips")]
+    if Url::parse(&args.broker).is_ok_and(|url| url.scheme() == "pulsar+ssl") {
+        super::ensure_rustls_provider_installed()?;
+    }

    let mut builder = Pulsar::builder(args.broker.clone(), TokioExecutor);

@@ -766,7 +766,7 @@ pub(crate) fn build_redis_client(args: &RedisArgs) -> Result<Client, TargetError

    let secure_scheme = matches!(args.url.scheme(), "rediss" | "valkeys");
    if secure_scheme {
-        super::ensure_rustls_provider_installed();
+        super::ensure_rustls_provider_installed()?;
        let tls_certs = TlsCertificates {
            client_tls: read_client_tls(&args.tls)?,
            root_cert: read_root_cert(&args.tls)?,
@@ -305,6 +305,8 @@ where
        args: &WebhookArgs,
        resolver: impl reqwest::dns::Resolve + 'static,
    ) -> Result<Client, TargetError> {
+        #[cfg(feature = "fips")]
+        super::ensure_rustls_provider_installed()?;
        let mut client_builder = Client::builder()
            .no_proxy()
            .dns_resolver(resolver)
@@ -848,6 +850,51 @@ mod tests {
        }
    }

+    #[test]
+    fn webhook_provider_policy_isolated() {
+        const CHILD: &str = "RUSTFS_TEST_WEBHOOK_PROVIDER_POLICY";
+        if std::env::var_os(CHILD).is_some() {
+            if cfg!(feature = "fips") {
+                let mut provider = rustfs_tls_runtime::default_crypto_provider();
+                provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+                provider.install_default().expect("install incompatible host provider");
+            } else {
+                assert!(rustls::crypto::CryptoProvider::get_default().is_none());
+            }
+            let args = WebhookArgs {
+                skip_tls_verify: true,
+                ..base_args()
+            };
+            let result = WebhookTarget::<serde_json::Value>::build_http_client_with_resolver(
+                &args,
+                StaticResolver("127.0.0.1".parse().expect("loopback IP")),
+            );
+            if cfg!(feature = "fips") {
+                let error = result.expect_err("FIPS webhook must reject the incompatible provider");
+                assert!(error.to_string().contains("process-wide rustls crypto provider"));
+            } else {
+                result.expect("default webhook client should build");
+                assert!(
+                    rustls::crypto::CryptoProvider::get_default().is_none(),
+                    "building a default webhook must not install a process-wide provider"
+                );
+            }
+            return;
+        }
+        let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+            .arg("webhook_provider_policy_isolated")
+            .arg("--nocapture")
+            .env(CHILD, "1")
+            .output()
+            .expect("start isolated webhook provider test");
+        assert!(
+            output.status.success(),
+            "stdout={} stderr={}",
+            String::from_utf8_lossy(&output.stdout),
+            String::from_utf8_lossy(&output.stderr)
+        );
+    }
+
    async fn http_status_url(status: u16) -> Url {
        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.expect("bind test server");
        let address = listener.local_addr().expect("test server address");
@@ -610,10 +610,12 @@ pub fn validate_certified_key(
            }
        }
    }
-    let certified = CertifiedKey::from_der(certs.to_vec(), key.clone_key(), &rustls::crypto::aws_lc_rs::default_provider())
-        .map_err(|source| TlsRuntimeError::InvalidCertifiedKey {
-            name: name.to_string(),
-            source,
+    let certified =
+        CertifiedKey::from_der(certs.to_vec(), key.clone_key(), &crate::default_crypto_provider()).map_err(|source| {
+            TlsRuntimeError::InvalidCertifiedKey {
+                name: name.to_string(),
+                source,
+            }
        })?;
    certified
        .keys_match()
@@ -21,6 +21,7 @@ pub mod fingerprint;
 pub mod material;
 pub mod metrics;
 pub mod outbound;
+pub mod provider;
 pub mod server;
 pub mod source;
 pub mod state;
@@ -45,6 +46,7 @@ pub use outbound::{
    GlobalOutboundTlsStateSummary, GlobalPublishedOutboundTlsState, load_global_outbound_tls_generation,
    load_global_outbound_tls_state, publish_global_outbound_tls_state, summarize_global_outbound_tls_state,
 };
+pub use provider::{CryptoProviderError, CryptoProviderStatus, default_crypto_provider, ensure_crypto_provider};
 pub use server::{ReloadableServerCertResolver, spawn_server_cert_reload_loop};
 pub use source::{TlsFileLayout, TlsSource, TlsSourceKind, TlsSourceSnapshot};
 pub use state::OutboundOnlySnapshotArgs;
@@ -0,0 +1,108 @@
+// 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.
+
+use rustls::crypto::CryptoProvider;
+
+/// The outcome of ensuring the process-wide rustls provider.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum CryptoProviderStatus {
+    Installed,
+    AlreadyInstalled,
+}
+
+/// An incompatible process-wide rustls provider prevents FIPS startup.
+#[derive(Debug, thiserror::Error)]
+#[error(
+    "FIPS TLS requested, but the process-wide rustls crypto provider is not FIPS compatible; configure the host to install a FIPS provider before starting RustFS"
+)]
+pub struct CryptoProviderError {
+    _private: (),
+}
+
+/// Construct the provider selected by this crate's feature configuration.
+pub fn default_crypto_provider() -> CryptoProvider {
+    #[cfg(feature = "fips")]
+    {
+        rustls::crypto::default_fips_provider()
+    }
+    #[cfg(not(feature = "fips"))]
+    {
+        rustls::crypto::aws_lc_rs::default_provider()
+    }
+}
+
+/// Install or reuse the process-wide rustls provider.
+///
+/// # Errors
+/// Returns an error in a FIPS build if the provider actually installed in the
+/// process is absent or does not satisfy rustls's FIPS checks.
+pub fn ensure_crypto_provider() -> Result<CryptoProviderStatus, CryptoProviderError> {
+    // Rustls's process-wide provider is write-once; check again after a race.
+    let status = if CryptoProvider::get_default().is_some() || default_crypto_provider().install_default().is_err() {
+        CryptoProviderStatus::AlreadyInstalled
+    } else {
+        CryptoProviderStatus::Installed
+    };
+    #[cfg(feature = "fips")]
+    if !CryptoProvider::get_default().is_some_and(|provider| provider.fips()) {
+        return Err(CryptoProviderError { _private: () });
+    }
+    Ok(status)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{CryptoProviderStatus, default_crypto_provider, ensure_crypto_provider};
+
+    #[test]
+    fn provider_policy_isolated() {
+        const SCENARIO: &str = "RUSTFS_TLS_TEST_PROVIDER_SCENARIO";
+        if let Ok(scenario) = std::env::var(SCENARIO) {
+            match scenario.as_str() {
+                "fresh" => {
+                    assert_eq!(ensure_crypto_provider().expect("initial install"), CryptoProviderStatus::Installed);
+                    assert_eq!(ensure_crypto_provider().expect("provider reuse"), CryptoProviderStatus::AlreadyInstalled);
+                }
+                "incompatible" => {
+                    let mut provider = default_crypto_provider();
+                    provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+                    provider.install_default().expect("install incompatible provider");
+                    let result = ensure_crypto_provider();
+                    if cfg!(feature = "fips") {
+                        assert!(result.is_err(), "FIPS mode must reject the installed provider");
+                    } else {
+                        assert_eq!(result.expect("reuse existing provider"), CryptoProviderStatus::AlreadyInstalled);
+                    }
+                }
+                _ => panic!("unknown provider scenario"),
+            }
+            return;
+        }
+
+        for scenario in ["fresh", "incompatible"] {
+            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+                .arg("provider_policy_isolated")
+                .arg("--nocapture")
+                .env(SCENARIO, scenario)
+                .output()
+                .expect("start isolated provider test");
+            assert!(
+                output.status.success(),
+                "{scenario}: stdout={} stderr={}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
+    }
+}
@@ -33,10 +33,10 @@ zstd-safe = { version = "8.2.0", features = ["seekable"] }
 toml = "1.1.8"
 semver = "1.0.28"
 apache-avro.workspace = true
-arrow-array = "59.3.2"
-arrow-ipc = { version = "59.3.2", features = ["zstd"] }
-arrow-schema = "59.3.2"
-parquet = { version = "59.3.2", default-features = false, features = ["arrow", "zstd"] }
+arrow-array = "60.1.0"
+arrow-ipc = { version = "60.1.0", features = ["zstd"] }
+arrow-schema = "60.1.0"
+parquet = { version = "60.1.0", default-features = false, features = ["arrow", "zstd"] }
 kafka-protocol = { version = "0.18.0", default-features = false, features = ["zstd"] }
 bytes.workspace = true

@@ -64,7 +64,7 @@ RustFS S3 Tables is an Iceberg REST Catalog and table-bucket implementation on t
 | Namespaces | Supported | Create, list, load, exists, drop on both prefixes; `pageSize`/`pageToken` pagination with context-bound tokens; identifiers limited to 512 ASCII characters. |
 | Tables | Supported | Direct and staged create, register, list, load, exists, rename, commit, metadata-location get/update, and drop on both prefixes. Staged create is stateless until an `assert-create` commit atomically publishes the initial metadata and table entry. Rename uses a bucket-scoped persistent fence, recoverable intent, and conditional publication; the source name is reusable only via an ETag-conditional tombstone replacement. Commit identifiers must match the URL; unknown requirements/updates fail as bad requests; register overwrite, purge-on-drop, and v3-only encryption-key updates return an explicit unsupported-operation response. |
 | Commit CAS | Supported | Single-table commits validate base metadata, version token, referenced object existence, warehouse scope, and Iceberg requirements before advancing the pointer; external metadata transitions preserve monotonic assignment watermarks and immutable retained definitions. `idempotency-key-lifetime` is not advertised; mutation-wide `Idempotency-Key` semantics are unsupported. |
-| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. |
+| Commit recovery | Supported | Commit log, idempotency lookup, diagnostics, and recovery routes expose and repair finalization gaps without moving the pointer. Object-backed recovery reports a staged request as SUPERSEDED only when a reachable finalized same-base winner proves it lost CAS and its idempotency binding is intact. The original staged receipt is retained; its old request still conflicts. The winning history must be retained with the losing receipt; missing or contradictory history requires manual review. |
 | Snapshot refs | Supported | List, create/replace, delete via commits; `main` is protected; refs with explicit retention need forced delete. |
 | Iceberg views | Supported | Create, list, load, replace, exists, drop with view-scoped authorization; only view format version 1. |
 | LoadTable and table credentials endpoint | Supported | Vending only on negotiated `vended-credentials`; one temporary session scoped to the warehouse prefix and current metadata location; missing credential permission falls back to metadata-only with an explicit reason; responses carry `Cache-Control: no-store, private`. |
@@ -1,29 +1,29 @@
 # Cryptographic compliance positioning

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

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

 ## Status: not FIPS 140-3 validated

-**RustFS is not FIPS 140-3 (or 140-2) validated as a product, and in the default build no component it links runs as a validated cryptographic module.** There is no CMVP certificate covering RustFS itself, or the libraries it uses in the default shipped configuration. An opt-in `--features fips` build flag on `rustfs` switches the `rustls` / `aws-lc-rs` provider to `aws-lc-fips-sys` for TLS, X.509, and JWT operations, while object data-path and local KMS operations remain on unvalidated RustCrypto software implementations. Validation is a property of a specific module build, its documented boundary, and a certificate — none of which RustFS as a whole has or pursues.
+**RustFS is not FIPS 140-3 (or 140-2) validated as a product, and in the default build no component it links runs as a validated cryptographic module.** There is no CMVP certificate covering RustFS itself, or the libraries it uses in the default shipped configuration. An opt-in `--features fips` build flag on `rustfs` enables the `aws-lc-fips-sys` backend in the applicable dependency graph; it does not establish that every TLS, X.509, or JWT operation uses an approved service. Object data-path and local KMS operations remain on unvalidated RustCrypto software implementations. Validation is a property of a specific module build, its documented boundary, and a certificate — none of which RustFS as a whole has or pursues.

 ### What the process actually links

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

 Two consequences are commonly assumed the other way:

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

 ## Terminology red lines for external material

@@ -44,7 +44,7 @@ These rules apply to the README, CHANGELOG, release notes, marketing pages, sale

 Boilerplate when the topic cannot be avoided:

-> RustFS encrypts object data with AES-256-GCM and supports ChaCha20-Poly1305. These are FIPS-approved algorithms, but the implementations are not FIPS 140-3 validated cryptographic modules and RustFS makes no FIPS validation claim.
+> RustFS supports AES-256-GCM and ChaCha20-Poly1305 for object encryption. AES-256-GCM can be a FIPS-approved algorithm when used under the applicable conditions; ChaCha20-Poly1305 is not FIPS-approved. These implementations are not FIPS 140-3 validated cryptographic modules and RustFS makes no FIPS validation claim.

 ### Guard

@@ -52,13 +52,25 @@ Boilerplate when the topic cannot be avoided:

 ## The `rustfs` and `rustfs-tls-runtime` `fips` feature

-`rustfs/Cargo.toml` (`fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips"]`) and `crates/tls-runtime/Cargo.toml` (`fips = ["rustls/fips"]`) define an opt-in Cargo feature (disabled by default) for builds that require the `aws-lc-fips-sys` cryptographic module on the `aws-lc-rs` / `rustls` path:
+`rustfs/Cargo.toml` and the TLS runtime, protocols, targets, S3 client, and ECStore library manifests define opt-in `fips` features (disabled by default) for the `aws-lc-fips-sys` / `rustls` path. The RustFS feature propagates the policy to these libraries; when enabling a protocol library independently, also enable the required protocol feature.

- **What it enables:** Activates `rustls/fips` and `tokio-rustls/fips`, which pull in `aws-lc-rs/fips` (`aws-lc-fips-sys`) and `rustls-webpki/aws-lc-rs-fips`. Because Cargo unifies features across the dependency graph, other `aws-lc-rs` consumers in the binary (`jsonwebtoken`, `rcgen`, `google-cloud-auth`, `async-nats`) also link against `aws-lc-fips-sys`.
- **Runtime provider installation:** In `rustfs/src/startup_runtime_hooks.rs`, `build_default_crypto_provider()` selects `rustls::crypto::default_fips_provider()` (instead of `rustls::crypto::aws_lc_rs::default_provider()`) and asserts `provider.fips()` on startup, restricting TLS negotiation to FIPS-approved cipher suites (e.g., `TLS_AES_256_GCM_SHA384`, `TLS_AES_128_GCM_SHA256`, `TLS_ECDHE_*_AES_*_GCM_*`) and key exchange groups (excluding `TLS_CHACHA20_POLY1305_SHA256` and `X25519`).
- **Build requirements:** Compiling `aws-lc-fips-sys` requires CMake, Go, and Clang/`libclang` in the build environment (unlike the default `aws-lc-sys` build).
+Cargo's `--all-features` also enables `fips`, including for these individual libraries. Such builds inherit the FIPS backend's target and native-tool requirements; select supported feature combinations deliberately rather than assuming every all-features build is supported.
+
+```bash
+cargo build --release --locked -p rustfs --bin rustfs --features fips
+```
+
+- **What it enables:** Activates `rustls/fips` and `tokio-rustls/fips`, which pull in `aws-lc-rs/fips` (`aws-lc-fips-sys`) and `rustls-webpki/aws-lc-rs-fips`. Cargo unifies features for the same dependency version in a build graph; consumers of that `aws-lc-rs` instance use its FIPS backend, but this does not establish approval for every operation or cover independent dependency versions.
+- **Runtime provider installation:** `rustfs-tls-runtime` installs or reuses the process-wide provider and checks the actual provider in FIPS builds. RustFS performs this check before observability initialization in standalone and embedded startup. FTPS, WebDAV, transition clients, and relevant notification TLS clients also invoke the shared policy before using the process-wide provider. An incompatible host provider causes a configuration error. The check does not inspect clients that use their own provider or a different TLS implementation.
+- **Build requirements:** The FIPS backend requires a C/C++ toolchain, CMake, and Go; targets without suitable pre-generated bindings also require bindgen and Clang/`libclang`. Check the locked dependency version and target platform for their exact requirements.
 - **What it does not change:** It does not alter the object data-path AEAD (`crates/kms`, `crates/rio`, `crates/rio-v2`) or `rustfs-crypto` config/IAM blob encryption, which continue to use pure-Rust `RustCrypto` crates.

+The RustFS server TLS configuration uses the process-wide rustls provider. The Kafka notification client and remote replication SDK explicitly select AWS-LC, so their providers are independent of the process default; for the locked rustls version, the unified `rustls/fips` feature restricts that provider's suites and groups. ECStore's insecure replication-client path still uses the process default: an application using ECStore independently must directly depend on `rustfs-tls-runtime` with its `fips` feature and call `rustfs_tls_runtime::ensure_crypto_provider()` before constructing that client when it requires the FIPS policy. Skipping TLS certificate verification does not authenticate the remote target, regardless of provider selection. Feature unification applies only to the same dependency version in a build graph; independent builds and third-party TLS stacks require separate verification. No complete outbound TLS or platform validation matrix is asserted here.
+
+NATS can upgrade a connection to TLS based on the server's response, so the targets library checks the process provider before connecting in a FIPS build, including for a plaintext-looking NATS URL. Webhook targets construct a TLS-capable HTTP client even for an HTTP endpoint and also check the provider when built with `fips`. An incompatible host provider therefore prevents these initializations even if a particular connection would not use TLS.
+
+A provider check or linked FIPS backend does not establish CMVP validation for a deployment or approval of every operation. Verify the specific module version, certificate status, security policy, build, operating environment, and approved services against the [CMVP validated modules list](https://csrc.nist.gov/projects/cryptographic-module-validation-program/validated-modules). The feature is opt-in; before enabling it, test client algorithms, certificates, TLS versions, rolling upgrades, and rollback in the target environment.
+
 ## The `rustfs-crypto` `fips` feature: what it actually does

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

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

 ## Algorithm disablement and migration policy
@@ -14,7 +14,7 @@ Pick the lowest layer that can prove the change; add a higher-layer test only wh
 | Outbound target matrix | Replication of every object shape (empty, plain, retention, legal hold, multipart, checksummed, SSE-S3) against every remote-target failure mode the fake target models; an explicit expectation table pins known-red cells to an open issue | `python3 scripts/e2e_binary.py build`, then `python3 scripts/e2e_binary.py run -- cargo nextest run -p e2e_test -E 'test(/^replication_target_matrix_test::/)'` | With `e2e-repl-nightly`; required locally for any change to outbound client defaults (SOP: [`docs/postmortems/2026-09-03-replication-checksum-default-regression.md`](../postmortems/2026-09-03-replication-checksum-default-regression.md)) |
 | s3s-e2e conformance | External S3 conformance tool against a live server | `./scripts/e2e-run.sh ./target/debug/rustfs <data-dir>` | Non-documentation PRs, required through `Test and Lint` (second half of the `End-to-End Tests` job) |
 | S3 compatibility | `ceph/s3-tests` (boto3; allow-list `scripts/s3-tests/implemented_tests.txt`) and MinIO `mint` | `scripts/s3-tests/run.sh`; mint via `.github/workflows/mint.yml` | Non-documentation PRs: implemented and lifecycle S3 lanes required through `Test and Lint`; the separate s3-tests workflow provides additional PR reporting and a weekly full sweep; mint: weekly, report-only |
-| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos on storage-sensitive PRs and nightly via `e2e-distributed` |
+| Chaos / fault-injection | Single-node disk fault injection (`crates/e2e_test/src/chaos.rs`, `crates/e2e_test/src/fault_proxy.rs`) plus the 4-node kill/fresh-drive/blackhole cases in `crates/e2e_test/src/distributed/chaos_test.rs` | Part of the e2e crate (`e2e-reliability` and `e2e-distributed`) | Reliability cases with `e2e-full`; 4-node chaos nightly via `e2e-distributed` |
 | Fuzz | `cargo-fuzz` targets over untrusted parsing surfaces; isolated sub-workspace under `fuzz/` | `./scripts/fuzz/run.sh` (see [`fuzz/README.md`](../../fuzz/README.md)) | PR smoke on the paths listed in `.github/workflows/fuzz.yml`, plus nightly corpus |
 | Benchmarks | Criterion benches under each crate's `benches/` | `cargo bench -p <crate>` | On demand; never a gate |
 | Conditional reads | Paired ordinary/hit/miss GET latency, throughput and read-lock measurements | `scripts/run_conditional_read_abba.py` | Local conditional-read acceptance; [guide and scope](conditional-read-benchmark.md) |
@@ -88,7 +88,7 @@ All profiles are defined in `.config/nextest.toml`; its block comments hold the
 | `e2e-full` | Merge-queue / main-push single-node e2e lane |
 | `e2e-repl-nightly` | Nightly slow / cross-process replication lane |
 | `e2e-nightly` | Nightly serial multi-process cluster fault lane |
-| `e2e-distributed` | Storage-sensitive PR and nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
+| `e2e-distributed` | Nightly 4-node 4-disk S3 / lock / versioning / replication / quota / expand / decommission / rebalance / site-replication / chaos / upgrade (history + IAM AK/SK) lane |
 | `e2e-protocols` | Nightly fixed-port FTPS/SFTP/WebDAV lane, run with `-j 1` |

 Membership of each e2e profile is pinned by a digest in `.config/e2e-<profile>-selection.txt` and checked by `scripts/check_test_wiring.py --check-profile <profile>` before the lane runs. To list what a profile selects on your platform (the result is platform-dependent because some modules are linux-only):
@@ -35,7 +35,7 @@ gh api repos/rustfs/rustfs/rulesets/6436880 \
  --jq '.rules[] | select(.type == "required_status_checks") | .parameters'
 ```

-The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy.
+The aggregate requires the validation lanes already selected by `ci.yml`; this closes the gap where a failing critical lane left the required workspace check green. Independent workflows remain report-only unless separately required. Before adding a new expensive lane or moving existing PR coverage to a schedule, collect representative execution and regression evidence, establish ownership and a working scheduled replacement, and update this reference with the resulting policy. `e2e-distributed.yml` and `e2e-upgrade.yml` are that kind of replacement: the 4-node lane runs on its nightly schedule, and the single-node upgrade lane runs on its weekly Monday schedule, on version tags, and on manual dispatch. Neither is part of `Test and Lint`. Dispatch the workflow against a candidate branch when that signal is needed before merge.

 Superseded PR attempts are cancelled. A running main validation finishes, with only the newest pending main run retained, so frequent merges cannot continuously cancel the full baseline. Having a `merge_group` trigger does not itself require use of the merge queue.

@@ -66,7 +66,6 @@ The four site-replication state-writer concurrency proofs reserve the available
 | PR touching `paths` in `fuzz.yml` | `Build Fuzz Harness`, `Smoke / <target>` | `fuzz.yml` `fuzz-build`, `pr-fuzz-smoke` | Report-only | `MAX_TOTAL_TIME=60 ./scripts/fuzz/run.sh` |
 | PR touching `paths` in `windows-filesystem.yml` | `Rename Safety`, `Startup Smoke` | `windows-filesystem.yml` `rename-safety`, `startup-smoke` | Report-only | the `cargo test -p rustfs-ecstore --lib <filter>` commands in the job, on Windows; `cargo build --locked -p rustfs --bin rustfs`, then `rustfs.exe --version` and a single-volume server polled on `/health/ready` |
 | PR touching `paths` in `coverage.yml` | `Workspace line coverage` | `coverage.yml` `coverage` | Report-only | `make coverage`; `python3 scripts/check_security_coverage.py target/llvm-cov/coverage.json` |
-| PR touching `paths` in `e2e-upgrade.yml` | `Upgrade compatibility` | `e2e-upgrade.yml` `upgrade` | Report-only | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | PR touching `paths` in `oidc-keycloak.yml` | `OIDC Keycloak live gate` | `oidc-keycloak.yml` `oidc-keycloak-live` | Report-only | `cargo build --locked -p rustfs --bin rustfs`, then `bash scripts/test/oidc_keycloak_live.sh ./target/debug/rustfs` |
 | PR touching `paths` in `targets-integration.yml` | `PostgreSQL, MySQL, AMQP, and NATS` | `targets-integration.yml` `targets-live` | Report-only | start the containers as in the job, export the `RUSTFS_TEST_*` DSNs, then the job's `cargo test --locked -p rustfs-targets --test <name> -- --ignored --test-threads=1` commands |
 | PR, documentation-only selection | `Quick Checks`, `Typos`, `Test and Lint` | `ci.yml` `quick-checks`, `typos`, `required-checks` | Required directly or via aggregate | Quick Checks commands; `python3 scripts/ci_gate.py --self-test` |
@@ -90,7 +89,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `cache-warm.yml` (every six hours, plus main pushes that change `Cargo.lock`, `rust-toolchain.toml` or the setup action) | `warm-ci-dev`, `warm-ci-feat-rio`, `warm-ci-feat-proto`, `warm-ci-uring` | no verdict; sole writer of the `rust-cache` keys that `ci.yml` restores | yes | dispatch `cache-warm.yml` |
 | `build.yml` (nightly) | `build-rustfs` over the six-target platform matrix in `prepare-platform-matrix` (four Linux, macOS aarch64, Windows x86_64) | build/package integrity | yes | dispatch `build.yml` with an exact platform set |
 | `e2e-replication-nightly.yml` (nightly) | `repl-nightly`, `cluster-nightly`, `protocols-nightly` | three independent gates; JUnit, membership listing, server logs | yes | `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-repl-nightly -p e2e_test`; `--features e2e-test-hooks` on both steps with `--profile e2e-nightly`; `--features ftps,webdav,sftp` on both steps with `-j 1 --profile e2e-protocols` |
-| `e2e-distributed.yml` (storage-sensitive PRs + nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes, with `never_ran_grace_until` | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
+| `e2e-distributed.yml` (nightly) | `distributed` | fail-closed 4-node 4-disk S3, durability, replication, movement, fault, and direct/rolling upgrade gate; JUnit, membership listing, per-node server logs | yes | download the pinned previous release as in the workflow, export `RUSTFS_UPGRADE_SOURCE_BINARY`, run `python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks`, then `python3 scripts/e2e_binary.py run --features e2e-test-hooks -- cargo nextest run --profile e2e-distributed -p e2e_test` |
 | `e2e-s3tests.yml` (weekly) | `s3tests` (single and distributed, four shards each), `upstream-head-canary` | compatibility gate; report, JUnit, node IDs, server logs | yes | `scripts/s3-tests/run.sh` against an existing single or distributed target |
 | `fuzz.yml` (nightly) | `nightly-fuzz-corpus` per target | gate; corpus and crash artifacts | yes (disabled) | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
 | `minio-interop.yml` (nightly) | `minio-interop` | EC + SSE read-parity gate | yes, with `never_ran_grace_until` | pinned Docker fixture steps in the workflow |
@@ -100,7 +99,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `mint.yml` (weekly) | `mint` | report-only by design; per-suite PASS/FAIL/NA and raw `log.json` | yes | pinned Docker sequence in the workflow |
 | `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes (disabled) | `make coverage` |
 | `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
-| `e2e-upgrade.yml` (weekly) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | see the PR row |
+| `e2e-upgrade.yml` (weekly, and on version tags) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | yes | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | yes (disabled) | see the PR row |
 | `gateway-integration-sync.yml` (weekdays) | `sync` | merges `main` into `gateway/integration` as a `--no-ff` merge commit; a conflicting merge is aborted and opens a `sync/<source>-<date>` pull request listing the conflicting files, which keeps the run green; a rejected push or missing pull request fails it | yes | dispatch, optionally with `source_ref` |
@@ -58,7 +58,7 @@ required-features = ["swift"]

 [features]
 default = ["ftps", "webdav", "gcs"]
-fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips"]
+fips = ["rustls/fips", "tokio-rustls/fips", "rustfs-tls-runtime/fips", "rustfs-ecstore/fips", "rustfs-protocols/fips", "rustfs-s3-client/fips", "rustfs-targets/fips"]
 gcs = ["rustfs-ecstore/gcs", "dep:google-cloud-auth"]
 metrics-gpu = ["rustfs-obs/gpu"]
 ftps = ["rustfs-protocols/ftps"]
@@ -63,6 +63,9 @@ pub(crate) async fn init_startup_server_preflight(
    config: &Config,
    env_compat_report: &ExternalEnvCompatReport,
 ) -> std::result::Result<(), StartupServerPreflightError> {
+    #[cfg(feature = "fips")]
+    crate::startup_runtime_hooks::install_default_crypto_provider().map_err(StartupServerPreflightError::Other)?;
+
    crate::config::init_config_snapshot(config);
    init_license(config.license.clone());
    init_startup_observability(config.obs_endpoint.clone()).await?;
@@ -25,7 +25,7 @@ pub(crate) async fn init_startup_runtime_foundation(config: &Config) -> Result<(
    log_startup_runtime_diagnostics();
    init_profiling_runtime().await;
    rustfs_trusted_proxies::init();
-    install_default_crypto_provider();
+    install_default_crypto_provider()?;
    init_object_max_versions_config()?;
    init_outbound_tls_material(config).await
 }
@@ -14,10 +14,7 @@

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

-fn build_default_crypto_provider() -> rustls::crypto::CryptoProvider {
-    let provider = default_provider();
-    #[cfg(feature = "fips")]
-    assert!(
-        provider.fips(),
-        "FIPS crypto provider requested via `fips` feature, but AWS-LC FIPS mode is not active"
-    );
-    provider
-}
-
-pub(crate) fn install_default_crypto_provider() {
-    if build_default_crypto_provider().install_default().is_err() {
+pub(crate) fn install_default_crypto_provider() -> Result<()> {
+    if ensure_crypto_provider().map_err(std::io::Error::other)? == CryptoProviderStatus::AlreadyInstalled {
        debug!(
            target: "rustfs::main",
            event = EVENT_CRYPTO_PROVIDER_STATE,
            component = LOG_COMPONENT_MAIN,
            subsystem = LOG_SUBSYSTEM_STARTUP,
-            provider = "aws_lc_rs",
+            provider = "process_default",
            state = "already_installed",
            "Rustls crypto provider state checked"
        );
    }
+    Ok(())
 }

 pub(crate) async fn init_embedded_runtime_hooks(obs_endpoint: String) -> Result<()> {
+    install_embedded_default_crypto_provider()?;
+
    // Observability is a process-global tracing subscriber and can only be
    // installed once. A second embedded server in the same process reuses the
    // first server's subscriber — try to publish anyway (first server wins),
@@ -161,37 +151,228 @@ pub(crate) async fn init_embedded_runtime_hooks(obs_endpoint: String) -> Result<
        }
    }

-    install_embedded_default_crypto_provider();
    rustfs_trusted_proxies::init();

    Ok(())
 }

-fn install_embedded_default_crypto_provider() {
-    if let Err(err) = build_default_crypto_provider().install_default() {
+fn install_embedded_default_crypto_provider() -> Result<()> {
+    if ensure_crypto_provider().map_err(std::io::Error::other)? == CryptoProviderStatus::AlreadyInstalled {
        debug!(
+            event = EVENT_CRYPTO_PROVIDER_STATE,
            component = LOG_COMPONENT_EMBEDDED,
            subsystem = LOG_SUBSYSTEM_EMBEDDED,
-            event = EVENT_CRYPTO_PROVIDER_STATE,
            state = "already_installed",
-            error = ?err,
            "Embedded crypto provider state changed"
        );
    }
+    Ok(())
 }

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

+    #[cfg(feature = "fips")]
    #[test]
    fn default_crypto_provider_matches_fips_feature() {
        let provider = build_default_crypto_provider();
-        assert_eq!(provider.fips(), cfg!(feature = "fips"));
+        assert!(provider.fips());
+    }
+
+    #[cfg(feature = "fips")]
+    #[test]
+    fn fips_provider_builds_fips_tls_configs() {
+        let provider = Arc::new(build_default_crypto_provider());
+        assert!(provider.cipher_suites.iter().all(|suite| suite.fips()));
+        assert!(provider.kx_groups.iter().all(|group| group.fips()));
+        assert!(
+            !provider
+                .cipher_suites
+                .iter()
+                .any(|suite| { suite.suite() == rustls::CipherSuite::TLS13_CHACHA20_POLY1305_SHA256 })
+        );
+        assert!(
+            !provider
+                .kx_groups
+                .iter()
+                .any(|group| group.name() == rustls::NamedGroup::X25519)
+        );
+
+        let client = rustls::ClientConfig::builder_with_provider(provider.clone())
+            .with_safe_default_protocol_versions()
+            .expect("supported protocol versions")
+            .with_root_certificates(rustls::RootCertStore::empty())
+            .with_no_client_auth();
+        let server = rustls::ServerConfig::builder_with_provider(provider)
+            .with_safe_default_protocol_versions()
+            .expect("supported protocol versions")
+            .with_no_client_auth()
+            .with_cert_resolver(Arc::new(rustls::server::ResolvesServerCertUsingSni::new()));
+        assert!(client.fips());
+        assert!(server.fips());
+    }
+
+    #[cfg(feature = "fips")]
+    fn tls_handshake_with_provider(
+        client_provider: rustls::crypto::CryptoProvider,
+        version: &'static rustls::SupportedProtocolVersion,
+    ) -> std::result::Result<(), rustls::Error> {
+        let cert = rcgen::generate_simple_self_signed(vec!["localhost".to_owned()]).expect("generate certificate");
+        let cert_der = cert.cert.der().clone();
+        let private_key = rustls::pki_types::PrivatePkcs8KeyDer::from(cert.signing_key.serialize_der());
+        let mut roots = rustls::RootCertStore::empty();
+        roots.add(cert_der.clone()).expect("trust certificate");
+
+        let server_config = rustls::ServerConfig::builder_with_provider(Arc::new(build_default_crypto_provider()))
+            .with_protocol_versions(&[version])
+            .expect("server protocol version")
+            .with_no_client_auth()
+            .with_single_cert(vec![cert_der], private_key.into())
+            .expect("server identity");
+        assert!(server_config.fips());
+        let client_config = rustls::ClientConfig::builder_with_provider(Arc::new(client_provider))
+            .with_protocol_versions(&[version])
+            .expect("client protocol version")
+            .with_root_certificates(roots)
+            .with_no_client_auth();
+
+        let mut server = rustls::ServerConnection::new(Arc::new(server_config)).expect("server connection");
+        let mut client = rustls::ClientConnection::new(
+            Arc::new(client_config),
+            rustls::pki_types::ServerName::try_from("localhost").expect("server name"),
+        )
+        .expect("client connection");
+        for _ in 0..8 {
+            let mut client_bytes = Vec::new();
+            client.write_tls(&mut client_bytes).expect("write client TLS");
+            if !client_bytes.is_empty() {
+                server
+                    .read_tls(&mut std::io::Cursor::new(client_bytes))
+                    .expect("read server TLS");
+                server.process_new_packets()?;
+            }
+
+            let mut server_bytes = Vec::new();
+            server.write_tls(&mut server_bytes).expect("write server TLS");
+            if !server_bytes.is_empty() {
+                client
+                    .read_tls(&mut std::io::Cursor::new(server_bytes))
+                    .expect("read client TLS");
+                client.process_new_packets()?;
+            }
+            if !client.is_handshaking() && !server.is_handshaking() {
+                return Ok(());
+            }
+        }
+        panic!("TLS handshake did not complete within eight exchanges");
+    }
+
+    #[cfg(feature = "fips")]
+    #[test]
+    fn fips_provider_tls_handshake_accepts_and_rejects_algorithms() {
+        for version in [&rustls::version::TLS12, &rustls::version::TLS13] {
+            tls_handshake_with_provider(build_default_crypto_provider(), version).expect("FIPS TLS handshake");
+            let mut client_provider = build_default_crypto_provider();
+            client_provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+            assert!(tls_handshake_with_provider(client_provider, version).is_err());
+        }
+        let mut client_provider = build_default_crypto_provider();
+        client_provider.cipher_suites = vec![rustls::crypto::aws_lc_rs::cipher_suite::TLS13_CHACHA20_POLY1305_SHA256];
+        assert!(tls_handshake_with_provider(client_provider, &rustls::version::TLS13).is_err());
+    }
+
+    #[test]
+    fn provider_installation_isolated() {
+        const SCENARIO: &str = "RUSTFS_TEST_PROVIDER_SCENARIO";
+        if let Ok(scenario) = std::env::var(SCENARIO) {
+            let incompatible = || {
+                let mut provider = build_default_crypto_provider();
+                provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
+                assert!(!provider.fips());
+                provider
+            };
+            match scenario.as_str() {
+                "fresh" => {}
+                "compatible" => {
+                    build_default_crypto_provider().install_default().expect("install provider");
+                }
+                "incompatible" => {
+                    incompatible().install_default().expect("install provider");
+                }
+                "race" => {
+                    let barrier = Arc::new(std::sync::Barrier::new(2));
+                    let other_barrier = barrier.clone();
+                    let other = std::thread::spawn(move || {
+                        other_barrier.wait();
+                        let _ = incompatible().install_default();
+                    });
+                    barrier.wait();
+                    let result = install_default_crypto_provider();
+                    other.join().expect("provider installer thread");
+                    let actual_fips = rustls::crypto::CryptoProvider::get_default()
+                        .expect("provider installed")
+                        .fips();
+                    assert_eq!(result.is_ok(), !cfg!(feature = "fips") || actual_fips);
+                    assert_eq!(install_embedded_default_crypto_provider().is_ok(), !cfg!(feature = "fips") || actual_fips);
+                    return;
+                }
+                _ => panic!("unknown provider scenario"),
+            }
+            let result = install_default_crypto_provider();
+            let actual_fips = rustls::crypto::CryptoProvider::get_default()
+                .expect("provider installed")
+                .fips();
+            assert_eq!(result.is_ok(), !cfg!(feature = "fips") || actual_fips);
+            if cfg!(feature = "fips") && scenario == "incompatible" {
+                assert!(
+                    result
+                        .unwrap_err()
+                        .to_string()
+                        .contains("process-wide rustls crypto provider")
+                );
+                let runtime = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .expect("embedded runtime");
+                let error = runtime
+                    .block_on(super::init_embedded_runtime_hooks(String::new()))
+                    .unwrap_err();
+                assert!(error.to_string().contains("process-wide rustls crypto provider"));
+                let config = crate::config::Config::new("127.0.0.1:9000", Vec::new());
+                let report = rustfs_utils::ExternalEnvCompatReport::default();
+                let error = runtime
+                    .block_on(crate::startup_preflight::init_startup_server_preflight(&config, &report))
+                    .unwrap_err();
+                assert!(error.to_string().contains("process-wide rustls crypto provider"));
+                assert!(rustfs_obs::get_global_guard().is_err(), "observability must not initialize first");
+            }
+            assert_eq!(install_embedded_default_crypto_provider().is_ok(), !cfg!(feature = "fips") || actual_fips);
+            return;
+        }
+
+        for scenario in ["fresh", "compatible", "incompatible", "race"] {
+            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+                .arg("provider_installation_isolated")
+                .arg("--nocapture")
+                .env(SCENARIO, scenario)
+                .output()
+                .expect("start isolated test");
+            assert!(
+                output.status.success(),
+                "{scenario}: stdout={} stderr={}",
+                String::from_utf8_lossy(&output.stdout),
+                String::from_utf8_lossy(&output.stderr)
+            );
+        }
    }

    #[tokio::test]
@@ -44,45 +44,58 @@ pub(crate) fn table_matches_staged_base(table: &TableEntry, commit_log: &CommitL
 pub(crate) struct TableCommitHistoryIndex<'a> {
    table_id: &'a str,
    reachable_states: BTreeSet<(&'a str, &'a str)>,
+    reachable_version_tokens: BTreeSet<&'a str>,
+    committed_successors: BTreeMap<(&'a str, &'a str), &'a CommitLogEntry>,
    ambiguous_states: BTreeSet<(&'a str, &'a str)>,
    cycle_detected: bool,
+    version_token_reused: bool,
 }

 impl<'a> TableCommitHistoryIndex<'a> {
    pub(crate) fn new(table: &'a TableEntry, commits: impl IntoIterator<Item = &'a CommitLogEntry>) -> Self {
-        let mut by_new_state = BTreeMap::<(&str, &str), Option<(&str, &str)>>::new();
+        let mut by_new_state = BTreeMap::<(&str, &str), Option<&CommitLogEntry>>::new();
        let mut ambiguous_states = BTreeSet::new();
        for commit in commits
            .into_iter()
            .filter(|commit| commit.table_id == table.table_id && !matches!(commit.status, CommitLogStatus::Failed))
        {
            let key = (commit.new_metadata_location.as_str(), commit.new_version_token.as_str());
-            let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
            by_new_state
                .entry(key)
                .and_modify(|candidate| {
                    *candidate = None;
                    ambiguous_states.insert(key);
                })
-                .or_insert(Some(previous));
+                .or_insert(Some(commit));
        }

        let mut reachable_states = BTreeSet::new();
+        let mut reachable_version_tokens = BTreeSet::new();
+        let mut committed_successors = BTreeMap::new();
+        let mut version_token_reused = false;
        let mut state = (table.metadata_location.as_str(), table.version_token.as_str());
        let cycle_detected = loop {
            if !reachable_states.insert(state) {
                break true;
            }
-            let Some(Some(previous)) = by_new_state.get(&state) else {
+            version_token_reused |= !reachable_version_tokens.insert(state.1);
+            let Some(commit) = by_new_state.get(&state).copied().flatten() else {
                break false;
            };
-            state = *previous;
+            let previous = (commit.previous_metadata_location.as_str(), commit.expected_version_token.as_str());
+            if matches!(commit.status, CommitLogStatus::Committed) {
+                committed_successors.insert(previous, commit);
+            }
+            state = previous;
        };
        Self {
            table_id: &table.table_id,
            reachable_states,
+            reachable_version_tokens,
+            committed_successors,
            ambiguous_states,
            cycle_detected,
+            version_token_reused,
        }
    }

@@ -97,6 +110,49 @@ impl<'a> TableCommitHistoryIndex<'a> {
                .reachable_states
                .contains(&(target.new_metadata_location.as_str(), target.new_version_token.as_str()))
    }
+
+    fn proven_superseding_commit(&self, target: &CommitLogEntry) -> Option<&CommitLogEntry> {
+        if self.cycle_detected
+            || self.version_token_reused
+            || !self.ambiguous_states.is_empty()
+            || self.table_id != target.table_id
+            || !matches!(target.status, CommitLogStatus::Staged)
+            || self.reachable_version_tokens.contains(target.new_version_token.as_str())
+        {
+            return None;
+        }
+        let base = (target.previous_metadata_location.as_str(), target.expected_version_token.as_str());
+        self.committed_successors
+            .get(&base)
+            .copied()
+            .filter(|winner| winner.commit_id != target.commit_id)
+    }
+}
+
+/// Classifies an object-backed staged loser without rewriting its receipt.
+///
+/// A reachable finalized successor from the same pointer/token base proves that
+/// the old request lost CAS. Its STAGED record and idempotency binding remain
+/// available, and replay of that old request must still conflict. Missing or
+/// contradictory proof remains a manual-review state.
+pub(crate) fn table_commit_recovery_entry_from_history(
+    table: &TableEntry,
+    commit_log: &CommitLogEntry,
+    idempotency_commit: Option<&CommitLogEntry>,
+    history: &TableCommitHistoryIndex<'_>,
+) -> TableCommitRecoveryEntry {
+    let mut entry = table_commit_recovery_entry(table, commit_log, idempotency_commit, history.proves_committed(commit_log));
+    if entry.recovery_state == TableCommitRecoveryState::ManualReview
+        && matches!(
+            entry.idempotency_index_status,
+            TableCommitIdempotencyIndexStatus::NotRequired | TableCommitIdempotencyIndexStatus::Matches
+        )
+        && let Some(winner) = history.proven_superseding_commit(commit_log)
+    {
+        entry.recovery_state = TableCommitRecoveryState::Superseded;
+        entry.reason = format!("staged request lost CAS to reachable committed successor {}", winner.commit_id);
+    }
+    entry
 }

 pub(crate) fn table_catalog_recovery_summary(
@@ -1415,6 +1415,7 @@ pub(crate) struct TableCatalogDiagnosticsReport {
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub(crate) enum TableCommitRecoveryState {
    Committed,
+    Superseded,
    StagedBeforeTableUpdate,
    FinalizationRequired,
    IdempotencyIndexRepairRequired,
@@ -3065,11 +3065,11 @@ where
                }
                None => None,
            };
-            commits.push(table_commit_recovery_entry(
+            commits.push(table_commit_recovery_entry_from_history(
                entry,
                commit_log,
                idempotency_commit.as_ref(),
-                history.proves_committed(commit_log),
+                &history,
            ));
        }
        commits.sort_by(|left, right| left.commit_id.cmp(&right.commit_id));
@@ -3168,12 +3168,8 @@ where
                }
                None => None,
            };
-            let recovery_entry = table_commit_recovery_entry(
-                &entry,
-                commit_log,
-                idempotency_commit.as_ref(),
-                history.proves_committed(commit_log),
-            );
+            let recovery_entry =
+                table_commit_recovery_entry_from_history(&entry, commit_log, idempotency_commit.as_ref(), &history);
            if matches!(
                recovery_entry.recovery_state,
                TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired
@@ -20,6 +20,7 @@ use std::sync::Arc;

 mod strong_maintenance_planning;
 mod strong_maintenance_planning_paths;
+mod superseded_commit_recovery;

 const TABLE_CATALOG_TEST_TIMEOUT: StdDuration = StdDuration::from_secs(30);

@@ -0,0 +1,487 @@
+// Copyright 2026 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.
+
+use super::*;
+
+struct StagedLoserFixture {
+    backend: TestCatalogObjectBackend,
+    store: ObjectTableCatalogStore<TestCatalogObjectBackend>,
+    loser: TableCommitRequest,
+    winner: TableCommitRequest,
+    committed: TableCommitResult,
+}
+
+async fn staged_loser_fixture(interrupt_winner_finalization: bool) -> StagedLoserFixture {
+    let backend = TestCatalogObjectBackend::default();
+    let store = ObjectTableCatalogStore::new(backend.clone());
+    let bucket = "analytics";
+    let namespace = Namespace::parse("sales").unwrap();
+    let table = IdentifierSegment::parse("orders").unwrap();
+    let initial = test_table_entry(
+        bucket,
+        &namespace,
+        &table,
+        default_table_metadata_file_path(&namespace, &table, "00001.metadata.json"),
+    );
+    store.put_table_bucket(test_bucket_entry(bucket)).await.unwrap();
+    store
+        .create_namespace(test_namespace_entry(bucket, &namespace))
+        .await
+        .unwrap();
+    store.create_table(initial.clone()).await.unwrap();
+    for name in ["00001", "loser", "winner", "reloaded"] {
+        let location = default_table_metadata_file_path(&namespace, &table, &format!("{name}.metadata.json"));
+        let mut metadata = super::super::test_support::table_metadata_json(&initial.table_uuid, &initial.warehouse_location);
+        metadata["properties"] = serde_json::json!({"writer": name});
+        backend
+            .seed_object(bucket, &location, serde_json::to_vec(&metadata).unwrap())
+            .await;
+    }
+    let paths = TableCatalogObjectPaths::default();
+    let table_path = paths.table_entry_path(bucket, &namespace, &table);
+    backend.fail_next_put(RUSTFS_META_BUCKET, &table_path).await;
+    let loser = TableCommitRequest {
+        table_bucket: bucket.to_string(),
+        namespace: namespace.public_name(),
+        table: table.as_str().to_string(),
+        commit_id: "paused-writer-a0".to_string(),
+        idempotency_key: Some("a0-request".to_string()),
+        operation: "set-properties".to_string(),
+        expected_version_token: initial.version_token.clone(),
+        expected_metadata_location: initial.metadata_location.clone(),
+        new_metadata_location: default_table_metadata_file_path(&namespace, &table, "loser.metadata.json"),
+        requirements: Vec::new(),
+        writer: None,
+    };
+    assert_matches!(store.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Internal(_)));
+    let winner = TableCommitRequest {
+        commit_id: "winner-c1".to_string(),
+        idempotency_key: Some("c1-request".to_string()),
+        new_metadata_location: default_table_metadata_file_path(&namespace, &table, "winner.metadata.json"),
+        ..loser.clone()
+    };
+    if interrupt_winner_finalization {
+        let winner_log = paths.commit_log_entry_path(bucket, &initial.table_id, &winner.commit_id);
+        backend.fail_put_attempt(RUSTFS_META_BUCKET, &winner_log, 2).await;
+    }
+    let committed = store.commit_table(winner.clone()).await.unwrap();
+    StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    }
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_recovers_without_rewriting_receipts_or_pointer() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    } = staged_loser_fixture(false).await;
+    let bucket = loser.table_bucket.as_str();
+    let before = backend.state.lock().await.objects.clone();
+
+    let recovered = store.recover_table_commits(bucket, "sales", "orders").await.unwrap();
+    assert_eq!(recovered.manual_review_count, 0, "a durable same-base winner proves the staged loser");
+    assert_eq!(recovered.finalized_count, 0, "a losing writer must never be finalized as committed");
+    let lost = recovered
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(lost.status, CommitLogStatus::Staged);
+    assert_eq!(serde_json::to_value(&lost.recovery_state).unwrap(), "SUPERSEDED");
+    assert_eq!(lost.idempotency_index_status, TableCommitIdempotencyIndexStatus::Matches);
+    let after = backend.state.lock().await.objects.clone();
+    assert_eq!(after.len(), before.len());
+    for (path, original) in &before {
+        let actual = after.get(path).unwrap();
+        assert_eq!(actual.data, original.data, "recovery changed bytes at {path:?}");
+        assert_eq!(actual.etag, original.etag, "recovery rewrote {path:?}");
+    }
+
+    let restarted = ObjectTableCatalogStore::new(backend.clone());
+    let diagnostics = restarted.diagnose_table_catalog(bucket, "sales", "orders", 0).await.unwrap();
+    assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::Healthy);
+    assert_matches!(restarted.commit_table(loser.clone()).await, Err(TableCatalogStoreError::Conflict(_)));
+    let mut stale_request = loser.clone();
+    stale_request.commit_id = "stale-new-request".to_string();
+    stale_request.idempotency_key = Some("stale-new-key".to_string());
+    assert_matches!(restarted.commit_table(stale_request).await, Err(TableCatalogStoreError::Conflict(_)));
+    let mut different_payload = loser.clone();
+    different_payload.operation = "replace".to_string();
+    assert_matches!(
+        restarted.commit_table(different_payload.clone()).await,
+        Err(TableCatalogStoreError::Conflict(_))
+    );
+    assert_eq!(
+        restarted.load_table(bucket, "sales", "orders").await.unwrap(),
+        Some(committed.table.clone())
+    );
+    assert_eq!(restarted.commit_table(winner).await.unwrap().table, committed.table);
+
+    let reloaded = TableCommitRequest {
+        commit_id: "reloaded-writer-a1".to_string(),
+        idempotency_key: Some("a1-request".to_string()),
+        expected_version_token: committed.table.version_token.clone(),
+        expected_metadata_location: committed.table.metadata_location.clone(),
+        new_metadata_location: default_table_metadata_file_path(
+            &Namespace::parse("sales").unwrap(),
+            &IdentifierSegment::parse("orders").unwrap(),
+            "reloaded.metadata.json",
+        ),
+        ..different_payload
+    };
+    let next = restarted.commit_table(reloaded.clone()).await.unwrap();
+    assert_eq!(next.table.generation, committed.table.generation + 1);
+    assert_eq!(restarted.commit_table(reloaded).await.unwrap().table, next.table);
+    let restarted = ObjectTableCatalogStore::new(backend);
+    assert_eq!(
+        restarted
+            .diagnose_table_catalog(bucket, "sales", "orders", 0)
+            .await
+            .unwrap()
+            .recovery_status,
+        TableCatalogRecoveryStatus::Healthy
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_waits_for_durable_winner_finalization() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        committed,
+        ..
+    } = staged_loser_fixture(true).await;
+    let paths = TableCatalogObjectPaths::default();
+    let log_path = paths.commit_log_entry_path(&loser.table_bucket, &committed.table.table_id, &loser.commit_id);
+    let before = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
+    let planned = store
+        .plan_table_commit_recovery("analytics", "sales", "orders")
+        .await
+        .unwrap();
+    assert_eq!(planned.manual_review_count, 1, "a merely staged winner is insufficient proof");
+    assert_eq!(planned.finalization_required_count, 1);
+
+    let recovered = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(recovered.finalized_count, 1);
+    assert_eq!(recovered.manual_review_count, 0);
+    let loser_state = recovered
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::Superseded);
+    assert_eq!(loser_state.status, CommitLogStatus::Staged);
+    let after = backend.read_object(RUSTFS_META_BUCKET, &log_path).await.unwrap().unwrap();
+    assert_eq!(after.data, before.data);
+    assert_eq!(after.etag, before.etag);
+    assert_eq!(store.load_table("analytics", "sales", "orders").await.unwrap(), Some(committed.table));
+    assert_eq!(
+        store
+            .diagnose_table_catalog("analytics", "sales", "orders", 0)
+            .await
+            .unwrap()
+            .recovery_status,
+        TableCatalogRecoveryStatus::Healthy
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_requires_retained_winner_proof_after_restart() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        winner,
+        committed,
+    } = staged_loser_fixture(false).await;
+    let healthy = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(healthy.manual_review_count, 0);
+    let winner_path =
+        TableCatalogObjectPaths::default().commit_log_entry_path("analytics", &committed.table.table_id, &winner.commit_id);
+    assert!(
+        backend
+            .state
+            .lock()
+            .await
+            .objects
+            .remove(&(RUSTFS_META_BUCKET.to_string(), winner_path))
+            .is_some()
+    );
+    let restarted = ObjectTableCatalogStore::new(backend);
+    let before = restarted.load_table("analytics", "sales", "orders").await.unwrap();
+    let report = restarted.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(report.manual_review_count, 1);
+    assert_eq!(report.finalized_count, 0);
+    assert_eq!(report.commits[0].recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(report.commits[0].status, CommitLogStatus::Staged);
+    assert_matches!(restarted.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
+    assert_eq!(restarted.load_table("analytics", "sales", "orders").await.unwrap(), before);
+    let diagnostics = restarted
+        .diagnose_table_catalog("analytics", "sales", "orders", 0)
+        .await
+        .unwrap();
+    assert_eq!(diagnostics.recovery_status, TableCatalogRecoveryStatus::ManualReviewRequired);
+    assert!(
+        diagnostics
+            .recommended_actions
+            .contains(&TableCatalogRecoveryAction::ReviewCommitLog)
+    );
+}
+
+#[tokio::test]
+async fn superseded_staged_commit_does_not_repair_a_conflicting_index() {
+    let StagedLoserFixture {
+        backend,
+        store,
+        loser,
+        committed,
+        ..
+    } = staged_loser_fixture(false).await;
+    let mut conflicting = store
+        .get_commit_by_id("analytics", &committed.table.table_id, &loser.commit_id)
+        .await
+        .unwrap()
+        .unwrap();
+    conflicting.new_version_token = "contradictory-token".to_string();
+    let path = TableCatalogObjectPaths::default().commit_idempotency_entry_path(
+        "analytics",
+        &committed.table.table_id,
+        loser.idempotency_key.as_deref().unwrap(),
+    );
+    backend
+        .seed_object(RUSTFS_META_BUCKET, &path, serde_json::to_vec(&conflicting).unwrap())
+        .await;
+    let before = backend.state.lock().await.objects.clone();
+    let report = store.recover_table_commits("analytics", "sales", "orders").await.unwrap();
+    assert_eq!(report.manual_review_count, 1);
+    assert_eq!(report.finalized_count, 0);
+    let loser_state = report
+        .commits
+        .iter()
+        .find(|entry| entry.commit_id == loser.commit_id)
+        .unwrap();
+    assert_eq!(loser_state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(loser_state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
+    let after = backend.state.lock().await.objects.clone();
+    assert_eq!(before.len(), after.len());
+    for (key, original) in &before {
+        let actual = after.get(key).unwrap();
+        assert_eq!(actual.data, original.data);
+        assert_eq!(actual.etag, original.etag);
+    }
+    assert_matches!(store.commit_table(loser).await, Err(TableCatalogStoreError::Conflict(_)));
+}
+
+fn history_fixture() -> (TableEntry, CommitLogEntry, CommitLogEntry) {
+    let namespace = Namespace::parse("sales").unwrap();
+    let table_name = IdentifierSegment::parse("orders").unwrap();
+    let metadata = |name| default_table_metadata_file_path(&namespace, &table_name, name);
+    let mut table = test_table_entry("analytics", &namespace, &table_name, metadata("winner.metadata.json"));
+    table.version_token = "winner-token".to_string();
+    table.generation = 2;
+    let loser = CommitLogEntry {
+        version: TABLE_CATALOG_ENTRY_VERSION,
+        commit_id: "loser".to_string(),
+        idempotency_key: None,
+        table_id: table.table_id.clone(),
+        operation: "set-properties".to_string(),
+        expected_version_token: "base-token".to_string(),
+        new_version_token: "loser-token".to_string(),
+        previous_metadata_location: metadata("base.metadata.json"),
+        new_metadata_location: metadata("loser.metadata.json"),
+        requirements: Vec::new(),
+        status: CommitLogStatus::Staged,
+        writer: None,
+        created_at: None,
+        updated_at: None,
+    };
+    let winner = CommitLogEntry {
+        commit_id: "winner".to_string(),
+        new_version_token: table.version_token.clone(),
+        new_metadata_location: table.metadata_location.clone(),
+        status: CommitLogStatus::Committed,
+        ..loser.clone()
+    };
+    (table, loser, winner)
+}
+
+fn classify_history(
+    table: &TableEntry,
+    target: &CommitLogEntry,
+    commits: &[CommitLogEntry],
+    indexed: Option<&CommitLogEntry>,
+) -> TableCommitRecoveryEntry {
+    let history = TableCommitHistoryIndex::new(table, commits.iter());
+    table_commit_recovery_entry_from_history(table, target, indexed, &history)
+}
+
+#[test]
+fn superseded_history_requires_a_reachable_finalized_same_base_winner() {
+    for case in [
+        "staged-winner",
+        "failed-winner",
+        "wrong-base-token",
+        "wrong-base-location",
+        "foreign-table",
+        "disconnected-winner",
+        "same-commit-id",
+    ] {
+        let (mut table, loser, mut winner) = history_fixture();
+        match case {
+            "staged-winner" => winner.status = CommitLogStatus::Staged,
+            "failed-winner" => winner.status = CommitLogStatus::Failed,
+            "wrong-base-token" => winner.expected_version_token = "other-base-token".to_string(),
+            "wrong-base-location" => winner.previous_metadata_location = "other/base.metadata.json".to_string(),
+            "foreign-table" => winner.table_id = "other-table-id".to_string(),
+            "disconnected-winner" => {
+                table.metadata_location = "other/current.metadata.json".to_string();
+                table.version_token = "other-current-token".to_string();
+            }
+            "same-commit-id" => winner.commit_id.clone_from(&loser.commit_id),
+            _ => unreachable!(),
+        }
+        let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
+        assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview, "{case}");
+    }
+}
+
+#[test]
+fn superseded_history_is_stable_after_later_commits() {
+    let (mut table, loser, winner) = history_fixture();
+    let immediate = classify_history(&table, &loser, &[loser.clone(), winner.clone()], None);
+    assert_eq!(immediate.recovery_state, TableCommitRecoveryState::Superseded);
+    assert!(immediate.reason.contains(&winner.commit_id));
+    let later = CommitLogEntry {
+        commit_id: "later".to_string(),
+        previous_metadata_location: table.metadata_location.clone(),
+        expected_version_token: table.version_token.clone(),
+        new_metadata_location: "later/current.metadata.json".to_string(),
+        new_version_token: "later-token".to_string(),
+        ..winner.clone()
+    };
+    table.metadata_location.clone_from(&later.new_metadata_location);
+    table.version_token.clone_from(&later.new_version_token);
+    table.generation += 1;
+    let after = classify_history(&table, &loser, &[loser.clone(), winner, later], None);
+    assert_eq!(after.recovery_state, TableCommitRecoveryState::Superseded);
+    assert_eq!(after.reason, immediate.reason);
+}
+
+#[test]
+fn superseded_history_rejects_failed_or_committed_losers() {
+    for status in [CommitLogStatus::Failed, CommitLogStatus::Committed] {
+        let (table, mut loser, winner) = history_fixture();
+        loser.status = status;
+        let recovery = classify_history(&table, &loser, &[loser.clone(), winner], None);
+        assert_eq!(recovery.recovery_state, TableCommitRecoveryState::ManualReview);
+    }
+}
+
+#[test]
+fn superseded_history_rejects_ambiguity_cycles_and_reused_tokens() {
+    for case in [
+        "ambiguous-current",
+        "cycle",
+        "reused-history-token",
+        "loser-token-in-history",
+        "off-history-ambiguity",
+    ] {
+        let (mut table, mut loser, winner) = history_fixture();
+        let mut extra = CommitLogEntry {
+            commit_id: "extra".to_string(),
+            ..winner.clone()
+        };
+        let mut commits = vec![loser.clone(), winner.clone()];
+        match case {
+            "ambiguous-current" => commits.push(extra),
+            "cycle" => {
+                extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
+                extra.expected_version_token.clone_from(&winner.new_version_token);
+                extra.new_metadata_location.clone_from(&winner.previous_metadata_location);
+                extra.new_version_token.clone_from(&winner.expected_version_token);
+                commits.push(extra);
+            }
+            "reused-history-token" => {
+                extra.previous_metadata_location.clone_from(&winner.new_metadata_location);
+                extra.expected_version_token.clone_from(&winner.new_version_token);
+                extra.new_metadata_location = "later/current.metadata.json".to_string();
+                extra.new_version_token.clone_from(&winner.expected_version_token);
+                table.metadata_location.clone_from(&extra.new_metadata_location);
+                table.version_token.clone_from(&extra.new_version_token);
+                commits.push(extra);
+            }
+            "loser-token-in-history" => {
+                loser.new_version_token.clone_from(&winner.expected_version_token);
+                commits[0] = loser.clone();
+            }
+            "off-history-ambiguity" => {
+                extra.new_metadata_location = "unknown/metadata.json".to_string();
+                extra.new_version_token = "unknown-token".to_string();
+                commits.push(extra.clone());
+                extra.commit_id = "another-extra".to_string();
+                commits.push(extra);
+            }
+            _ => unreachable!(),
+        }
+        assert_eq!(
+            classify_history(&table, &loser, &commits, None).recovery_state,
+            TableCommitRecoveryState::ManualReview,
+            "{case}"
+        );
+    }
+}
+
+#[test]
+fn superseded_history_requires_an_intact_idempotency_binding() {
+    let (table, mut loser, winner) = history_fixture();
+    loser.idempotency_key = Some("loser-request".to_string());
+    let commits = [loser.clone(), winner];
+    assert_eq!(
+        classify_history(&table, &loser, &commits, Some(&loser)).recovery_state,
+        TableCommitRecoveryState::Superseded
+    );
+    let missing = classify_history(&table, &loser, &commits, None);
+    assert_eq!(missing.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(missing.idempotency_index_status, TableCommitIdempotencyIndexStatus::Missing);
+    let mut stale = loser.clone();
+    stale.status = CommitLogStatus::Committed;
+    let state = classify_history(&table, &loser, &commits, Some(&stale));
+    assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Stale);
+    let mut conflicting = loser.clone();
+    conflicting.commit_id = "another-request".to_string();
+    let state = classify_history(&table, &loser, &commits, Some(&conflicting));
+    assert_eq!(state.recovery_state, TableCommitRecoveryState::ManualReview);
+    assert_eq!(state.idempotency_index_status, TableCommitIdempotencyIndexStatus::Conflicting);
+}
+
+#[test]
+fn superseded_history_never_relabels_a_committed_staged_ancestor() {
+    let (table, mut ancestor, winner) = history_fixture();
+    ancestor.new_metadata_location.clone_from(&winner.previous_metadata_location);
+    ancestor.new_version_token.clone_from(&winner.expected_version_token);
+    ancestor.previous_metadata_location = "older/base.metadata.json".to_string();
+    ancestor.expected_version_token = "older-base-token".to_string();
+    let recovery = classify_history(&table, &ancestor, &[ancestor.clone(), winner], None);
+    assert_eq!(recovery.recovery_state, TableCommitRecoveryState::FinalizationRequired);
+}
@@ -16,7 +16,7 @@ import unittest
 ROOT = Path(__file__).resolve().parent.parent
 ALWAYS_JOBS = ("classify-changes", "typos", "quick-checks")
 CODE_JOBS = (
-    "workspace-lint-and-build", "workspace-tests", "core-contract", "connect-feature-boundaries", "test-and-lint-rio-v2",
+    "workspace-lint-and-build", "workspace-tests", "core-contract", "connect-feature-boundaries", "fips-provider", "test-and-lint-rio-v2",
    "test-and-lint-protocols", "build-rustfs-debug-binary", "uring-integration",
    "e2e-tests", "s3-implemented-tests", "s3-lifecycle-behavior-tests",
 )
@@ -28,12 +28,12 @@ CODE_JOBS = (
 # wall). Workspace lint and tests, the short core contract, the debug build,
 # E2E smoke and the S3 lanes stay on the pull-request path because they cover
 # the product's core correctness and S3 compatibility surface.
-NIGHTLY_JOBS = ("connect-feature-boundaries", "uring-integration", "test-and-lint-rio-v2", "test-and-lint-protocols")
+NIGHTLY_JOBS = ("connect-feature-boundaries", "fips-provider", "uring-integration", "test-and-lint-rio-v2", "test-and-lint-protocols")
 # Bump when a plan produced by an older `select` can no longer be checked
 # against this revision's expected job contract. The workflow refuses a plan
 # that does not carry the current marker and falls back to the full matrix, so
 # a base planner predating the contract cannot authorize a narrower selection.
-PLAN_CONTRACT = "2"
+PLAN_CONTRACT = "3"
 # These jobs still exercise the server and its black-box harness when only E2E
 # Rust sources change. Production code, manifests and shared test configuration
 # always select the full matrix. No production package depends on e2e_test.
@@ -210,6 +210,17 @@ def configurations(jobs: dict[str, str]) -> list[dict]:
        ("connect-offline-tests", "connect-feature-boundaries", "test", ["offline-enrollment-e2e-root"]),
        ("connect-release", "connect-feature-boundaries", "release", []),
        ("connect-release-rejection", "connect-feature-boundaries", "release", ["connect-e2e-short-credentials"]),
+        ("fips-provider-default-lint", "fips-provider", "dev", []),
+        ("fips-provider-fips-lint", "fips-provider", "dev", ["fips"]),
+        ("fips-tls-runtime-default", "fips-provider", "test", []),
+        ("fips-tls-runtime-fips", "fips-provider", "test", ["fips"]),
+        ("fips-protocols", "fips-provider", "test", ["ftps", "webdav", "fips"]),
+        ("fips-targets", "fips-provider", "test", ["fips"]),
+        ("fips-s3-client", "fips-provider", "test", ["fips"]),
+        ("fips-targets-default", "fips-provider", "test", []),
+        ("fips-s3-client-default", "fips-provider", "test", []),
+        ("fips-provider-default", "fips-provider", "test", []),
+        ("fips-provider-fips", "fips-provider", "test", ["fips"]),
        ("rio-v2-lint", "test-and-lint-rio-v2", "dev", ["rio-v2"]),
        ("rio-v2-tests", "test-and-lint-rio-v2", "test", ["rio-v2"]),
        ("protocol-swift-lint", "test-and-lint-protocols", "dev", ["swift"]),
@@ -234,7 +234,7 @@ class PlannerTests(unittest.TestCase):
        self.assertEqual(result.returncode, 0, result.stderr)
        values = dict(line.split('=', 1) for line in output.read_text().splitlines())
        self.assertEqual(set(values), {'mode', 'plan', 'connect', 'packages', 'full_validation', 'rehearse_shards', 'plan_contract'})
-        self.assertEqual(values['plan_contract'], '2')
+        self.assertEqual(values['plan_contract'], gate.PLAN_CONTRACT)
        plan = json.loads(values['plan'])
        self.assertEqual(plan['policy']['revision'], self.repo.base)
        self.assertEqual(plan['policy']['rollout'], 'shadow')
@@ -242,6 +242,15 @@ class PlannerTests(unittest.TestCase):


 class GateTests(unittest.TestCase):
+    def test_fips_provider_is_required_at_nightly_and_skipped_on_pr(self):
+        self.assertEqual(gate.expected_results('full', 'schedule', 'refs/heads/main')['fips-provider'], 'success')
+        self.assertEqual(gate.expected_results('full', 'pull_request', 'refs/heads/main')['fips-provider'], 'skipped')
+        plan = self.plan('schedule')
+        good = self.needs(plan)
+        self.assertEqual(gate.verify_results(good, 'schedule', plan['ref'], plan['head'], plan['tree']), [])
+        good['fips-provider']['result'] = 'skipped'
+        self.assertTrue(gate.verify_results(good, 'schedule', plan['ref']))
+
    def plan(self, event='pull_request', mode='full', connect=True):
        jobs = gate.expected_results(mode, event, 'refs/heads/main', connect)
        return {'schema_version': 1, 'event': event, 'ref': 'refs/heads/main', 'head': 'a' * 40, 'tree': 'b' * 40,
@@ -220,6 +220,126 @@ SH
        self.assertNotIn("        continue-on-error: true", self.publish)
        self.assertNotIn("          overwrite: true", upload)

+    def dispatch_shell(self):
+        step = self.steps["Dispatch nightly candidate to auto-testing"]
+        start = step.index("        run: |") + 1
+        return "\n".join(line[10:] for line in step[start:] if not line.strip() or line.startswith("          "))
+
+    def candidate_manifest(self, **overrides):
+        digest = overrides.pop("package_sha256", self.digest)
+        workflow_sha = overrides.pop("workflow_sha", self.sha)
+        manifest = {
+            "schema": 2,
+            "workflow_sha": workflow_sha,
+            "source_ref": "main",
+            "source_sha": self.sha,
+            "build_run_id": 12345,
+            "build_run_attempt": 1,
+            "package_sha256": digest,
+            "package_url": f"https://dl.rustfs.com/artifacts/rustfs/packages/nightly/runs/12345/1/{digest}/rustfs.deb",
+        }
+        manifest.update(overrides)
+        return manifest
+
+    def run_dispatch(self, manifest=None, **overrides):
+        manifest = self.candidate_manifest() if manifest is None else manifest
+        candidate = self.root / "candidate.json"
+        candidate.write_text(json.dumps(manifest))
+        shim = self.root / "dispatch-curl.sh"
+        shim.write_text(r'''curl() {
+  printf '%s\0' "$@" > "$FAKE_CURL_ARGS"
+  local outfile="" data=""
+  local args=("$@")
+  local i=0
+  while [[ $i -lt ${#args[@]} ]]; do
+    case "${args[$i]}" in
+      -o) outfile="${args[$((i+1))]}"; i=$((i+2));;
+      --data-binary)
+        data="${args[$((i+1))]}"
+        cp "${data#@}" "$FAKE_DISPATCH_BODY"
+        i=$((i+2));;
+      *) i=$((i+1));;
+    esac
+  done
+  [[ "$FAKE_DISPATCH_STATUS" != "curl-failure" ]] || return 28
+  if [[ -n "$outfile" ]]; then printf '%s' "$FAKE_DISPATCH_RESPONSE" > "$outfile"; fi
+  printf '%s' "$FAKE_DISPATCH_STATUS"
+}
+''')
+        env = dict(os.environ, BASH_ENV=str(shim), CANDIDATE_FILE=str(candidate), DEB_DATE="2026-09-06",
+                   SOURCE_REPO="rustfs/rustfs", DISPATCH_TOKEN="test-dispatch-token", RUNNER_TEMP=str(self.root),
+                   GITHUB_RUN_ID="12345", GITHUB_RUN_ATTEMPT="1", GITHUB_SHA=manifest["workflow_sha"],
+                   FAKE_CURL_ARGS=str(self.root / "curl-args"), FAKE_DISPATCH_BODY=str(self.root / "dispatch-body.json"),
+                   FAKE_DISPATCH_STATUS="204", FAKE_DISPATCH_RESPONSE="")
+        env.update(overrides)
+        return subprocess.run(["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.dispatch_shell()],
+                              cwd=self.root, env=env, capture_output=True, text=True)
+
+    def test_dispatch_follows_release_publish_without_blocking_vault_lanes(self):
+        names = list(self.steps)
+        self.assertLess(names.index("Publish packages to auto-testing release assets"), names.index("Dispatch nightly candidate to auto-testing"))
+        step = "\n".join(self.steps["Dispatch nightly candidate to auto-testing"])
+        self.assertIn("        if: ${{ steps.publish.outputs.candidate_file != '' }}", step)
+        self.assertIn("          DISPATCH_TOKEN: ${{ secrets.AUTO_TESTING_DISPATCH_TOKEN }}", step)
+        self.assertNotIn("secrets.GITHUB_TOKEN", step)
+        self.assertNotIn("github.token", step)
+        self.assertNotIn("continue-on-error", step)
+        self.assertIn("https://api.github.com/repos/rustfs/auto-testing/dispatches", step)
+        self.assertIn("nightly-candidate-ready", step)
+        lines = WORKFLOW.read_text().splitlines()
+        for lane in ("kms-vault-lane", "kms-vault-ha-failover"):
+            job = yaml_block(lines, lane, 2)
+            needs = next(line for line in job if line.strip().startswith("needs:"))
+            self.assertEqual(needs.strip(), "needs: resolve-source")
+            self.assertNotIn("Dispatch nightly candidate", "\n".join(job))
+        alert = "\n".join(yaml_block(lines, "alert-on-failure", 2))
+        self.assertIn("needs: [build, kms-vault-lane, kms-vault-ha-failover]", alert)
+        chain = (ROOT / ".github/workflows/rustfs-functional-chain.yml").read_text()
+        self.assertIn('workflows: ["Nightly GNU Build"]', chain)
+        self.assertIn("types: [completed]", chain)
+
+    def test_dispatch_sends_schema2_candidate_without_the_token(self):
+        result = self.run_dispatch()
+        self.assertEqual(result.returncode, 0, result.stderr)
+        body = json.loads((self.root / "dispatch-body.json").read_text())
+        self.assertEqual(body["event_type"], "nightly-candidate-ready")
+        payload = body["client_payload"]
+        self.assertLessEqual(len(payload), 10)
+        self.assertEqual(set(payload), {"event_type", "source_repo", "latest_alias_url", "auto_testing_release", "channel", "deb_date", "nightly_workflow", "candidate"})
+        self.assertEqual(payload["candidate"], self.candidate_manifest())
+        self.assertEqual(payload["latest_alias_url"], "https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb")
+        self.assertEqual(payload["auto_testing_release"], "https://github.com/rustfs/auto-testing/releases/tag/nightly")
+        self.assertEqual(payload["channel"], "nightly")
+        self.assertEqual(payload["deb_date"], "2026-09-06")
+        self.assertEqual(payload["nightly_workflow"], ".github/workflows/nightly-gnu.yml")
+        self.assertEqual(payload["source_repo"], "rustfs/rustfs")
+        validate_manifest(payload["candidate"], {"id": 12345, "run_attempt": 1, "head_sha": self.sha})
+        args = (self.root / "curl-args").read_bytes().split(b"\0")
+        self.assertIn(b"https://api.github.com/repos/rustfs/auto-testing/dispatches", args)
+        self.assertIn(b"Authorization: Bearer test-dispatch-token", args)
+        self.assertNotIn("test-dispatch-token", json.dumps(body))
+        self.assertIn("dispatched nightly-candidate-ready", result.stdout)
+
+    def test_dispatch_failure_stops_before_treating_the_candidate_as_notified(self):
+        for status, response in (("422", '{"message":"No more than 10 properties are allowed"}'), ("500", "unavailable"), ("curl-failure", "")):
+            with self.subTest(status=status):
+                result = self.run_dispatch(FAKE_DISPATCH_STATUS=status, FAKE_DISPATCH_RESPONSE=response)
+                self.assertNotEqual(result.returncode, 0, result.stdout)
+                self.assertIn("repository_dispatch", result.stderr)
+                if response:
+                    self.assertIn(response, result.stderr)
+                self.assertNotIn("dispatched nightly-candidate-ready", result.stdout)
+        for marker in ("curl-args", "dispatch-body.json"):
+            (self.root / marker).unlink(missing_ok=True)
+        missing = self.run_dispatch(DISPATCH_TOKEN="")
+        self.assertNotEqual(missing.returncode, 0)
+        self.assertIn("AUTO_TESTING_DISPATCH_TOKEN is required", missing.stderr)
+        self.assertFalse((self.root / "curl-args").exists())
+        rejected = self.run_dispatch(manifest=self.candidate_manifest(extra="nope"))
+        self.assertNotEqual(rejected.returncode, 0, rejected.stdout)
+        self.assertIn("nightly-candidate-ready contract", rejected.stderr)
+        self.assertFalse((self.root / "dispatch-body.json").exists())
+

 class ValidationPackageTests(unittest.TestCase):
    def setUp(self):
Author	SHA1	Message	Date
Hauser Zhi	2f55cf69e1	perf(filemeta): grow metacache read chunks geometrically (#9133 ) Large metacache elements (an xl.meta grown past 16 MiB with its version history) were accumulated in fixed 64 KiB reads, paying hundreds of read rounds per entry on the listing path. Double the per-round chunk up to a 4 MiB bound so each reservation stays incremental and bounded (rustfs/rustfs#2715 corrupt-length-prefix safety is unchanged) while wide listings and oversized entries decode in far fewer rounds. Add regression coverage for rustfs/rustfs#9120: a listing whose aggregate stream exceeds the per-value 16 MiB bound, and an oversized entry listed among ordinary neighbors with peek/skip interleaving.	2026-11-03 06:07:41 +08:00
Hauser Zhi	fff6e94378	fix(metrics): reconcile peer-health gauge on successful RPCs and probes (#9134 ) The rustfs_cluster_servers_offline_total gauge was event-driven only: record_peer_reachable fired solely after a fresh connector.connect() succeeded. When post-recovery traffic was served by a cached channel (tonic auto-reconnect, addr-form cache-key split, or delete-recovery channel), no fresh dial occurred and the gauge stayed stale until the process restarted. Add record_peer_reachable_if_offline (fast-path read / slow-path write) and wire it into every successful peer interaction: - remote-disk recovery probe success (gRPC DiskInfo) - periodic TCP connectivity check success (bounded-interval reconcile) - data-path RPC success (execute_with_timeout_for_op_and_health_action) - lock-path RPC success (record_rpc_success and detached completion) Fixes rustfs/rustfs#9119 Tracking: rustfs/backlog#3138	2026-11-02 19:07:37 +00:00
Hauser Zhi	41b13aeb8b	fix(tls): unify FIPS provider policy across clients (#9125 ) * fix(tls): enforce active FIPS provider at startup Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * ci: provision FIPS validation runner Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(tls): validate FIPS before observability startup Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * perf(tls): reuse the installed crypto provider Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): gate Linux-only commit helpers Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(tls): centralize FIPS provider policy Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com> Co-authored-by: Chris An <anzhengchao@gmail.com>	2026-11-02 16:58:42 +00:00
weile321	88587a8f94	fix(s3select-api): clamp overflowing u64 GetRange values to i64::MAX (#9118 )	2026-11-02 14:05:07 +00:00
hector	8c8b5da63f	ci(nightly): dispatch nightly-candidate-ready after asset publish (#9131 ) After the nightly packages are on the auto-testing nightly release and the schema 2 R2 candidate exists, post repository_dispatch to rustfs/auto-testing. Vault lanes stay parallel, and the workflow_run functional chain is left in place. Co-authored-by: hector <majinghe@users.noreply.github.com>	2026-11-02 23:05:07 +08:00
Henry Guo	00a238f838	fix(table-catalog): resolve superseded staged commits (#9124 ) fix(table-catalog): diagnose superseded staged commits Co-authored-by: Chris An <anzhengchao@gmail.com>	2026-11-02 23:04:22 +08:00
hector	965d805a4f	ci: run distributed and upgrade e2e on schedule only (#9126 ) Remove pull_request from the distributed and upgrade workflows. Distributed stays on the nightly cron, and upgrade stays on the Monday cron plus version tags. Both remain manual-dispatch lanes and stay outside the required merge checks. Co-authored-by: majinghe <majinghe@users.noreply.github.com>	2026-11-02 23:03:36 +08:00
Hauser Zhi	9d66e04ae3	test(e2e): honor namespace backpressure in mixed workload (#9122 ) Give each mixed-workload request a bounded policy for the server's explicit namespace lock timeout response and Retry-After advice. Keep unrelated SDK failures and data mismatches immediately visible. Preserve 24 workers, four rounds, cross-node checks and all retained objects. Cover exhausted SDK retries, bodyless HEAD, fatal responses, persistent contention and incorrect acknowledged bytes, and include the new cases in the protected distributed profile inventory. Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 23:02:22 +08:00
dependabot[bot]	12c58d25b5	chore(deps): bump the dependencies group across 1 directory with 5 updates (#9113 ) * chore(deps): bump the dependencies group across 1 directory with 5 updates Bumps the dependencies group with 5 updates in the / directory: \| Package \| From \| To \| \| --- \| --- \| --- \| \| [pulsar](https://github.com/streamnative/pulsar-rs) \| `6.9.0` \| `6.9.1` \| \| [arrow-array](https://github.com/apache/arrow-rs) \| `59.3.2` \| `60.1.0` \| \| [arrow-ipc](https://github.com/apache/arrow-rs) \| `59.3.2` \| `60.1.0` \| \| [arrow-schema](https://github.com/apache/arrow-rs) \| `59.3.2` \| `60.1.0` \| \| [parquet](https://github.com/apache/arrow-rs) \| `59.3.2` \| `60.1.0` \| Updates `pulsar` from 6.9.0 to 6.9.1 - [Release notes](https://github.com/streamnative/pulsar-rs/releases) - [Commits](https://github.com/streamnative/pulsar-rs/compare/v6.9.0...v6.9.1) Updates `arrow-array` from 59.3.2 to 60.1.0 - [Release notes](https://github.com/apache/arrow-rs/releases) - [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0) Updates `arrow-ipc` from 59.3.2 to 60.1.0 - [Release notes](https://github.com/apache/arrow-rs/releases) - [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0) Updates `arrow-schema` from 59.3.2 to 60.1.0 - [Release notes](https://github.com/apache/arrow-rs/releases) - [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0) Updates `parquet` from 59.3.2 to 60.1.0 - [Release notes](https://github.com/apache/arrow-rs/releases) - [Changelog](https://github.com/apache/arrow-rs/blob/main/CHANGELOG.md) - [Commits](https://github.com/apache/arrow-rs/compare/59.3.2...60.1.0) --- updated-dependencies: - dependency-name: arrow-array dependency-version: 60.1.0 dependency-type: direct:production update-type: version-update:semver-major dependency-group: dependencies - dependency-name: arrow-ipc dependency-version: 60.1.0 dependency-type: direct:production update-type: version-update:semver-major dependency-group: dependencies - dependency-name: arrow-schema dependency-version: 60.1.0 dependency-type: direct:production update-type: version-update:semver-major dependency-group: dependencies - dependency-name: parquet dependency-version: 60.1.0 dependency-type: direct:production update-type: version-update:semver-major dependency-group: dependencies - dependency-name: pulsar dependency-version: 6.9.1 dependency-type: direct:production update-type: version-update:semver-patch dependency-group: dependencies ... Signed-off-by: dependabot[bot] <support@github.com> * chore(deps): update dependencies and pin pulsar 6.9.0 Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * chore(deps): ignore pulsar 6.9 updates Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com> Co-authored-by: Hauser <housemecn@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 23:01:55 +08:00
Ramakrishna Chilaka	1ac1d883f4	perf(ecstore): shard disk path lock registries with SWAR byte keys (#9127 )	2026-11-02 23:00:48 +08:00