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.

.config/e2e-distributed-selection.txt

@@ -1,2 +1,2 @@
-sha256-linux=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
+sha256-linux=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6
-sha256-darwin=686dcc075e41a75dde3d37e5bd0393d016469f02f83aa3508f92bfc17f26a4d4
+sha256-darwin=2f97cbbeb2dbf52919b704a3d07336c162a41f95079d941d74251f9279c494e6

.config/nextest.toml

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

.github/dependabot.yml

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

.github/scheduled-validations.json

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

.github/workflows/ci.yml

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

.github/workflows/e2e-distributed.yml

@@ -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,
+# `[profile.e2e-distributed]` in `.config/nextest.toml`. The nightly schedule
-# nightly runs, and manual dispatches all execute the same fail-closed suite.
+# 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:

.github/workflows/e2e-upgrade.yml

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

.github/workflows/nightly-gnu.yml

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

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

@@ -92,6 +92,17 @@ jobs:
           CHAIN_OUTPUT: ${{ runner.temp }}/validation-sweep-candidate.json
         run: |
           set -euo pipefail
+          # Scheduled runs carry no inputs and the resolver only derives a
+          # build from a workflow_run payload (which a schedule cannot
+          # provide), so resolve the newest successful nightly build
+          # explicitly before invoking it.
+          if [ -z "${BUILD_RUN_ID:-}" ]; then
+            LATEST="$(gh api "repos/rustfs/rustfs/actions/workflows/nightly-gnu.yml/runs?branch=main&status=success&per_page=1" --jq '.workflow_runs[0]')"
+            BUILD_RUN_ID="$(printf '%s' "${LATEST}" | jq -r '.id')"
+            BUILD_RUN_ATTEMPT="$(printf '%s' "${LATEST}" | jq -r '.run_attempt')"
+            export BUILD_RUN_ID BUILD_RUN_ATTEMPT
+            echo "scheduled sweep pinned to nightly ${BUILD_RUN_ID} attempt ${BUILD_RUN_ATTEMPT}"
+          fi
           python3 scripts/resolve_functional_candidate.py
           python3 - <<'PY'
           import json
@@ -99,15 +110,15 @@ jobs:
           from pathlib import Path
           root = Path(os.environ["RUNNER_TEMP"]) / "validation-sweep"
           root.mkdir(parents=True, exist_ok=True)
-          candidate = json.loads((Path(os.environ["RUNNER_TEMP"]) / "validation-sweep-candidate.json").read_text())
+          chain = json.loads((Path(os.environ["RUNNER_TEMP"]) / "validation-sweep-candidate.json").read_text())
-          manifest = candidate["manifest"]
+          manifest = chain["candidate"]["manifest"]
           (root / "candidate.json").write_text(json.dumps({
               "package_url": manifest["package_url"],
               "package_sha256": manifest["package_sha256"],
               "source_sha": manifest["source_sha"],
           }, sort_keys=True) + "\n")
           with open(os.environ["GITHUB_OUTPUT"], "a") as output:
-              output.write("testing_sha=" + candidate["testing_sha"] + "\n")
+              output.write("testing_sha=" + chain["testing_sha"] + "\n")
               output.write("package_url=" + manifest["package_url"] + "\n")
               output.write("source_sha=" + manifest["source_sha"] + "\n")
           PY
@@ -121,6 +132,26 @@ jobs:
           path: auto-testing
           persist-credentials: false
 
+      - name: Prepare validation Python runtime
+        # The pool / ilm / replication validation runners require a managed
+        # Python, exactly like the chain's validation job provides.
+        uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.12"
+
+      - name: Install validation Python requirements
+        # Case scripts call python3 and import boto3 at startup. Install the
+        # functional chain's client into a per-run env so that interpreter
+        # sees it without changing the runner's shared Python.
+        run: |
+          set -euo pipefail
+          validation_python="${RUNNER_TEMP}/rustfs-validation-python-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
+          test ! -e "${validation_python}"
+          python3 -m venv "${validation_python}"
+          "${validation_python}/bin/python" -m pip install \
+            -r auto-testing/scripts/validation_ilm/requirements.txt
+          printf '%s\n' "${validation_python}/bin" >> "${GITHUB_PATH}"
+
       - name: Select sweep cases
         id: select
         env:
@@ -224,7 +255,8 @@ jobs:
             # topology, this purge guarantees the next case starts clean even
             # when the previous one died mid-run.
             for node in ${RUSTFS_NODES}; do
-              ssh -o BatchMode=yes -o ConnectTimeout=10 "${RUSTFS_SSH_USER}@${node}" '
+              # -n keeps ssh from consuming the case list this loop reads from stdin
+              ssh -n -o BatchMode=yes -o ConnectTimeout=10 "${RUSTFS_SSH_USER}@${node}" '
                 SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n"
                 ${SUDO} systemctl stop rustfs 2>/dev/null || true
                 if ${SUDO} dpkg -l rustfs 2>/dev/null | grep -q "^ii"; then
@@ -243,6 +275,7 @@ jobs:
           SWEEP_ROOT: ${{ runner.temp }}/validation-sweep
           PACKAGE_URL: ${{ steps.candidate.outputs.package_url }}
           SOURCE_SHA: ${{ steps.candidate.outputs.source_sha }}
+          AUTO_TESTING_SHA: ${{ steps.candidate.outputs.testing_sha }}
         run: |
           set -euo pipefail
           python3 - <<'PY'
@@ -351,7 +384,10 @@ jobs:
             ${{ runner.temp }}/validation-sweep/sweep-cases.txt
             ${{ runner.temp }}/validation-sweep/candidate.json
             ${{ runner.temp }}/validation-sweep/results/
-          if-no-files-found: error
+            ${{ runner.temp }}/validation-sweep/run/
+          # warn (not error): an early infra failure legitimately leaves no
+          # per-case evidence, and the failing step is already red.
+          if-no-files-found: warn
           retention-days: 30
 
       - name: Cleanup environment (after)

Cargo.lock

@@ -299,7 +299,7 @@ dependencies = [
  "strum",
  "thiserror 2.0.21",
  "uuid",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]
 
 [[package]]
@@ -370,17 +370,17 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "3f04e0a89cc175eaf3740282537851e6fd9d7276f97730b9d0ea2859c11a85df"
 dependencies = [
  "arrow-arith",
- "arrow-array",
+ "arrow-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
  "arrow-cast",
  "arrow-csv",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-ipc",
+ "arrow-ipc 59.3.2",
  "arrow-json",
  "arrow-ord",
  "arrow-row",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "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-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
  "arrow-ord",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "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-buffer 59.3.2",
- "arrow-schema",
+ "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,14 +539,29 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f5dd1afadda2eb821aaf851bb4c0adef60c30cc0f1a29bd02fb99dd07fb9f58d"
 dependencies = [
- "arrow-array",
+ "arrow-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "arrow-select 59.3.2",
  "flatbuffers",
  "lz4_flex",
- "zstd 0.13.3",
+ "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]]
@@ -501,12 +570,12 @@ version = "59.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2bc4d56a0dad84cc51c5812ccff48b642047a1f3be80349cd84318983e79edb"
 dependencies = [
- "arrow-array",
+ "arrow-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
  "arrow-cast",
  "arrow-ord",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "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-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "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",
@@ -976,9 +1066,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-kms"
-version = "1.128.0"
+version = "1.129.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ef519fda62fd0fdc066948f5ef33ecb8ae1bf0187bf1a12e8a775c93c4c47e7e"
+checksum = "5c5cdb29ad25d71c9ee54c408a672c98d90a24a0ea6eb063025483e0e894b411"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -1001,9 +1091,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-s3"
-version = "1.156.0"
+version = "1.157.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1072c9aa20159b9b417115342d38c4c53108fc98e5359fcb86b7d4575747eb59"
+checksum = "6ea9794c741520d8baec9462bcd0ad12164d4f5f78d008a8da9d6a0b7222caa8"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -1037,9 +1127,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sso"
-version = "1.118.0"
+version = "1.119.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b4a89786db8ee6dff2c9302a354ef081354fcb0f527e16ee3cb249a6a5cb6df7"
+checksum = "3c414f25ec4aa9dc18f5cc8f55e18848bb509fcda4ce7c77bee6072e95806bc9"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -1062,9 +1152,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-ssooidc"
-version = "1.120.0"
+version = "1.121.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "521c6a11a25f1244efddf947523f277a51a63f0a7ec228f0f62d826e88106366"
+checksum = "bc638bb4355e6514334dcc3228cd4622f5ee54d704bd6a9010a5a91d7856d9d9"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -1087,9 +1177,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-sts"
-version = "1.123.0"
+version = "1.124.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8754d38d29448bd1d4aa4a6312ed010dfa296b3436eb2c551cc6e42fc6a66a38"
+checksum = "3f8a0dd8fbde04ba58ad722113c7addb11bf94cf549b1ea0eb9ea427fb87198a"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -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-ipc 59.3.2",
- "arrow-schema",
+ "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-data 59.3.2",
- "arrow-ipc",
+ "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",
@@ -3960,6 +4050,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
  "block-buffer 0.10.4",
+ "const-oid 0.9.6",
  "crypto-common 0.1.7",
  "subtle",
 ]
@@ -5295,9 +5386,9 @@ dependencies = [
 
 [[package]]
 name = "hotpath"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b39e941010d86063ff233bf8325269bf68097fb24ba67000fbddc0a410e8056e"
+checksum = "d9632c1fd213e63c1b88d8f3e59543da4999b17f859cc1e1d67cf8f2207e8935"
 dependencies = [
  "arc-swap",
  "async-channel",
@@ -5330,15 +5421,15 @@ dependencies = [
 
 [[package]]
 name = "hotpath-drain"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d0faae1a9e1fd8ce5debaaaf452dcbe833fc8a8a4f3779d2ec8828b116c3c16a"
+checksum = "738324f75aec847dea984f6806402f4f1bdfab437b4700fc8c3ebcad2dff6424"
 
 [[package]]
 name = "hotpath-macros"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a24f5179ff680132003306354efd0ac8cde12e87b1a788719c836e91af35cb58"
+checksum = "5d92fdfec803932e5764ea93c201cf7c6c2d8cff1871da44f64f5094c9c3ef8a"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -5959,7 +6050,7 @@ dependencies = [
  "lz4",
  "snap",
  "uuid",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]
 
 [[package]]
@@ -6052,6 +6143,9 @@ name = "lazy_static"
 version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb"
+dependencies = [
+ "spin 0.9.9",
+]
 
 [[package]]
 name = "lexical-core"
@@ -6261,9 +6355,9 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.34"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c"
 
 [[package]]
 name = "lru"
@@ -6627,9 +6721,9 @@ dependencies = [
 
 [[package]]
 name = "ml-kem"
-version = "0.3.3"
+version = "0.3.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9ba01ebcf90f6aff8e9b9f55bb69b19e6250e4ade5787d0040eaa27dd226238"
+checksum = "84ca78b66f46050d49fa21693ae8adab2081d8101b65d8067316b6b70b64a0d1"
 dependencies = [
  "hybrid-array",
  "kem",
@@ -6642,9 +6736,9 @@ dependencies = [
 
 [[package]]
 name = "module-lattice"
-version = "0.2.3"
+version = "0.2.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0c61b87c9683ab7cb1c6871d261ad5479b6b10ceb52c4352aaca3b5d35a8febe"
+checksum = "c4a1e412c8ee83a1c678f427806219c49c96b2511b9f07497514f0e43fb26155"
 dependencies = [
  "ctutils",
  "hybrid-array",
@@ -6940,6 +7034,22 @@ dependencies = [
  "rand_core 0.10.1",
 ]
 
+[[package]]
+name = "num-bigint-dig"
+version = "0.8.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e661dda6640fad38e827a6d4a310ff4763082116fe217f279885c97f511bb0b7"
+dependencies = [
+ "lazy_static",
+ "libm",
+ "num-integer",
+ "num-iter",
+ "num-traits",
+ "rand 0.8.8",
+ "smallvec",
+ "zeroize",
+]
+
 [[package]]
 name = "num-complex"
 version = "0.4.6"
@@ -6964,6 +7074,16 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "num-iter"
+version = "0.1.46"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c92800bd69a1eac91786bcfe9da64a897eb72911b8dc3095decbd07429e8048b"
+dependencies = [
+ "num-integer",
+ "num-traits",
+]
+
 [[package]]
 name = "num-rational"
 version = "0.4.2"
@@ -7035,18 +7155,19 @@ dependencies = [
 
 [[package]]
 name = "oauth2"
-version = "5.1.0"
+version = "5.0.0"
-source = "git+https://github.com/ramosbugs/oauth2-rs?rev=89585f4a280afd7b285eaf1c9be0068874e03d22#89585f4a280afd7b285eaf1c9be0068874e03d22"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "51e219e79014df21a225b1860a479e2dcd7cbd9130f4defd4bd0e191ea31d67d"
 dependencies = [
  "base64 0.22.1",
  "chrono",
  "getrandom 0.2.17",
  "http 1.5.0",
- "rand 0.10.3",
+ "rand 0.8.8",
  "serde",
  "serde_json",
  "serde_path_to_error",
- "sha2 0.11.1",
+ "sha2 0.10.9",
  "thiserror 1.0.69",
  "url",
 ]
@@ -7208,29 +7329,31 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
 [[package]]
 name = "openidconnect"
-version = "4.0.1"
+version = "4.1.0"
-source = "git+https://github.com/ramosbugs/openidconnect-rs?rev=2f77ee73bc4554e1cb6279adb386005429fb4e41#2f77ee73bc4554e1cb6279adb386005429fb4e41"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "df0d098a59353679624d406825940139e2b2d554a52b64c9fb1b1bc42a11fe5c"
 dependencies = [
  "base64 0.22.1",
  "chrono",
  "dyn-clone",
  "ed25519-dalek 3.0.0",
- "hmac 0.13.0",
+ "getrandom 0.4.3",
+ "hmac 0.12.1",
  "http 1.5.0",
  "itertools 0.14.0",
  "log",
  "oauth2",
  "p256",
  "p384",
- "rand 0.10.3",
+ "rand 0.8.8",
- "rsa",
+ "rsa 0.9.10",
  "serde",
  "serde-value",
  "serde_json",
  "serde_path_to_error",
  "serde_plain",
  "serde_with",
- "sha2 0.11.1",
+ "sha2 0.10.9",
  "subtle",
  "thiserror 1.0.69",
  "url",
@@ -7558,12 +7681,12 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8eb260d566f2fbdabe356d26aaa5d13403aac29b5b4e44463b8f9bd7b2b12680"
 dependencies = [
  "ahash",
- "arrow-array",
+ "arrow-array 59.3.2",
- "arrow-buffer",
+ "arrow-buffer 59.3.2",
- "arrow-data",
+ "arrow-data 59.3.2",
- "arrow-ipc",
+ "arrow-ipc 59.3.2",
- "arrow-schema",
+ "arrow-schema 59.3.2",
- "arrow-select",
+ "arrow-select 59.3.2",
  "base64 0.23.1",
  "brotli 8.0.4",
  "bytes",
@@ -7582,7 +7705,33 @@ dependencies = [
  "snap",
  "tokio",
  "twox-hash",
- "zstd 0.13.3",
+ "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]]
@@ -7834,6 +7983,17 @@ dependencies = [
  "futures-io",
 ]
 
+[[package]]
+name = "pkcs1"
+version = "0.7.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8ffb9f10fa047879315e6625af03c164b16962a5368d724ed16323b68ace47f"
+dependencies = [
+ "der 0.7.10",
+ "pkcs8 0.10.2",
+ "spki 0.7.3",
+]
+
 [[package]]
 name = "pkcs1"
 version = "0.8.0-rc.4"
@@ -8932,9 +9092,9 @@ dependencies = [
 
 [[package]]
 name = "reflink-copy"
-version = "0.1.30"
+version = "0.1.32"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d9dd7ab4af0363d5ccfd2838d782a28196cf32a5cc2e4fe3c5dc83f2be588b8b"
+checksum = "1ad64f61357bcf16105a0f353f9783733409c8bd889bc326ad857959605ea927"
 dependencies = [
  "cfg-if",
  "libc",
@@ -9096,6 +9256,26 @@ dependencies = [
  "xmlparser",
 ]
 
+[[package]]
+name = "rsa"
+version = "0.9.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b8573f03f5883dcaebdfcf4725caa1ecb9c15b2ef50c43a07b816e06799bb12d"
+dependencies = [
+ "const-oid 0.9.6",
+ "digest 0.10.7",
+ "num-bigint-dig",
+ "num-integer",
+ "num-traits",
+ "pkcs1 0.7.5",
+ "pkcs8 0.10.2",
+ "rand_core 0.6.4",
+ "signature 2.2.0",
+ "spki 0.7.3",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "rsa"
 version = "0.10.0-rc.18"
@@ -9106,7 +9286,7 @@ dependencies = [
  "crypto-bigint",
  "crypto-primes",
  "digest 0.11.3",
- "pkcs1",
+ "pkcs1 0.8.0-rc.4",
  "pkcs8 0.11.0",
  "rand_core 0.10.1",
  "sha2 0.11.1",
@@ -9211,13 +9391,13 @@ dependencies = [
  "p521",
  "pageant",
  "pbkdf2 0.13.0",
- "pkcs1",
+ "pkcs1 0.8.0-rc.4",
  "pkcs5 0.8.1",
  "pkcs8 0.11.0",
  "polyval",
  "rand 0.10.3",
  "rand_core 0.10.1",
- "rsa",
+ "rsa 0.10.0-rc.18",
  "russh-cryptovec",
  "russh-util",
  "salsa20 0.11.0",
@@ -9409,7 +9589,7 @@ dependencies = [
  "regex",
  "reqwest",
  "rmp-serde",
- "rsa",
+ "rsa 0.10.0-rc.18",
  "rust-embed",
  "rustfs-audit",
  "rustfs-common",
@@ -9612,7 +9792,7 @@ dependencies = [
  "jsonwebtoken",
  "pbkdf2 0.13.0",
  "rand 0.10.3",
- "rsa",
+ "rsa 0.10.0-rc.18",
  "serde",
  "serde_json",
  "sha2 0.11.1",
@@ -10150,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",
 ]
@@ -12006,7 +12186,7 @@ dependencies = [
  "p384",
  "p521",
  "rand_core 0.10.1",
- "rsa",
+ "rsa 0.10.0-rc.18",
  "sec1",
  "serde",
  "sha1 0.11.0",
@@ -12356,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",
@@ -12377,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",
@@ -13114,9 +13294,9 @@ dependencies = [
 
 [[package]]
 name = "unicode-properties"
-version = "0.1.4"
+version = "0.1.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7df058c713841ad818f1dc5d3fd88063241cc61f49f5fbea4b951e8cf5a8d71d"
+checksum = "993ad287eb20e8ba755312801cfe1413e0f25d604075863acaddca10b04c1833"
 
 [[package]]
 name = "unicode-segmentation"
@@ -14018,9 +14198,9 @@ dependencies = [
 
 [[package]]
 name = "zip"
-version = "9.1.3"
+version = "9.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1156b39d142a34b999c2310a50455e0cb3760bf7efad0122f796d813bee26ccb"
+checksum = "e408c6cc4699dd017d53dc080cd51c2cea2b5354af57a1479bd265ed42911487"
 dependencies = [
  "aes 0.9.3",
  "bzip2",
@@ -14031,6 +14211,7 @@ dependencies = [
  "getrandom 0.4.3",
  "hmac 0.13.0",
  "indexmap 2.14.2",
+ "libc",
  "lzma-rust2",
  "memchr",
  "nix 0.31.3",
@@ -14071,15 +14252,15 @@ dependencies = [
 
 [[package]]
 name = "zstd"
-version = "0.13.3"
+version = "0.13.4"
 dependencies = [
  "apache-avro",
- "arrow-array",
+ "arrow-array 60.1.0",
- "arrow-ipc",
+ "arrow-ipc 60.1.0",
- "arrow-schema",
+ "arrow-schema 60.1.0",
  "bytes",
  "kafka-protocol",
- "parquet",
+ "parquet 60.1.0",
  "semver",
  "toml",
  "zstd 0.14.3",

Cargo.toml

@@ -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"
@@ -217,7 +217,7 @@ chacha20poly1305 = { version = "0.11.0" }
 crc-fast = "1.10.0"
 hmac = { version = "0.13.0" }
 jsonwebtoken = { version = "11.1.0" }
-openidconnect = { default-features = false, version = "4.0" }
+openidconnect = { default-features = false, version = "4.1.0" }
 pbkdf2 = "0.13.0"
 p256 = { version = "0.14.0", features = ["ecdsa", "pkcs8"] }
 rsa = { version = "=0.10.0-rc.18" }
@@ -253,9 +253,9 @@ atoi = "3.1.1"
 atomic_enum = "0.3.0"
 aws-config = { version = "1.13.0" }
 aws-credential-types = { version = "1.3.1" }
-aws-sdk-kms = { default-features = false, version = "1.128.0" }
+aws-sdk-kms = { default-features = false, version = "1.129.0" }
-aws-sdk-s3 = { default-features = false, version = "1.156.0" }
+aws-sdk-s3 = { default-features = false, version = "1.157.0" }
-aws-sdk-sts = { default-features = false, version = "1.123.0" }
+aws-sdk-sts = { default-features = false, version = "1.124.0" }
 aws-smithy-async = { version = "1.3.1" }
 aws-smithy-http-client = { default-features = false, version = "1.5.2" }
 aws-smithy-runtime-api = { version = "1.19.2" }
@@ -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.1.3"
+zip = "9.2.2"
 zstd = "0.14.3"
 
 # Observability and Metrics
@@ -384,9 +384,9 @@ 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.2", default-features = false }
+hotpath = { version = "0.30.3", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.49" }
 
@@ -427,7 +427,3 @@ strip = "none"
 # Pin upstream fixes until compatible crates.io releases are available.
 [patch.crates-io]
 zstd = { path = "crates/zstd-compat" }
-# Preserve RustFS OIDC flows while migrating the old elliptic-curve dependency chain.
-# Upstream: openidconnect-rs#244 and oauth2-rs#352.
-openidconnect = { git = "https://github.com/ramosbugs/openidconnect-rs", rev = "2f77ee73bc4554e1cb6279adb386005429fb4e41" }
-oauth2 = { git = "https://github.com/ramosbugs/oauth2-rs", rev = "89585f4a280afd7b285eaf1c9be0068874e03d22" }

crates/e2e_test/README.md

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

crates/e2e_test/src/distributed/concurrency_stability_test.rs

@@ -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
+                after_lock_contention(&format!("{context} COPY retained"), || {
-                    .copy_object()
+                    copier
-                    .bucket(&bucket)
+                        .copy_object()
-                    .key(&copy_key)
+                        .bucket(&bucket)
-                    .copy_source(format!("{bucket}/{key}"))
+                        .key(&copy_key)
-                    .send()
+                        .copy_source(format!("{bucket}/{key}"))
-                    .await?;
+                        .send()
-                assert_object_bytes(writer, &bucket, &copy_key, &body).await?;
+                })
+                .await??;
+                assert_workload_bytes(writer, &bucket, &copy_key, &body).await?;
 
-                writer.delete_object().bucket(&bucket).key(&key).send().await?;
+                after_lock_contention(&format!("{context} DELETE source"), || {
-                let missing = reader
+                    writer.delete_object().bucket(&bucket).key(&key).send()
-                    .head_object()
+                })
-                    .bucket(&bucket)
+                .await??;
-                    .key(&key)
+                let missing = after_lock_contention(&format!("{context} HEAD deleted source"), || {
-                    .send()
+                    reader.head_object().bucket(&bucket).key(&key).send()
-                    .await
+                })
-                    .expect_err("deleted source key must not remain visible");
+                .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}"))?;
         }

crates/ecstore/Cargo.toml

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

crates/ecstore/src/cluster/rpc/remote_disk.rs

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

crates/ecstore/src/cluster/rpc/remote_locker.rs

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

crates/ecstore/src/disk/disk_store.rs

@@ -853,6 +853,8 @@ pub(crate) struct DiskHealthMetricEpoch {
     fallback_operation_metrics: StdRwLock<HashMap<&'static str, Arc<DiskOperationMetrics>>>,
     #[cfg(test)]
     multipart_cleanup_observer: std::sync::Mutex<Option<std::sync::Weak<ScopedMultipartCleanupCompletions>>>,
+    #[cfg(test)]
+    operation_timeout_override: std::sync::Mutex<Option<Duration>>,
     /// Caller API operations currently executing through the disk health wrapper.
     api_waiting: AtomicU32,
     /// Operations rejected because the disk was unavailable for the caller.
@@ -880,6 +882,8 @@ impl Default for DiskHealthMetricEpoch {
             fallback_operation_metrics: StdRwLock::new(HashMap::new()),
             #[cfg(test)]
             multipart_cleanup_observer: std::sync::Mutex::new(None),
+            #[cfg(test)]
+            operation_timeout_override: std::sync::Mutex::new(None),
             api_waiting: AtomicU32::new(0),
             total_errors_availability: AtomicU64::new(0),
             total_errors_timeout: AtomicU64::new(0),
@@ -1573,6 +1577,20 @@ impl Default for DiskHealthTracker {
     }
 }
 
+#[cfg(test)]
+#[derive(Debug)]
+pub(crate) struct ScopedDiskOperationTimeout {
+    metrics: Arc<DiskHealthMetricEpoch>,
+    previous: Option<Duration>,
+}
+
+#[cfg(test)]
+impl Drop for ScopedDiskOperationTimeout {
+    fn drop(&mut self) {
+        *self.metrics.operation_timeout_override.lock().expect("test timeout lock") = self.previous;
+    }
+}
+
 /// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
 /// This is similar to Go's xlStorageDiskIDCheck.
 #[derive(Debug, Clone)]
@@ -1914,6 +1932,23 @@ impl LocalDiskWrapper {
         self.health_check
     }
 
+    #[cfg(test)]
+    pub(crate) fn override_operation_timeout_for_test(&self, timeout: Duration) -> ScopedDiskOperationTimeout {
+        assert!(!timeout.is_zero(), "fault injection needs a positive timeout");
+        // The metrics epoch follows this drive across clones and reconnects.
+        // Its scoped override cannot shorten unrelated healthy drives' I/O.
+        let previous = self
+            .metrics
+            .operation_timeout_override
+            .lock()
+            .expect("test timeout lock")
+            .replace(timeout);
+        ScopedDiskOperationTimeout {
+            metrics: Arc::clone(&self.metrics),
+            previous,
+        }
+    }
+
     /// Enable health monitoring after disk creation.
     /// Used to defer health checks until after startup format loading completes.
     pub fn enable_health_check(&self) {
@@ -2638,6 +2673,13 @@ impl LocalDiskWrapper {
             }
             return result;
         }
+        #[cfg(test)]
+        let timeout_duration = self
+            .metrics
+            .operation_timeout_override
+            .lock()
+            .expect("test timeout lock")
+            .map_or(timeout_duration, |injected| timeout_duration.min(injected));
         // Execute the operation with timeout
         let result = tokio::time::timeout(timeout_duration, operation).await;
 
@@ -6044,6 +6086,67 @@ mod tests {
         assert!(health.offline_duration().is_none());
     }
 
+    #[tokio::test]
+    async fn scoped_operation_timeout_is_per_disk_and_preserves_caller_policy() {
+        let directories = [
+            tempfile::tempdir().expect("faulted drive"),
+            tempfile::tempdir().expect("healthy drive"),
+        ];
+        let mut wrappers = Vec::new();
+        for directory in &directories {
+            let endpoint = Endpoint::try_from(directory.path().to_str().expect("UTF-8 drive path")).expect("drive endpoint");
+            let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local drive"));
+            wrappers.push(LocalDiskWrapper::new(disk, false));
+        }
+        let reconnected = LocalDiskWrapper::new_with_reconnect_state(
+            Arc::clone(&wrappers[0].disk),
+            false,
+            Some(wrappers[0].health_tracker_epoch_for_reconnect()),
+        );
+        let scope = wrappers[0].override_operation_timeout_for_test(Duration::from_secs(1));
+        tokio::time::pause();
+        let operation = || async {
+            tokio::time::sleep(Duration::from_secs(2)).await;
+            Ok(())
+        };
+        let (faulted, healthy) = tokio::join!(
+            reconnected.track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_secs(3),
+                TimeoutHealthAction::IgnoreFailure,
+            ),
+            wrappers[1].track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_secs(3),
+                TimeoutHealthAction::IgnoreFailure,
+            ),
+        );
+        assert_eq!(faulted.expect_err("only the selected drive must time out"), DiskError::Timeout);
+        healthy.expect("healthy drive keeps its caller's deadline");
+        let shorter_started = tokio::time::Instant::now();
+        let shorter = wrappers[0]
+            .track_disk_health_with_op_and_timeout_action(
+                "read_metadata",
+                operation,
+                Duration::from_millis(500),
+                TimeoutHealthAction::IgnoreFailure,
+            )
+            .await;
+        assert_eq!(shorter.expect_err("fault injection cannot extend a caller deadline"), DiskError::Timeout);
+        assert!(shorter_started.elapsed() < Duration::from_secs(1));
+        wrappers[0]
+            .track_disk_health_with_op("read_metadata", operation, Duration::ZERO)
+            .await
+            .expect("explicit unbounded physical ownership cannot acquire a test deadline");
+        drop(scope);
+        reconnected
+            .track_disk_health_with_op("read_metadata", operation, Duration::from_secs(3))
+            .await
+            .expect("dropping the scope restores reconnected callers' deadlines");
+    }
+
     #[tokio::test]
     async fn ignored_timeout_does_not_mark_drive_failure() {
         let dir = tempfile::tempdir().expect("temp dir should be created");

crates/ecstore/src/disk/fd_cache_tests.rs

@@ -5,6 +5,7 @@ use super::{FdCache, FdCacheEntry, FdKey};
 use std::fs::File;
 use std::os::unix::fs::FileExt;
 use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
 
 fn key(volume: &str, path: &str, direct: bool) -> FdKey {
     FdKey {
@@ -52,7 +53,11 @@ async fn uring_fd_cache_exact_invalidation_removes_both_modes_and_preserves_othe
 async fn uring_fd_cache_exact_invalidation_bumps_one_generation_and_rejects_old_epoch_inserts() {
     let (_directory, file) = fixture();
     let cache = FdCache::new();
-    let entry = Arc::new(FdCacheEntry { file, len: 3 });
+    let entry = Arc::new(FdCacheEntry {
+        file,
+        len: 3,
+        invalidated: AtomicBool::new(false),
+    });
     let old_generation = cache.generation();
     cache.invalidate_exact("v", "part").await;
     let current_generation = cache.generation();
@@ -125,7 +130,7 @@ async fn uring_fd_cache_volume_and_clear_invalidation_cover_both_modes() {
         cache.insert(key("v", "part", direct), Arc::clone(&file)).await;
         assert!(
             cache.get(&key("v", "part", direct)).await.is_some(),
-            "new entries must survive the earlier volume predicate"
+            "new entries must survive the earlier volume invalidation"
         );
     }
     cache.clear();
@@ -164,6 +169,7 @@ async fn uring_fd_cache_invalidation_keeps_borrowed_old_inode_alive_and_allows_r
     let replacement = Arc::new(FdCacheEntry {
         file: Arc::new(File::open(directory.path().join("part")).expect("open replacement inode")),
         len: 11,
+        invalidated: AtomicBool::new(false),
     });
     for direct in [false, true] {
         let key = key("v", "part", direct);

crates/ecstore/src/disk/local.rs

@@ -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,
@@ -5167,6 +5167,7 @@ impl LocalIoBackend for StdBackend {
                     Arc::new(FdCacheEntry {
                         file: Arc::new(file),
                         len: metadata_len,
+                        invalidated: AtomicBool::new(false),
                     })
                 });
                 #[cfg(not(target_os = "linux"))]
@@ -5607,6 +5608,9 @@ const FD_CACHE_CAPACITY: u64 = 512;
 #[cfg(any(target_os = "linux", all(unix, test)))]
 const FD_CACHE_TTL: std::time::Duration = std::time::Duration::from_secs(5);
 
+#[cfg(any(target_os = "linux", all(unix, test)))]
+const FD_CACHE_MAINTENANCE_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
+
 #[cfg(target_os = "linux")]
 fn is_io_uring_fd_cache_enabled() -> bool {
     rustfs_utils::get_env_bool(ENV_RUSTFS_IO_URING_FD_CACHE, DEFAULT_RUSTFS_IO_URING_FD_CACHE)
@@ -5702,6 +5706,8 @@ struct FdCacheEntry {
     file: Arc<std::fs::File>,
     /// File length captured together with the descriptor.
     len: u64,
+    #[cfg(any(target_os = "linux", all(unix, test)))]
+    invalidated: AtomicBool,
 }
 
 /// Per-disk cache of open descriptors for io_uring reads (backlog#1145).
@@ -5735,6 +5741,13 @@ struct FdCacheEntry {
 #[cfg(any(target_os = "linux", all(unix, test)))]
 struct FdCache {
     cache: moka::future::Cache<FdKey, Arc<FdCacheEntry>>,
+    // Moka's entry count can lag inserts. Only a cache that has never accepted
+    // an insert can safely skip the foreground invalidation scan.
+    has_inserted_entries: AtomicBool,
+    // The only sender belongs to this cache. Dropping it lets the worker finish
+    // maintenance and release its cache clone without retaining the backend.
+    maintenance_tx: tokio::sync::watch::Sender<()>,
+    maintenance_task: std::sync::OnceLock<tokio::task::JoinHandle<()>>,
     /// Bumped by every invalidation. A miss-path open snapshots this before it
     /// opens and refuses to insert if it moved, so an fd opened before a
     /// heal/delete commit can never be resurrected into the cache after the
@@ -5752,20 +5765,78 @@ impl FdCache {
     /// `new` (`FD_CACHE_TTL`); tests inject a short TTL to exercise the backstop
     /// eviction (rustfs/backlog#1180) without a multi-second wait.
     fn with_ttl(ttl: std::time::Duration) -> Self {
+        let (maintenance_tx, _) = tokio::sync::watch::channel(());
         Self {
             cache: moka::future::Cache::builder()
                 .max_capacity(FD_CACHE_CAPACITY)
                 .time_to_live(ttl)
-                // Required for `invalidate_entries_if`; without it that call
-                // fails at runtime instead of dropping stale descriptors.
-                .support_invalidation_closures()
                 .build(),
+            has_inserted_entries: AtomicBool::new(false),
+            maintenance_tx,
+            maintenance_task: std::sync::OnceLock::new(),
             generation: std::sync::atomic::AtomicU64::new(0),
         }
     }
 
+    fn schedule_maintenance(&self) {
+        self.maintenance_task.get_or_init(|| {
+            let cache = self.cache.clone();
+            let mut notifications = self.maintenance_tx.subscribe();
+            // Invalidation runs on the disk API's Tokio runtime. Starting lazily
+            // keeps construction runtime-independent and idle caches task-free.
+            tokio::spawn(async move {
+                let mut interval = tokio::time::interval_at(
+                    tokio::time::Instant::now() + FD_CACHE_MAINTENANCE_INTERVAL,
+                    FD_CACHE_MAINTENANCE_INTERVAL,
+                );
+                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
+                loop {
+                    let stopping = tokio::select! {
+                        changed = notifications.changed() => changed.is_err(),
+                        _ = interval.tick() => false,
+                    };
+                    if stopping {
+                        cache.invalidate_all();
+                    }
+                    cache.run_pending_tasks().await;
+                    if stopping {
+                        break;
+                    }
+                    if cache.entry_count() != 0 {
+                        for (key, entry) in cache.iter() {
+                            if entry.invalidated.load(Ordering::Acquire) {
+                                // Concurrent replacement may evict a fresh fd;
+                                // a later read safely reopens it.
+                                cache.invalidate(key.as_ref()).await;
+                            }
+                        }
+                        cache.run_pending_tasks().await;
+                    }
+                }
+            })
+        });
+        self.maintenance_tx.send_replace(());
+    }
+
+    fn invalidate_matching(&self, matches: impl Fn(&FdKey) -> bool) {
+        if !self.has_inserted_entries.load(Ordering::Acquire) {
+            return;
+        }
+        // Entries already present are visited unless concurrently removed. The
+        // generation fence rejects an older open inserted after this scan.
+        for (key, entry) in self.cache.iter() {
+            if matches(&key) {
+                entry.invalidated.store(true, Ordering::Release);
+            }
+        }
+        self.schedule_maintenance();
+    }
+
     async fn get(&self, key: &FdKey) -> Option<Arc<FdCacheEntry>> {
-        self.cache.get(key).await
+        self.cache
+            .get(key)
+            .await
+            .filter(|entry| !entry.invalidated.load(Ordering::Acquire))
     }
 
     /// Current invalidation generation, snapshotted by the miss path before it
@@ -5783,6 +5854,7 @@ impl FdCache {
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             return;
         }
+        self.has_inserted_entries.store(true, Ordering::Release);
         self.cache.insert(key.clone(), entry).await;
         if self.generation.load(Ordering::Acquire) != gen_at_open {
             self.cache.invalidate(&key).await;
@@ -5809,38 +5881,24 @@ impl FdCache {
     /// Drop every descriptor for `volume` whose path is `prefix` or lies under it.
     /// A component-boundary check keeps `a/b` from invalidating `a/bc`.
     ///
-    /// moka applies the predicate to entries inserted at or before this call and
+    /// Mark matching entries immediately so reads reject them while maintenance
-    /// guarantees a later `get` never returns one of them, so the invalidation is
+    /// removes them. The scan borrows its paths and retains no invalidation
-    /// effective immediately even though the removal itself is deferred. Reserved
+    /// closures when the cache is empty or idle.
-    /// for paths whose exact keys are unknown (`delete`); every predicate is
-    /// re-evaluated by subsequent `get`s, so registering one per write would tax
-    /// the read path.
     fn invalidate_under(&self, volume: &str, prefix: &str) {
-        // Bump before registering the predicate so a concurrent miss-path insert
+        // Bump before scanning the cache so a concurrent miss-path insert
         // that snapshotted the old generation is refused (rustfs/backlog#1176).
         self.generation.fetch_add(1, Ordering::AcqRel);
-        let volume = volume.to_owned();
+        let prefix = prefix.trim_end_matches('/');
-        let prefix = prefix.trim_end_matches('/').to_owned();
+        self.invalidate_matching(|k| {
-        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| {
+            k.volume == volume && (k.path == prefix || k.path.strip_prefix(prefix).is_some_and(|r| r.starts_with('/')))
-            k.volume == volume && (k.path == prefix || k.path.strip_prefix(&prefix).is_some_and(|r| r.starts_with('/')))
+        });
-        };
-        if self.cache.invalidate_entries_if(matches).is_err() {
-            // Closure support is enabled at construction, so this is unreachable.
-            // If it ever changes, over-invalidate rather than serve a stale
-            // descriptor: correctness first, the cache refills on the next read.
-            self.cache.invalidate_all();
-        }
     }
 
     /// Drop every descriptor for `volume` — the whole bucket is gone
     /// (rustfs/backlog#1177).
     fn invalidate_volume(&self, volume: &str) {
         self.generation.fetch_add(1, Ordering::AcqRel);
-        let volume = volume.to_owned();
+        self.invalidate_matching(|k| k.volume == volume);
-        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| k.volume == volume;
-        if self.cache.invalidate_entries_if(matches).is_err() {
-            self.cache.invalidate_all();
-        }
     }
 
     /// Drop ALL descriptors — this disk instance is being retired
@@ -5848,6 +5906,7 @@ impl FdCache {
     fn clear(&self) {
         self.generation.fetch_add(1, Ordering::AcqRel);
         self.cache.invalidate_all();
+        self.schedule_maintenance();
     }
 
     /// Unconditional insert. Production reads go through `insert_if_fresh` (the
@@ -5856,7 +5915,17 @@ impl FdCache {
     #[cfg(test)]
     async fn insert(&self, key: FdKey, file: Arc<std::fs::File>) {
         let len = file.metadata().map(|metadata| metadata.len()).unwrap_or_default();
-        self.cache.insert(key, Arc::new(FdCacheEntry { file, len })).await;
+        self.has_inserted_entries.store(true, Ordering::Release);
+        self.cache
+            .insert(
+                key,
+                Arc::new(FdCacheEntry {
+                    file,
+                    len,
+                    invalidated: AtomicBool::new(false),
+                }),
+            )
+            .await;
     }
 
     #[cfg(test)]
@@ -6455,6 +6524,7 @@ impl UringBackend {
                 let file = Arc::new(FdCacheEntry {
                     file: Arc::new(file),
                     len,
+                    invalidated: AtomicBool::new(false),
                 });
                 if let (Some((cache, key)), Some(gen_at_open)) = (cache_entry, gen_at_open) {
                     cache.insert_if_fresh(key, Arc::clone(&file), gen_at_open).await;
@@ -6627,6 +6697,7 @@ impl UringBackend {
                 let entry = Arc::new(FdCacheEntry {
                     file: Arc::new(file),
                     len,
+                    invalidated: AtomicBool::new(false),
                 });
                 if let (Some(cache), Some(gen_at_open)) = (cache, gen_at_open) {
                     cache.insert_if_fresh(key.clone(), Arc::clone(&entry), gen_at_open).await;
@@ -34128,11 +34199,217 @@ mod test {
         });
     }
 
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_invalidation_releases_descriptors_without_further_operations() {
+        let cache = FdCache::new();
+        for round in 0..4 {
+            let file = Arc::new(tempfile::tempfile().expect("create a descriptor to invalidate"));
+            let released = Arc::downgrade(&file);
+            let parts = if round == 1 || round == 2 { FD_CACHE_CAPACITY } else { 1 };
+            for part in 0..parts {
+                cache
+                    .insert(
+                        FdKey {
+                            volume: "bucket".into(),
+                            path: format!("object-{round}/part.{part}"),
+                            direct: false,
+                        },
+                        file.clone(),
+                    )
+                    .await;
+            }
+            drop(file);
+            if round % 2 == 0 {
+                cache.invalidate_under("bucket", &format!("object-{round}"));
+            } else {
+                cache.invalidate_volume("bucket");
+            }
+            tokio::time::timeout(std::time::Duration::from_secs(10), async {
+                while released.upgrade().is_some() {
+                    tokio::task::yield_now().await;
+                }
+            })
+            .await
+            .expect("invalidation must release descriptors without another cache operation");
+        }
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_invalidation_rejects_entries_before_maintenance() {
+        let cache = FdCache::new();
+        let key = FdKey {
+            volume: "bucket".into(),
+            path: "object/part.1".into(),
+            direct: false,
+        };
+        cache
+            .insert(key.clone(), Arc::new(tempfile::tempfile().expect("create a cached descriptor")))
+            .await;
+        assert!(cache.get(&key).await.is_some());
+        cache.invalidate_under("bucket", "object");
+
+        // No yield has allowed the maintenance task to remove the entry yet.
+        assert!(cache.cache.contains_key(&key));
+        assert!(cache.get(&key).await.is_none(), "a marked descriptor must never be served");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_idle_maintenance_releases_expired_descriptors() {
+        let cache = FdCache::with_ttl(std::time::Duration::from_millis(50));
+        let file = Arc::new(tempfile::tempfile().expect("create a descriptor with a short TTL"));
+        let released = Arc::downgrade(&file);
+        let key = FdKey {
+            volume: "retained".into(),
+            path: "part.1".into(),
+            direct: false,
+        };
+        cache.insert(key.clone(), file).await;
+        cache.invalidate_volume("other");
+        assert!(cache.get(&key).await.is_some(), "invalidation must preserve other volumes");
+
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while released.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("periodic maintenance must release expired descriptors while idle");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_maintenance_exits_when_cache_is_dropped() {
+        let mut cache = FdCache::new();
+        let file = Arc::new(tempfile::tempfile().expect("create a descriptor to retire"));
+        let released = Arc::downgrade(&file);
+        cache
+            .insert(
+                FdKey {
+                    volume: "bucket".into(),
+                    path: "part.1".into(),
+                    direct: false,
+                },
+                file,
+            )
+            .await;
+        cache.invalidate_volume("bucket");
+        let maintenance = cache.maintenance_task.take().expect("invalidation starts maintenance");
+        drop(cache);
+        tokio::time::timeout(std::time::Duration::from_secs(10), maintenance)
+            .await
+            .expect("dropping the cache must stop its maintenance task")
+            .expect("maintenance must complete without a panic");
+        assert!(released.upgrade().is_none(), "shutdown must release cached descriptors");
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_maintenance_survives_request_cancellation() {
+        let cache = FdCache::new();
+        let file = Arc::new(tempfile::tempfile().expect("create a descriptor to invalidate"));
+        let released = Arc::downgrade(&file);
+        cache
+            .insert(
+                FdKey {
+                    volume: "bucket".into(),
+                    path: "part.1".into(),
+                    direct: false,
+                },
+                file,
+            )
+            .await;
+        let mut request = Box::pin(async {
+            cache.invalidate_volume("bucket");
+            std::future::pending::<()>().await;
+        });
+        let mut context = std::task::Context::from_waker(std::task::Waker::noop());
+        assert!(std::future::Future::poll(request.as_mut(), &mut context).is_pending());
+        drop(request);
+
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while released.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("maintenance must continue after the invalidating request is cancelled");
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn fd_cache_construction_does_not_require_a_runtime() {
+        let cache = FdCache::new();
+        assert!(cache.maintenance_task.get().is_none());
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_invalidation_fences_an_open_before_the_first_insert() {
+        for mode in 0..2 {
+            let cache = FdCache::new();
+            let key = FdKey {
+                volume: "bucket".into(),
+                path: "object/part.1".into(),
+                direct: false,
+            };
+            let opened = Arc::new(FdCacheEntry {
+                file: Arc::new(tempfile::tempfile().expect("open the descriptor before invalidation")),
+                len: 0,
+                invalidated: AtomicBool::new(false),
+            });
+            let generation = cache.generation();
+            match mode {
+                0 => cache.invalidate_under("bucket", "object"),
+                _ => cache.invalidate_volume("bucket"),
+            }
+            cache.insert_if_fresh(key.clone(), opened, generation).await;
+            assert!(cache.get(&key).await.is_none(), "mode {mode} must reject a pre-invalidation open");
+
+            let fresh = Arc::new(FdCacheEntry {
+                file: Arc::new(tempfile::tempfile().expect("open a fresh descriptor after invalidation")),
+                len: 0,
+                invalidated: AtomicBool::new(false),
+            });
+            cache.insert_if_fresh(key.clone(), fresh.clone(), cache.generation()).await;
+            let cached = cache.get(&key).await.expect("a post-invalidation open must remain cacheable");
+            assert!(Arc::ptr_eq(&cached, &fresh));
+        }
+    }
+
+    #[cfg(unix)]
+    #[tokio::test]
+    async fn fd_cache_clear_releases_descriptors_without_further_reads() {
+        let cache = FdCache::new();
+        let file = Arc::new(tempfile::tempfile().expect("create a descriptor to retire"));
+        let released = Arc::downgrade(&file);
+        cache
+            .insert(
+                FdKey {
+                    volume: "bucket".into(),
+                    path: "part.1".into(),
+                    direct: false,
+                },
+                file,
+            )
+            .await;
+        cache.clear();
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while released.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("clear must release retired descriptors without another cache operation");
+    }
+
     /// Prefix invalidation must respect component boundaries: dropping the
     /// descriptors for object `a/b` must not also drop those for `a/bc`.
-    /// moka applies the predicate lazily, so this also pins the guarantee we rely
+    /// A `get` after `invalidate_under` must reject matched entries while
-    /// on — a `get` after `invalidate_under` never returns a matched entry.
+    /// background maintenance removes them.
-    #[cfg(target_os = "linux")]
+    #[cfg(unix)]
     #[tokio::test(flavor = "multi_thread")]
     async fn fd_cache_prefix_invalidation_respects_component_boundary() {
         use std::fs::File;
@@ -34368,8 +34645,32 @@ mod test {
             }
         }
 
-        // Invalidating by the object prefix drops the cached descriptor.
+        let key = FdKey {
+            volume: volume.into(),
+            path: object.into(),
+            direct: false,
+        };
+        let released = {
+            let entry = cache
+                .get(&key)
+                .await
+                .expect("repeated reads must retain the cached descriptor");
+            Arc::downgrade(&entry.file)
+        };
+
+        // Reads reject invalidated entries immediately; maintenance releases the descriptor.
         backend.invalidate_cached_fds_under(volume, "obj/abc");
+        assert!(
+            cache.get(&key).await.is_none(),
+            "prefix invalidation must immediately reject the cached descriptor"
+        );
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while released.upgrade().is_some() {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("prefix invalidation must release the descriptor without further cache operations");
         assert_eq!(cache.entry_count().await, 0, "prefix invalidation must drop the cached descriptor");
     }
 

crates/ecstore/src/disk/local/multipart_source_state.rs

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

crates/ecstore/src/disk/os.rs

@@ -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<()>>>>> =
+static DISK_VOLUME_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<RwLock<()>>> = LazyLock::new(ShardedPathLockRegistry::new);
-    LazyLock::new(|| Mutex::new(HashMap::new()));
 type NamespaceMutationLock = AsyncMutex<()>;
-type NamespaceMutationLockRegistry = HashMap<PathBuf, Weak<NamespaceMutationLock>>;
+static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<ShardedPathLockRegistry<NamespaceMutationLock>> =
-static DISK_NAMESPACE_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
+    LazyLock::new(ShardedPathLockRegistry::new);
-    LazyLock::new(|| Mutex::new(HashMap::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();
+    DISK_FILE_SYNC_LIMITERS.get_or_create(root, || Semaphore::new(MAX_PARALLEL_FILE_SYNCS))
-    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
 }
 
 // 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>>>> =
+struct RegisteredDiskBlockingQueue {
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+    #[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();
+    #[cfg(unix)]
-    queues.retain(|_, q| q.strong_count() > 0);
+    let root_key = canonical_unix_path_key(root);
-    let queue = queues.get(root).and_then(Weak::upgrade).unwrap_or_else(|| {
+    #[cfg(unix)]
-        let queue = DiskBlockingQueue::new();
+    let io_root_key = canonical_unix_path_key(io_root);
-        queues.insert(root.to_path_buf(), Arc::downgrade(&queue));
+
-        queue
+    let mut queues = DISK_BLOCKING_QUEUES.write();
-    });
+    queues.retain(|entry| entry.queue.strong_count() > 0);
-    if io_root != root {
+    let queue = queues
-        queues.insert(io_root.to_path_buf(), Arc::downgrade(&queue));
+        .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)
+    #[cfg(unix)]
-}
+    let root_key = canonical_unix_path_key(root);
-
-fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
     {
-        let queues = DISK_BLOCKING_QUEUES.lock();
+        let queues = DISK_BLOCKING_QUEUES.read();
-        for (root, weak) in queues.iter() {
+        for entry in queues.iter() {
-            if path.starts_with(root)
+            #[cfg(unix)]
-                && let Some(queue) = weak.upgrade()
+            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};
+    register_disk_blocking_queue(root, root)
-    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);
+fn disk_blocking_queue_for_path(path: &Path) -> Arc<DiskBlockingQueue> {
-    Arc::clone(&FALLBACK_BLOCKING_QUEUES[idx])
+    #[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);
+    #[cfg(unix)]
-    let mut locks = DISK_VOLUME_MUTATION_LOCKS.lock();
+    {
-    locks.retain(|_, lock| lock.strong_count() > 0);
+        let vol_bytes = volume.as_bytes();
-    if let Some(lock) = locks.get(&key).and_then(Weak::upgrade) {
+        if !vol_bytes.is_empty()
-        return lock;
+            && 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 key = root.join(volume);
-    let lock = Arc::new(RwLock::new(()));
+    DISK_VOLUME_MUTATION_LOCKS.get_or_create(&key, || RwLock::new(()))
-    locks.insert(key, Arc::downgrade(&lock));
-    lock
 }
 
 fn disk_namespace_mutation_lock(path: &Path) -> Arc<NamespaceMutationLock> {
-    let mut locks = DISK_NAMESPACE_MUTATION_LOCKS.lock();
+    DISK_NAMESPACE_MUTATION_LOCKS.get_or_create(path, || AsyncMutex::new(()))
-    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
 }
 
-static DISK_METADATA_MUTATION_LOCKS: LazyLock<Mutex<NamespaceMutationLockRegistry>> =
+#[cfg(unix)]
-    LazyLock::new(|| Mutex::new(HashMap::new()));
+#[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 lock = DISK_METADATA_MUTATION_LOCKS.get_or_create(object, || AsyncMutex::new(()));
-        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
-        }
-    };
     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);
+    #[cfg(unix)]
-    let (namespace_guard, staging_guard) = match staging_prefix {
+    let (namespace_guard, staging_guard) = {
-        Some(source) if source < destination_object => {
+        let dst_key = canonical_unix_path_key(destination_object);
-            let staging = disk_namespace_mutation_lock(source).lock_owned().await;
+        let destination_lock = disk_namespace_mutation_lock_by_key(dst_key.as_ref());
-            (destination_lock.lock_owned().await, Some(staging))
+        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;
+    #[cfg(not(unix))]
-            (destination, Some(disk_namespace_mutation_lock(source).lock_owned().await))
+    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"))]

crates/ecstore/src/services/rebalance/worker.rs

@@ -318,6 +318,21 @@ fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool {
     }
 }
 
+fn is_rebalance_metadata_admission_retryable(err: &Error) -> bool {
+    match rebalance_error_source(err) {
+        Error::Lock(err) => is_rebalance_transient_lock_error(err),
+        // No metadata was written when lock admission failed. Each attempt
+        // must acquire its quorum again before loading or saving this run.
+        Error::NamespaceLockQuorumUnavailable {
+            mode: "write",
+            bucket,
+            object,
+            ..
+        } => bucket == crate::disk::RUSTFS_META_BUCKET && object == REBAL_META_NAME,
+        _ => false,
+    }
+}
+
 fn is_rebalance_transient_io_error(err: &std::io::Error) -> bool {
     if err.kind() == std::io::ErrorKind::TimedOut {
         return true;
@@ -385,9 +400,7 @@ where
         match result {
             Ok(value) => return Ok(value),
             Err(err) => {
-                if attempt.saturating_add(1) >= max_attempts.max(1)
+                if attempt.saturating_add(1) >= max_attempts.max(1) || !is_rebalance_metadata_admission_retryable(&err) {
-                    || !matches!(rebalance_error_source(&err), Error::Lock(lock_err) if is_rebalance_transient_lock_error(lock_err))
-                {
                     return Err(err);
                 }
                 #[cfg(test)]
@@ -719,6 +732,82 @@ mod error_source_tests {
         }
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_recovers_after_write_quorum_failure() {
+        let mut attempts = 0;
+        retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(if attempts == 1 {
+                resolve_rebalance_meta_save_result(
+                    Err(rebalance_meta_lock_error(
+                        rustfs_lock::LockError::QuorumNotReached {
+                            required: 1,
+                            achieved: 0,
+                        },
+                        "write",
+                    )),
+                    "stop_rebalance",
+                )
+            } else {
+                Ok(())
+            })
+        })
+        .await
+        .expect("a later quorum-protected admission must be allowed to succeed");
+        assert_eq!(attempts, 2);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_retains_persistent_write_quorum_failure() {
+        for max_attempts in [0, 1, 3] {
+            let mut attempts = 0;
+            let result = retry_rebalance_metadata_access(None, max_attempts, || {
+                attempts += 1;
+                std::future::ready(Err::<(), _>(rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )))
+            })
+            .await;
+            let error = result.expect_err("exhausted admission attempts must never acknowledge a write");
+            assert_eq!(attempts, max_attempts.max(1));
+            assert!(matches!(
+                rebalance_error_source(&error),
+                Error::NamespaceLockQuorumUnavailable {
+                    mode: "write",
+                    required: 1,
+                    achieved: 0,
+                    ..
+                }
+            ));
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_stops_after_admission_is_cancelled() {
+        let mut attempts = 0;
+        let result = retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(Err::<(), _>(if attempts == 1 {
+                rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )
+            } else {
+                Error::OperationCanceled
+            }))
+        })
+        .await;
+        assert_eq!(attempts, 2, "a stale or cancelled run cannot enter another admission attempt");
+        assert!(matches!(result, Err(Error::OperationCanceled)));
+    }
+
     #[tokio::test]
     async fn rebalance_metadata_retry_does_not_retry_permanent_or_untyped_errors() {
         for err in [
@@ -731,6 +820,20 @@ mod error_source_tests {
                 required: 3,
                 achieved: 2,
             },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: "data-bucket".to_string(),
+                object: REBAL_META_NAME.to_string(),
+                required: 1,
+                achieved: 0,
+            },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
+                object: "other-metadata.bin".to_string(),
+                required: 1,
+                achieved: 0,
+            },
             Error::other("stale rebalance run rejected: lock acquisition timed out"),
             Error::other("rebalance distributed run fence lost"),
             Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),

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

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

crates/filemeta/src/metacache.rs

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

crates/iam/src/oidc/provider.rs

@@ -16,9 +16,10 @@ use super::{
     transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics},
 };
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
-use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
+use openidconnect::core::{CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
 use openidconnect::{
-    AsyncHttpClient, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, ProviderMetadataWithLogout, TokenUrl,
+    AsyncHttpClient, ClaimsVerificationError, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
+    ProviderMetadataWithLogout, SignatureVerificationError, TokenUrl,
 };
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -212,6 +213,35 @@ impl DiscoveredProviderMetadata {
     }
 }
 
+pub(super) fn verify_id_token_claims<'a, N: NonceVerifier>(
+    token: &'a CoreIdToken,
+    verifier: &CoreIdTokenVerifier<'_>,
+    nonce_verifier: N,
+) -> Result<&'a CoreIdTokenClaims, ClaimsVerificationError> {
+    let expected_length = match token.signing_alg().map_err(ClaimsVerificationError::SignatureVerification)? {
+        CoreJwsSigningAlgorithm::HmacSha256 => 32,
+        CoreJwsSigningAlgorithm::HmacSha384 => 48,
+        CoreJwsSigningAlgorithm::HmacSha512 => 64,
+        _ => return token.claims(verifier, nonce_verifier),
+    };
+    // RUSTFS_COMPAT_TODO(openidconnect-hmac-signature-width): Reject widths that upstream converts to fixed-size arrays. Remove after upstream returns verification errors for arbitrary HMAC signature lengths.
+    // openidconnect 4.1.0 converts HMAC signatures to fixed-size arrays before verifying them.
+    let invalid_length = || {
+        ClaimsVerificationError::SignatureVerification(SignatureVerificationError::CryptoError(
+            "invalid HMAC signature length".to_owned(),
+        ))
+    };
+    let compact = token.to_string();
+    let (_, signature) = compact.rsplit_once('.').ok_or_else(invalid_length)?;
+    let signature = base64_simd::URL_SAFE_NO_PAD
+        .decode_to_vec(signature)
+        .map_err(|_| invalid_length())?;
+    if signature.len() != expected_length {
+        return Err(invalid_length());
+    }
+    token.claims(verifier, nonce_verifier)
+}
+
 // This adapter is used only for discovery/JWKS fetches, never token exchange.
 struct JwksAcceptClient<'a> {
     inner: &'a ReqwestHttpClient,

crates/iam/src/oidc/provider_tests.rs

@@ -242,7 +242,7 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
     let now = time::OffsetDateTime::now_utc().unix_timestamp();
     let payload = serde_json::json!({
         "iss": config.config_url, "sub": "synthetic-subject", "aud": config.client_id,
-        "iat": now, "exp": now + 300,
+        "iat": now, "exp": now + 300, "nonce": "synthetic-nonce",
     });
 
     for (algorithm, name, signature_length) in [
@@ -275,22 +275,28 @@ fn configured_verifier_rejects_malformed_hmac_signature_lengths() {
                 .expect("parse configured provider metadata");
             let verifier = metadata.verifier(&config);
             assert!(
-                valid.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
+                verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
                 "valid {name} token must pass for workload={workload}"
             );
+            assert!(verify_id_token_claims(&valid, &verifier, &Nonce::new("synthetic-nonce".into())).is_ok());
+            assert!(verify_id_token_claims(&valid, &verifier, &Nonce::new("wrong-nonce".into())).is_err());
 
             for length in [1, 0, signature_length - 1, signature_length, signature_length + 1, 128] {
                 let signature = base64_simd::URL_SAFE_NO_PAD.encode_to_string(vec![0_u8; length]);
                 let malformed: CoreIdToken = format!("{signing_input}.{signature}")
                     .parse()
-                    .expect("invalid signature length must still reach cryptographic verification");
+                    .expect("invalid signature length must remain parseable before verification");
                 assert!(
-                    malformed.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_err(),
+                    verify_id_token_claims(&malformed, &verifier, |_: Option<&Nonce>| Ok(())).is_err(),
                     "{name} signature length {length} must return an error for workload={workload}"
                 );
+                assert!(
+                    verify_id_token_claims(&malformed, &verifier, &Nonce::new("synthetic-nonce".into())).is_err(),
+                    "{name} signature length {length} must return an error with a nonce for workload={workload}"
+                );
             }
             assert!(
-                valid.claims(&verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
+                verify_id_token_claims(&valid, &verifier, |_: Option<&Nonce>| Ok(())).is_ok(),
                 "invalid inputs must not prevent later valid {name} verification"
             );
         }

crates/iam/src/oidc/runtime.rs

@@ -16,7 +16,7 @@ use super::{
         OidcConfigQuery, OidcConfigSnapshot, OidcProviderConfig, OidcProviderSummary, OidcSiteReplicationSnapshot,
         SourcedOidcProviderConfig, load_oidc_config_snapshot,
     },
-    provider::{ProviderRuntime, discover_provider_for_recovery},
+    provider::{ProviderRuntime, discover_provider_for_recovery, verify_id_token_claims},
     state::{OidcAuthSession, OidcLogoutSession, OidcStateStore},
     transport::{OidcExtraRootCaProvider, ReqwestHttpClient, inspect_token_response_body, oidc_http_error_diagnostics},
 };
@@ -535,7 +535,7 @@ impl OidcSys {
         let verifier = client
             .id_token_verifier()
             .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-        let verified = id_token.claims(&verifier, &Nonce::new(session.nonce.clone()));
+        let verified = verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()));
         if let Err(e) = verified {
             let verification_error = e.to_string();
             warn!(
@@ -583,8 +583,7 @@ impl OidcSys {
             let verifier = client
                 .id_token_verifier()
                 .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-            id_token
+            verify_id_token_claims(id_token, &verifier, &Nonce::new(session.nonce.clone()))
-                .claims(&verifier, &Nonce::new(session.nonce.clone()))
                 .map_err(|retry_err| {
                     error!(
                         event = EVENT_OIDC_DIAGNOSTICS,
@@ -767,7 +766,7 @@ impl OidcSys {
             .metadata
             .verifier(&config)
             .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-        if let Err(e) = id_token.claims(&verifier, |_: Option<&Nonce>| Ok(())) {
+        if let Err(e) = verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(())) {
             state = self
                 .provider_runtime
                 .refresh_provider_state(&provider_id, &config)
@@ -780,8 +779,7 @@ impl OidcSys {
                 .metadata
                 .verifier(&config)
                 .set_other_audience_verifier_fn(|aud| trusted_aud(&config.other_audiences, aud));
-            id_token
+            verify_id_token_claims(&id_token, &verifier, |_: Option<&Nonce>| Ok(()))
-                .claims(&verifier, |_: Option<&Nonce>| Ok(()))
                 .map_err(|retry_err| format!("ID token verification failed after JWKS refresh: {retry_err}"))?;
         }
 

crates/io-metrics/src/internode_metrics.rs

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

crates/protocols/Cargo.toml

@@ -30,6 +30,7 @@ workspace = true
 
 [features]
 default = []
+fips = ["rustfs-tls-runtime?/fips"]
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",

crates/protocols/src/ftps/server.rs

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

crates/protocols/src/lib.rs

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

crates/protocols/src/webdav/server.rs

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

crates/s3-client/Cargo.toml

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

crates/s3-client/src/transition_api.rs

@@ -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() {
+        rustfs_tls_runtime::ensure_crypto_provider().map_err(std::io::Error::other)?;
-            // 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");
-        }
 
         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
+    #[cfg(feature = "fips")]
-    /// an error that surfaces to callers (the race-safe `get_default` check guards the install).
     #[test]
-    fn provider_install_is_idempotent() {
+    fn fips_transition_rejects_incompatible_host_provider() {
-        // Install once (may already be set by another test in this binary — that's fine).
+        const CHILD: &str = "RUSTFS_TEST_TRANSITION_FIPS_PROVIDER";
-        let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+        if std::env::var_os(CHILD).is_some() {
-
+            let mut provider = rustfs_tls_runtime::default_crypto_provider();
-        // A second install attempt on an already-set provider must not panic.
+            provider.kx_groups = vec![rustls::crypto::aws_lc_rs::kx_group::X25519];
-        let outcome = std::panic::catch_unwind(|| {
+            provider.install_default().expect("install incompatible host provider");
-            if rustls::crypto::CryptoProvider::get_default().is_none() {
+            let runtime = tokio::runtime::Builder::new_current_thread()
-                let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+                .enable_all()
-            }
+                .build()
-            // If a default is already present, the branch above is simply skipped.
+                .expect("test runtime");
-        });
+            let error = runtime
-        assert!(outcome.is_ok(), "provider install guard must not panic when a provider is already set");
+                .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]

crates/s3select-api/src/object_store.rs

@@ -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,
+        start: i64::try_from(range.start).unwrap_or(i64::MAX),
-        end: range.end.saturating_sub(1) as i64,
+        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"))]);

crates/targets/Cargo.toml

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

crates/targets/src/check.rs

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

crates/targets/src/target/amqp.rs

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

crates/targets/src/target/mod.rs

@@ -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
+/// Target modules may be the first to construct a TLS client. In a FIPS build,
-/// each may be the first to perform a TLS handshake. Idempotent: if a
+/// an incompatible provider installed by the host is rejected.
-/// provider is already registered, returns immediately.
+pub(crate) fn ensure_rustls_provider_installed() -> Result<(), TargetError> {
-pub(crate) fn ensure_rustls_provider_installed() {
+    rustfs_tls_runtime::ensure_crypto_provider()
-    if rustls::crypto::CryptoProvider::get_default().is_some() {
+        .map(|_| ())
-        return;
+        .map_err(|err| TargetError::Configuration(err.to_string()))
-    }
-    if let Err(err) = rustls::crypto::aws_lc_rs::default_provider().install_default() {
-        debug!("rustls provider already installed or unavailable: {err:?}");
-    }
 }
 
 #[cfg(test)]

crates/targets/src/target/mqtt.rs

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

crates/targets/src/target/mysql.rs

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

crates/targets/src/target/nats/mod.rs

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

crates/targets/src/target/postgres.rs

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

crates/targets/src/target/pulsar.rs

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

crates/targets/src/target/redis.rs

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

crates/targets/src/target/webhook.rs

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

crates/tls-runtime/src/certs.rs

@@ -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())
+    let certified =
-        .map_err(|source| TlsRuntimeError::InvalidCertifiedKey {
+        CertifiedKey::from_der(certs.to_vec(), key.clone_key(), &crate::default_crypto_provider()).map_err(|source| {
-            name: name.to_string(),
+            TlsRuntimeError::InvalidCertifiedKey {
-            source,
+                name: name.to_string(),
+                source,
+            }
         })?;
     certified
         .keys_match()

crates/tls-runtime/src/lib.rs

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

crates/tls-runtime/src/provider.rs

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

crates/zstd-compat/Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zstd"
-version = "0.13.3"
+version = "0.13.4"
 publish = false
 edition.workspace = true
 license.workspace = true
@@ -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-array = "60.1.0"
-arrow-ipc = { version = "59.3.2", features = ["zstd"] }
+arrow-ipc = { version = "60.1.0", features = ["zstd"] }
-arrow-schema = "59.3.2"
+arrow-schema = "60.1.0"
-parquet = { version = "59.3.2", default-features = false, features = ["arrow", "zstd"] }
+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
 

crates/zstd-compat/README.md

@@ -6,6 +6,6 @@ This private package satisfies the version constraints of Avro, Arrow IPC, Parqu
 
 The fixed implementation closes RUSTSEC-2026-0391, RUSTSEC-2026-0398, and RUSTSEC-2026-0403. A successful advisory scan alone is insufficient evidence: cargo-deny skips local path packages. The dependency-graph test therefore requires this adapter's sole runtime dependency to be the exact fixed registry implementation and rejects vulnerable registry codec versions. Security tests exercise abandoned stream contexts, malformed legacy frames, and seekable offsets past EOF. Consumer tests retain zstandard support in all four upstream formats.
 
-Run `cargo nextest run --locked -p zstd@0.13.3` for the focused tests. The seekable feature is enabled by a test-only dependency to cover the optional vulnerable API without enabling it in production.
+Run `cargo nextest run --locked -p zstd@0.13.4` for the focused tests. The seekable feature is enabled by a test-only dependency to cover the optional vulnerable API without enabling it in production.
 
 Remove this package and the root Cargo patch when all consumers accept the fixed zstd release line directly. The cleanup marker is registered in the architecture compatibility register.

crates/zstd-compat/tests/dependency_graph.rs

@@ -57,7 +57,7 @@ fn legacy_constraints_resolve_only_to_the_fixed_registry_implementation() {
         let version = Version::parse(package["version"].as_str().expect("package version")).expect("valid package semver");
         if name == "zstd" && package.get("source").is_none() {
             adapters += 1;
-            assert_eq!(version, Version::new(0, 13, 3));
+            assert_eq!(version, Version::new(0, 13, 4));
             let runtime = package["dependencies"].as_array().expect("adapter dependency edges");
             assert!(runtime.iter().any(|value| value.as_str() == Some("zstd 0.14.3")));
         } else {

deny.toml

@@ -36,10 +36,6 @@ allow-git = [
     # Official s3s repository pinned for Iceberg SigV4 path encoding support.
     # owner: GatewayJ  review: 2026-10
     "https://github.com/s3s-project/s3s",
-    # Upstream OIDC/OAuth2 crypto migrations, pinned to reviewed PR commits.
-    # owner: rustfs-maintainers  review: 2026-10
-    "https://github.com/ramosbugs/openidconnect-rs",
-    "https://github.com/ramosbugs/oauth2-rs",
     # Completion ownership patch fixed to rustfs/uring#20.
     # owner: overtrue  review: 2026-10
     "https://github.com/rustfs/uring.git",

docs/architecture/compat-cleanup-register.md

@@ -11,6 +11,8 @@
 
 ## Open Items
 
+- `openidconnect-hmac-signature-width` ID token verification guard: openidconnect 4.1.0 converts decoded HMAC signatures to fixed-size arrays, which panic on unexpected lengths. Check HS256, HS384, and HS512 signature widths before delegating all cryptographic, issuer, audience, expiry, and nonce checks to the upstream verifier. Remove the guard after a supported upstream release returns verification errors for arbitrary HMAC signature lengths.
+
 - `zstd-013-fixed-backend` dependency constraint bridge: Avro, Arrow IPC, Parquet, and Kafka still require zstd 0.13. The local adapter re-exports the fixed registry zstd 0.14.3 implementation and its zstd-safe types. Remove the adapter and Cargo patch after every dependency accepts the fixed zstd line without the 0.13 bridge.
 
 - `8565` historical rebalance replacement recovery: older schema-seven intents can exhaust their retry budget before format repair when retained rebalance metadata describes removed pools. After validating blank targets, terminal historical topology, and ownership, grant one durable handoff and one rebuild attempt. Remove after supported direct upgrades no longer admit snapshots lacking replacement_preformat_handoff_retry_done and all retained affected intents have been resolved.

docs/architecture/s3-tables-support-matrix.md

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

docs/operations/kms-cryptographic-compliance.md

@@ -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)).
+- **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 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.
+- **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`.
+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.
-- **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).
+```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` |
+| 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 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 |
+| 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

docs/testing/README.md

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

docs/testing/ci-gates.md

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

fuzz/Cargo.lock

@@ -273,9 +273,9 @@ dependencies = [
 
 [[package]]
 name = "aws-sdk-s3"
-version = "1.156.0"
+version = "1.157.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1072c9aa20159b9b417115342d38c4c53108fc98e5359fcb86b7d4575747eb59"
+checksum = "6ea9794c741520d8baec9462bcd0ad12164d4f5f78d008a8da9d6a0b7222caa8"
 dependencies = [
  "arc-swap",
  "aws-credential-types",
@@ -1837,18 +1837,18 @@ dependencies = [
 
 [[package]]
 name = "hotpath"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b39e941010d86063ff233bf8325269bf68097fb24ba67000fbddc0a410e8056e"
+checksum = "d9632c1fd213e63c1b88d8f3e59543da4999b17f859cc1e1d67cf8f2207e8935"
 dependencies = [
  "hotpath-macros",
 ]
 
 [[package]]
 name = "hotpath-macros"
-version = "0.30.2"
+version = "0.30.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a24f5179ff680132003306354efd0ac8cde12e87b1a788719c836e91af35cb58"
+checksum = "5d92fdfec803932e5764ea93c201cf7c6c2d8cff1871da44f64f5094c9c3ef8a"
 
 [[package]]
 name = "http"
@@ -2418,9 +2418,9 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.34"
+version = "0.4.35"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f9f8bd3e56ce4dfc153cf470fffbfa98c7620958b312ca5c3a4b8d5181fd13c6"
+checksum = "5d15f86d12da44cd8d4ec3c73984537eb7f4cadcb9776a575a0a8a05ca11005c"
 
 [[package]]
 name = "lru"
@@ -4134,8 +4134,7 @@ dependencies = [
 [[package]]
 name = "rustfs-uring"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
+source = "git+https://github.com/rustfs/uring.git?rev=eb8fc160e71d804427de1e905f5c226d580f71bf#eb8fc160e71d804427de1e905f5c226d580f71bf"
-checksum = "f7deda770b7981ee3bfaa202ecd18625fc794c962b8d58cd4f436952b7e66dcf"
 dependencies = [
  "io-uring",
  "libc",

rustfs/Cargo.toml

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

rustfs/src/startup_preflight.rs

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

rustfs/src/startup_runtime.rs

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

rustfs/src/startup_runtime_hooks.rs

@@ -14,10 +14,7 @@
 
 use crate::license::license_status;
 use crate::startup_runtime_sources;
-#[cfg(not(feature = "fips"))]
+use rustfs_tls_runtime::{CryptoProviderStatus, ensure_crypto_provider};
-use rustls::crypto::aws_lc_rs::default_provider;
-#[cfg(feature = "fips")]
-use rustls::crypto::default_fips_provider as default_provider;
 use std::future::Future;
 use std::io::Result;
 use tracing::{debug, info, warn};
@@ -108,31 +105,24 @@ where
     shutdown();
 }
 
-fn build_default_crypto_provider() -> rustls::crypto::CryptoProvider {
+pub(crate) fn install_default_crypto_provider() -> Result<()> {
-    let provider = default_provider();
+    if ensure_crypto_provider().map_err(std::io::Error::other)? == CryptoProviderStatus::AlreadyInstalled {
-    #[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() {
         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() {
+fn install_embedded_default_crypto_provider() -> Result<()> {
-    if let Err(err) = build_default_crypto_provider().install_default() {
+    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]

rustfs/src/table_catalog/iceberg/commit.rs

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

rustfs/src/table_catalog/model.rs

@@ -1415,6 +1415,7 @@ pub(crate) struct TableCatalogDiagnosticsReport {
 #[serde(rename_all = "SCREAMING_SNAKE_CASE")]
 pub(crate) enum TableCommitRecoveryState {
     Committed,
+    Superseded,
     StagedBeforeTableUpdate,
     FinalizationRequired,
     IdempotencyIndexRepairRequired,

rustfs/src/table_catalog/store/object.rs

@@ -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(
+            let recovery_entry =
-                &entry,
+                table_commit_recovery_entry_from_history(&entry, commit_log, idempotency_commit.as_ref(), &history);
-                commit_log,
-                idempotency_commit.as_ref(),
-                history.proves_committed(commit_log),
-            );
             if matches!(
                 recovery_entry.recovery_state,
                 TableCommitRecoveryState::FinalizationRequired | TableCommitRecoveryState::IdempotencyIndexRepairRequired

rustfs/src/table_catalog/tests.rs

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

rustfs/src/table_catalog/tests/superseded_commit_recovery.rs

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

rustfs/tests/connect_perf_object.rs

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

scripts/ci_gate.py

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

scripts/test_ci_plan.py

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

scripts/test_nightly_candidate.py

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