Merge branch 'main' into ci/validation-sweep-boto3

Keep main's validation sweep workflow where both sides edited the same steps. The Python 3.12 venv install from validation_ilm/requirements.txt is already on main and remains in the sweep job, with the scheduled nightly pin, chain envelope read, ssh -n, AUTO_TESTING_SHA, and artifact warn/run path. Co-authored-by: hector <majinghe@users.noreply.github.com>
fix(ecstore): release retained file descriptor cache invalidations (#9112 )
2026-11-03 01:23:53 +00:00 · 2026-11-02 12:39:52 +00:00 · 2026-11-02 10:56:08 +00:00 · 2026-11-02 18:07:28 +08:00 · 2026-11-02 16:38:53 +08:00 · 2026-11-02 08:34:13 +00:00
22 changed files with 838 additions and 302 deletions
@@ -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())
-          manifest = candidate["manifest"]
+          chain = json.loads((Path(os.environ["RUNNER_TEMP"]) / "validation-sweep-candidate.json").read_text())
+          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)
@@ -107,21 +107,6 @@ jobs:
        shell: pwsh
        run: cargo test -p rustfs-ecstore --lib multipart_owned_record_parent_barriers_ -- --nocapture

-      # Native multipart completion publishes its payload from a different
-      # parent than its metadata, and abort renames under a precondition; both
-      # need their own Windows handle discipline.
-      - name: Test native multipart complete and abort publication
-        shell: pwsh
-        run: >-
-          cargo test -p rustfs-ecstore --lib --
-          native_abort_missing_state_with_actual_owned_records_never_becomes_legacy_or_missing
-          native_abort_pending_intent_is_not_a_delete_certificate_and_raw_cleanup_preserves_exact_evidence
-          --nocapture
-
-      - name: Test directory removal fallback
-        shell: pwsh
-        run: "cargo test -p rustfs-ecstore --lib disk::fs::tests::test_remove -- --nocapture"
-
      - name: Test Windows path conversion
        shell: pwsh
        run: "cargo test -p rustfs-utils --no-default-features --features path --lib dunce:: -- --nocapture"
@@ -299,7 +299,7 @@ dependencies = [
 "strum",
 "thiserror 2.0.21",
 "uuid",
- "zstd 0.13.3",
+ "zstd 0.13.4",
 ]

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

 [[package]]
@@ -976,9 +976,9 @@ dependencies = [

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

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

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

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

 [[package]]
 name = "aws-sdk-sts"
-version = "1.123.0"
+version = "1.124.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8754d38d29448bd1d4aa4a6312ed010dfa296b3436eb2c551cc6e42fc6a66a38"
+checksum = "3f8a0dd8fbde04ba58ad722113c7addb11bf94cf549b1ea0eb9ea427fb87198a"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -3960,6 +3960,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9ed9a281f7bc9b7576e61468ba615a66a5c8cfdff42420a70aa82701a3b1e292"
 dependencies = [
 "block-buffer 0.10.4",
+ "const-oid 0.9.6",
 "crypto-common 0.1.7",
 "subtle",
 ]
@@ -5295,9 +5296,9 @@ dependencies = [

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

@@ -427,7 +427,3 @@ strip = "none"
 # Pin upstream fixes until compatible crates.io releases are available.
 [patch.crates-io]
 zstd = { path = "crates/zstd-compat" }
-# Preserve RustFS OIDC flows while migrating the old elliptic-curve dependency chain.
-# Upstream: openidconnect-rs#244 and oauth2-rs#352.
-openidconnect = { git = "https://github.com/ramosbugs/openidconnect-rs", rev = "2f77ee73bc4554e1cb6279adb386005429fb4e41" }
-oauth2 = { git = "https://github.com/ramosbugs/oauth2-rs", rev = "89585f4a280afd7b285eaf1c9be0068874e03d22" }
@@ -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");
@@ -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);
@@ -148,58 +148,47 @@ pub async fn make_dir_all(path: impl AsRef<Path>) -> io::Result<()> {
    native_io(move || create_dir_all_std(&path)).await
 }

-fn is_dir_error(path: &Path, e: &io::Error) -> bool {
+fn is_dir_error(e: &io::Error) -> bool {
    e.raw_os_error() == Some(libc::EISDIR)
        || e.kind() == io::ErrorKind::IsADirectory
        // macOS: remove_file on a directory returns EPERM
        || (cfg!(target_os = "macos") && e.raw_os_error() == Some(libc::EPERM))
-        || is_windows_dir_access_denied(path, e)
-}
-
-/// Windows reports `remove_file` on a directory (or directory symlink) as
-/// ERROR_ACCESS_DENIED, which is indistinguishable from a real permission
-/// failure without inspecting the entry itself.
-#[cfg(windows)]
-fn is_windows_dir_access_denied(path: &Path, e: &io::Error) -> bool {
-    use std::os::windows::fs::FileTypeExt;
-
-    e.kind() == io::ErrorKind::PermissionDenied
-        && std::fs::symlink_metadata(path).is_ok_and(|metadata| {
-            let file_type = metadata.file_type();
-            file_type.is_dir() || file_type.is_symlink_dir()
-        })
-}
-
-#[cfg(not(windows))]
-fn is_windows_dir_access_denied(_path: &Path, _e: &io::Error) -> bool {
-    false
 }

 #[tracing::instrument(level = "debug", skip_all)]
 pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
-    let path = path.as_ref().to_path_buf();
-    native_io(move || remove_std(path)).await
+    // Try remove_file first; fall back to remove_dir if it's a directory
+    let file_path = path.as_ref().to_path_buf();
+    match native_io(move || std::fs::remove_file(file_path)).await {
+        Ok(()) => Ok(()),
+        Err(e) if is_dir_error(&e) => {
+            let path = path.as_ref().to_path_buf();
+            native_io(move || std::fs::remove_dir(path)).await
+        }
+        Err(e) => Err(e),
+    }
 }

 #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
    // Try remove_file first; fall back to remove_dir_all if it's a directory
-    let path = path.as_ref().to_path_buf();
-    native_io(move || match std::fs::remove_file(&path) {
+    let file_path = path.as_ref().to_path_buf();
+    match native_io(move || std::fs::remove_file(file_path)).await {
        Ok(()) => Ok(()),
-        Err(e) if is_dir_error(&path, &e) => std::fs::remove_dir_all(&path),
+        Err(e) if is_dir_error(&e) => {
+            let path = path.as_ref().to_path_buf();
+            native_io(move || std::fs::remove_dir_all(path)).await
+        }
        Err(e) => Err(e),
-    })
-    .await
+    }
 }

 #[tracing::instrument(level = "debug", skip_all)]
 pub fn remove_std(path: impl AsRef<Path>) -> io::Result<()> {
    // Try remove_file first; fall back to remove_dir if it's a directory
-    let path = path.as_ref();
-    match std::fs::remove_file(path) {
+    match std::fs::remove_file(path.as_ref()) {
        Ok(()) => Ok(()),
-        Err(e) if is_dir_error(path, &e) => std::fs::remove_dir(path),
+        Err(e) if is_dir_error(&e) => std::fs::remove_dir(path.as_ref()),
        Err(e) => Err(e),
    }
 }
@@ -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
-    /// guarantees a later `get` never returns one of them, so the invalidation is
-    /// effective immediately even though the removal itself is deferred. Reserved
-    /// 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.
+    /// Mark matching entries immediately so reads reject them while maintenance
+    /// removes them. The scan borrows its paths and retains no invalidation
+    /// closures when the cache is empty or idle.
    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('/').to_owned();
-        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('/')))
-        };
-        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();
-        }
+        let prefix = prefix.trim_end_matches('/');
+        self.invalidate_matching(|k| {
+            k.volume == volume && (k.path == prefix || k.path.strip_prefix(prefix).is_some_and(|r| r.starts_with('/')))
+        });
    }

    /// 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();
-        let matches = move |k: &FdKey, _: &Arc<FdCacheEntry>| k.volume == volume;
-        if self.cache.invalidate_entries_if(matches).is_err() {
-            self.cache.invalidate_all();
-        }
+        self.invalidate_matching(|k| k.volume == volume);
    }

    /// 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
-    /// on — a `get` after `invalidate_under` never returns a matched entry.
-    #[cfg(target_os = "linux")]
+    /// A `get` after `invalidate_under` must reject matched entries while
+    /// background maintenance removes them.
+    #[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");
    }

@@ -1594,13 +1594,12 @@ impl LocalDisk {
                );
            }
            if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() {
-                #[cfg(any(target_os = "linux", windows))]
+                #[cfg(target_os = "linux")]
                let payload_guard = if multipart_publication.is_some() {
                    Some(
                        os::multipart_payload_commit_guard(
                            &commit.directory_guard,
                            src_data_path.parent().ok_or(DiskError::InvalidPath)?,
-                            commit.publication_root,
                            commit.mutation_lease.clone(),
                        )
                        .await
@@ -1609,9 +1608,9 @@ impl LocalDisk {
                } else {
                    None
                };
-                #[cfg(any(target_os = "linux", windows))]
+                #[cfg(target_os = "linux")]
                let directory_guard = payload_guard.as_ref().unwrap_or(&commit.directory_guard);
-                #[cfg(not(any(target_os = "linux", windows)))]
+                #[cfg(not(target_os = "linux"))]
                let directory_guard = &commit.directory_guard;
                if let Err(err) = os::rename_all_with_commit_guard(
                    src_data_path,
@@ -5479,47 +5479,18 @@ pub(crate) fn multipart_payload_commit_guard_std(
    })
 }

-#[cfg(windows)]
-fn multipart_payload_commit_guard_std(
-    metadata_guard: &RenameCommitGuard,
-    source_parent: &Path,
-    publication_root: &PublicationRoot,
-) -> io::Result<RenameCommitGuard> {
-    // Pin the payload staging tree like a distinct-tree metadata source; the
-    // target stays the exact parent handle the metadata transaction holds.
-    let source_parent_guard = lock_windows_directory_tree(source_parent, Some(source_parent), publication_root)?;
-    Ok(RenameCommitGuard {
-        source_parent: source_parent.to_path_buf(),
-        destination_parent: metadata_guard.destination_parent.clone(),
-        source_parent_guard,
-        destination_parent_guard: metadata_guard.destination_parent_guard.clone(),
-    })
-}
-
 /// Native MPU metadata and payload have distinct source parents. Reuse the
 /// exact target identity already held by the metadata transaction, while
 /// independently pinning the private payload source under the same root.
-#[cfg(any(target_os = "linux", windows))]
+#[cfg(target_os = "linux")]
 pub(crate) async fn multipart_payload_commit_guard(
    metadata_guard: &RenameCommitGuard,
    source_parent: &Path,
-    publication_root: &PublicationRoot,
    lease: Arc<NamespaceMutationLease>,
 ) -> io::Result<RenameCommitGuard> {
    let metadata_guard = metadata_guard.clone();
    let source_parent = source_parent.to_path_buf();
-    #[cfg(target_os = "linux")]
-    let operation = {
-        // The Linux guard already carries its publication root.
-        let _ = publication_root;
-        move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent)
-    };
-    #[cfg(windows)]
-    let operation = {
-        let publication_root = publication_root.clone();
-        move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent, &publication_root)
-    };
-    run_blocking_namespace_operation(lease, operation).await
+    run_blocking_namespace_operation(lease, move || multipart_payload_commit_guard_std(&metadata_guard, &source_parent)).await
 }

 pub(crate) fn prepare_rename_commit_guard(
@@ -6630,31 +6601,12 @@ pub(crate) fn rename_multipart_source_owned_std(
    {
        let _ = &guard;
        let source_guard = pin_multipart_source_directory(source_parent, base, root)?;
-        // The prepared Windows source holds DELETE access and shares only
-        // read, so the precondition's own opens of that entry (or, for a
-        // directory, of its guarded tree) would fail with a sharing violation.
-        // Check it first under a non-exclusive identity anchor, then require
-        // the exclusive handle to name the same file.
-        #[cfg(windows)]
-        let (_source_identity_anchor, expected_source_identity) = {
-            let anchor = open_windows_rename_source_identity(source, &source_guard)?;
-            precondition()?;
-            anchor
-        };
        let preparation = prepare_rename_with_retry(source, destination, base, root)?;
-        #[cfg(windows)]
-        if windows_file_identity(&preparation.source)? != expected_source_identity {
-            return Err(io::Error::new(
-                io::ErrorKind::InvalidData,
-                "rename source identity changed after its precondition",
-            ));
-        }
        #[cfg(all(any(test, feature = "test-util"), not(windows)))]
        {
            prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, source);
            prepared_publication_test_hooks::run(prepared_publication_test_hooks::Stage::Rename, destination);
        }
-        #[cfg(not(windows))]
        precondition()?;
        #[cfg(unix)]
        validate_prepared_rename_source(&pinned_source, source)?;
@@ -7241,7 +7193,7 @@ mod multipart_source_preservation_tests {
        let metadata_guard =
            prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
        let lease = acquire_namespace_mutation_lease(&target).await;
-        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
+        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, lease.clone())
            .await
            .expect("distinct pinned payload source");
        assert!(
@@ -7276,61 +7228,6 @@ mod multipart_source_preservation_tests {
        );
    }

-    #[cfg(windows)]
-    #[tokio::test]
-    async fn windows_multipart_payload_guard_publishes_distinct_source_into_metadata_target() {
-        // Build paths from the configured root: the resolved handle path may
-        // be verbatim, where `/` in a joined key is not a separator.
-        let (root, publication, _source, _destination) = fixture();
-        let root = root.path();
-        let canonical = root.join("multipart/upload");
-        let temporary = root.join("tmp/attempt");
-        let bucket = root.join("bucket");
-        let target = bucket.join("object");
-        for directory in [&canonical, &temporary, &target] {
-            std::fs::create_dir_all(directory).expect("two-source guard directories");
-        }
-        let payload = temporary.join("payload");
-        std::fs::create_dir(&payload).expect("private payload DD");
-        std::fs::write(payload.join("part.1"), b"native selected shard").expect("private payload shard");
-        let head = canonical.join("xl.meta");
-        let destination = target.join("xl.meta");
-        let metadata_guard =
-            prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
-        let lease = acquire_namespace_mutation_lease(&target).await;
-        assert!(
-            rename_all_with_commit_guard(&payload, target.join("payload"), &bucket, &publication, &metadata_guard, lease.clone())
-                .await
-                .is_err(),
-            "the metadata guard does not pin the private payload parent"
-        );
-        assert!(
-            payload.join("part.1").exists(),
-            "a rejected payload rename must leave its source in place"
-        );
-        let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
-            .await
-            .expect("distinct pinned payload source");
-        let mut prepared = create_prepared_rename_source_with_commit_guard(&head, &destination, &metadata_guard)
-            .expect("canonical prepared metadata");
-        prepared
-            .write_all(b"prepared native metadata", true)
-            .expect("durable canonical metadata");
-        rename_all_with_commit_guard(&payload, target.join("payload"), &bucket, &publication, &payload_guard, lease.clone())
-            .await
-            .expect("payload rename uses its own pinned TMP parent");
-        assert!(head.exists(), "payload publication must not consume the canonical head");
-        rename_all_with_prepared_source(prepared, &head, &destination, &bucket, &publication, &metadata_guard, lease)
-            .await
-            .expect("metadata rename keeps its canonical parent");
-        assert!(!head.exists());
-        assert_eq!(std::fs::read(&destination).expect("published metadata"), b"prepared native metadata");
-        assert_eq!(
-            std::fs::read(target.join("payload/part.1")).expect("published selected shard"),
-            b"native selected shard"
-        );
-    }
-
    #[cfg(target_os = "linux")]
    #[tokio::test]
    async fn multipart_payload_guard_rejects_replaced_private_source_or_shared_destination() {
@@ -7352,7 +7249,7 @@ mod multipart_source_preservation_tests {
            let metadata_guard =
                prepare_rename_commit_guard(&canonical, &target, &bucket, &publication).expect("canonical metadata guard");
            let lease = acquire_namespace_mutation_lease(&target).await;
-            let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, &publication, lease.clone())
+            let payload_guard = multipart_payload_commit_guard(&metadata_guard, &temporary, lease.clone())
                .await
                .expect("pin payload and shared target");
            let replaced = if replace_source { &temporary } else { &target };
@@ -318,6 +318,21 @@ fn is_rebalance_transient_lock_error(err: &rustfs_lock::LockError) -> bool {
    }
 }

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

+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_recovers_after_write_quorum_failure() {
+        let mut attempts = 0;
+        retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(if attempts == 1 {
+                resolve_rebalance_meta_save_result(
+                    Err(rebalance_meta_lock_error(
+                        rustfs_lock::LockError::QuorumNotReached {
+                            required: 1,
+                            achieved: 0,
+                        },
+                        "write",
+                    )),
+                    "stop_rebalance",
+                )
+            } else {
+                Ok(())
+            })
+        })
+        .await
+        .expect("a later quorum-protected admission must be allowed to succeed");
+        assert_eq!(attempts, 2);
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_retains_persistent_write_quorum_failure() {
+        for max_attempts in [0, 1, 3] {
+            let mut attempts = 0;
+            let result = retry_rebalance_metadata_access(None, max_attempts, || {
+                attempts += 1;
+                std::future::ready(Err::<(), _>(rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )))
+            })
+            .await;
+            let error = result.expect_err("exhausted admission attempts must never acknowledge a write");
+            assert_eq!(attempts, max_attempts.max(1));
+            assert!(matches!(
+                rebalance_error_source(&error),
+                Error::NamespaceLockQuorumUnavailable {
+                    mode: "write",
+                    required: 1,
+                    achieved: 0,
+                    ..
+                }
+            ));
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn rebalance_metadata_retry_stops_after_admission_is_cancelled() {
+        let mut attempts = 0;
+        let result = retry_rebalance_metadata_access(None, 3, || {
+            attempts += 1;
+            std::future::ready(Err::<(), _>(if attempts == 1 {
+                rebalance_meta_lock_error(
+                    rustfs_lock::LockError::QuorumNotReached {
+                        required: 1,
+                        achieved: 0,
+                    },
+                    "write",
+                )
+            } else {
+                Error::OperationCanceled
+            }))
+        })
+        .await;
+        assert_eq!(attempts, 2, "a stale or cancelled run cannot enter another admission attempt");
+        assert!(matches!(result, Err(Error::OperationCanceled)));
+    }
+
    #[tokio::test]
    async fn rebalance_metadata_retry_does_not_retry_permanent_or_untyped_errors() {
        for err in [
@@ -731,6 +820,20 @@ mod error_source_tests {
                required: 3,
                achieved: 2,
            },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: "data-bucket".to_string(),
+                object: REBAL_META_NAME.to_string(),
+                required: 1,
+                achieved: 0,
+            },
+            Error::NamespaceLockQuorumUnavailable {
+                mode: "write",
+                bucket: crate::disk::RUSTFS_META_BUCKET.to_string(),
+                object: "other-metadata.bin".to_string(),
+                required: 1,
+                achieved: 0,
+            },
            Error::other("stale rebalance run rejected: lock acquisition timed out"),
            Error::other("rebalance distributed run fence lost"),
            Error::Io(std::io::Error::from(std::io::ErrorKind::TimedOut)),
@@ -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();
@@ -16,9 +16,10 @@ use super::{
    transport::{DiscoveryRequest, OidcHttpError, ReqwestHttpClient, oidc_extra_root_certs, oidc_http_error_diagnostics},
 };
 use crate::{EVENT_OIDC_DIAGNOSTICS, LOG_COMPONENT_IAM, LOG_SUBSYSTEM_OIDC};
-use openidconnect::core::{CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
+use openidconnect::core::{CoreIdToken, CoreIdTokenClaims, CoreIdTokenVerifier, CoreJsonWebKeySet, CoreJwsSigningAlgorithm};
 use openidconnect::{
-    AsyncHttpClient, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, ProviderMetadataWithLogout, TokenUrl,
+    AsyncHttpClient, ClaimsVerificationError, ClientId, ClientSecret, DiscoveryError, IssuerUrl, JsonWebKeySetUrl, NonceVerifier,
+    ProviderMetadataWithLogout, SignatureVerificationError, TokenUrl,
 };
 use serde::Deserialize;
 use std::collections::HashMap;
@@ -212,6 +213,35 @@ impl DiscoveredProviderMetadata {
    }
 }

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

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

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

@@ -1,6 +1,6 @@
 [package]
 name = "zstd"
-version = "0.13.3"
+version = "0.13.4"
 publish = false
 edition.workspace = true
 license.workspace = true
@@ -6,6 +6,6 @@ This private package satisfies the version constraints of Avro, Arrow IPC, Parqu

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

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

 Remove this package and the root Cargo patch when all consumers accept the fixed zstd release line directly. The cleanup marker is registered in the architecture compatibility register.
@@ -57,7 +57,7 @@ fn legacy_constraints_resolve_only_to_the_fixed_registry_implementation() {
        let version = Version::parse(package["version"].as_str().expect("package version")).expect("valid package semver");
        if name == "zstd" && package.get("source").is_none() {
            adapters += 1;
-            assert_eq!(version, Version::new(0, 13, 3));
+            assert_eq!(version, Version::new(0, 13, 4));
            let runtime = package["dependencies"].as_array().expect("adapter dependency edges");
            assert!(runtime.iter().any(|value| value.as_str() == Some("zstd 0.14.3")));
        } else {
@@ -36,10 +36,6 @@ allow-git = [
    # Official s3s repository pinned for Iceberg SigV4 path encoding support.
    # owner: GatewayJ  review: 2026-10
    "https://github.com/s3s-project/s3s",
-    # Upstream OIDC/OAuth2 crypto migrations, pinned to reviewed PR commits.
-    # owner: rustfs-maintainers  review: 2026-10
-    "https://github.com/ramosbugs/openidconnect-rs",
-    "https://github.com/ramosbugs/oauth2-rs",
    # Completion ownership patch fixed to rustfs/uring#20.
    # owner: overtrue  review: 2026-10
    "https://github.com/rustfs/uring.git",
@@ -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.
@@ -273,9 +273,9 @@ dependencies = [

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

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

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

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

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

 [[package]]
 name = "lru"
@@ -4134,8 +4134,7 @@ dependencies = [
 [[package]]
 name = "rustfs-uring"
 version = "0.3.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f7deda770b7981ee3bfaa202ecd18625fc794c962b8d58cd4f436952b7e66dcf"
+source = "git+https://github.com/rustfs/uring.git?rev=eb8fc160e71d804427de1e905f5c226d580f71bf#eb8fc160e71d804427de1e905f5c226d580f71bf"
 dependencies = [
 "io-uring",
 "libc",
@@ -540,7 +540,7 @@ struct CleanupObservation {
    elapsed: Duration,
 }

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

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

 #[tokio::test]
@@ -750,7 +766,7 @@ async fn cleanup_diagnostics_identify_timeouts_without_resetting_the_shared_dead
            "bucket_delete",
        ),
    ] {
-        let observed = observe_cleanup(replies).await;
+        let observed = observe_cleanup(replies, Duration::ZERO).await;
        assert_cleanup_failure(&observed, stage, "TimedOut", None);
        assert!(observed.elapsed >= Duration::from_secs(5));
        assert!(observed.log.contains("remaining_ms=0"), "{}", observed.log);
Author	SHA1	Message	Date
hector	d52d720d17	Merge branch 'main' into ci/validation-sweep-boto3 Keep main's validation sweep workflow where both sides edited the same steps. The Python 3.12 venv install from validation_ilm/requirements.txt is already on main and remains in the sweep job, with the scheduled nightly pin, chain envelope read, ssh -n, AUTO_TESTING_SHA, and artifact warn/run path. Co-authored-by: hector <majinghe@users.noreply.github.com>	2026-11-02 12:39:52 +00:00
GatewayJ	45ba33291a	fix(ecstore): release retained file descriptor cache invalidations (#9112 ) * fix(ecstore): release retained descriptor cache invalidations * test(ecstore): await descriptor cleanup after invalidation	2026-11-02 10:56:08 +00:00
Hauser Zhi	4572bccea1	Merge branch 'main' into ci/validation-sweep-boto3	2026-11-02 18:07:28 +08:00
hector	ccc25bce07	ci(validation): install the validation Python client requirements The pool runner dies with 'boto3 is required for pool validation' before its assertions: the managed interpreter alone is not enough. Mirror the chain validation job exactly - build a venv from validation_ilm/requirements.txt and put its bin on PATH so every module runner (pool/ilm/replication) gets its client stack.	2026-11-02 16:38:53 +08:00
hector	636eded277	ci(validation): resolve the newest nightly build on scheduled sweeps (#9114 ) * ci(validation): resolve the newest nightly build on scheduled sweeps The first scheduled sweep failed at candidate resolution: a schedule carries no workflow inputs and resolve_functional_candidate.py only derives a build from a workflow_run payload, so the empty BUILD_RUN_ID crashed int(). Look up the newest successful nightly-gnu run explicitly when the inputs are empty and feed it to the resolver. Also downgrade the evidence artifact to if-no-files-found: warn so an early infra failure does not cascade into a second red step. * ci(validation): read the resolver chain envelope correctly CHAIN_OUTPUT is the chain envelope {schema, run_id, attempt, workflow_sha, testing_sha, candidate}; the manifest lives under candidate.manifest and the testing sha at the top level. * ci(validation): sweep harness fixes from the first live run - ssh -n in the inter-case cleanup loop: plain ssh consumed the case list from the loop's stdin, so the first dispatch executed only one of the twelve selected cases. - Provide the managed Python 3.12 runtime for pool/ilm/replication cases, mirroring the chain validation job (pool.status reported INFRA_ERROR after its assertions passed; the missing runtime is the prime suspect). - Record the auto-testing sha in the report step environment. - Retain the last case's evidence tree in the sweep artifact so gate verdicts like INFRA_ERROR can be diagnosed from the run itself. * ci(validation): install sweep Python client dependencies Scheduled sweeps failed pool.status before any assertion because pool_behavior.py exits when boto3 is missing. Install the auto-testing validation requirements into a per-run virtualenv and put it on PATH, matching the functional chain, so case scripts can import the client. Co-authored-by: hector <majinghe@users.noreply.github.com> --------- Co-authored-by: hector <majinghe@users.noreply.github.com>	2026-11-02 08:34:13 +00:00
hector	6199e30ffc	Merge branch 'main' into ci/validation-sweep-schedule-fix	2026-11-02 15:01:41 +08:00
hector	835b84c57a	ci(validation): install sweep Python client dependencies Scheduled sweeps failed pool.status before any assertion because pool_behavior.py exits when boto3 is missing. Install the auto-testing validation requirements into a per-run virtualenv and put it on PATH, matching the functional chain, so case scripts can import the client. Co-authored-by: hector <majinghe@users.noreply.github.com>	2026-11-02 06:56:18 +00:00
Hauser Zhi	ed4099ebe1	chore(deps): update crates and fix storage CI regressions (#9111 ) * chore(deps): refresh crates and adopt openidconnect 4.1 Use published OIDC/OAuth2 releases and reject malformed HMAC signature lengths before upstream verification. Refresh workspace and fuzz locks, and advance the local zstd compatibility patch to 0.13.4 while retaining the fixed 0.14.3 codec backend. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * chore(deps): refresh compatible lockfile resolutions Resolve Prost build and derive dependencies with itertools 0.14 and prefer windows-sys 0.61.2 within the existing supported version ranges. Retain all package versions, sources and checksums. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(connect): separate cleanup checks from short probe windows Use the functional measurement budget for the cleanup HTTP fixture so setup and measured I/O can finish before checking delete status handling. Exercise both normal scheduling and an in-flight PUT clock advance, and include target classification in unexpected-failure diagnostics. Retain the cleanup deadline, outer timeout and delete-stage assertions. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * test(ecstore): isolate stalled multipart disk timeouts Apply the short injected timeout only to the deliberately stalled disk. Healthy disks retain their normal operation budgets, while logical lock, physical owner, immediate CopyObject and final shard assertions remain. Keep overrides scoped across cloned and reconnected handles, preserve shorter caller deadlines and explicit zero timeouts, and report drive health when the real CopyObject assertion fails. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(rebalance): retry metadata write lock quorum admission Recognize typed write-lock quorum failures for rebalance.bin within the existing bounded metadata admission retry policy. Each attempt must acquire quorum before loading or saving metadata; data writes are not replayed by this retry. Cover recovered and persistent quorum failures, cancellation after a failed attempt, and rejection of unrelated resource or read-lock errors. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-11-02 06:49:24 +00:00
hector	14a2fd9569	ci(validation): sweep harness fixes from the first live run - ssh -n in the inter-case cleanup loop: plain ssh consumed the case list from the loop's stdin, so the first dispatch executed only one of the twelve selected cases. - Provide the managed Python 3.12 runtime for pool/ilm/replication cases, mirroring the chain validation job (pool.status reported INFRA_ERROR after its assertions passed; the missing runtime is the prime suspect). - Record the auto-testing sha in the report step environment. - Retain the last case's evidence tree in the sweep artifact so gate verdicts like INFRA_ERROR can be diagnosed from the run itself.	2026-11-02 09:27:20 +08:00
hector	bf201caf34	ci(validation): read the resolver chain envelope correctly CHAIN_OUTPUT is the chain envelope {schema, run_id, attempt, workflow_sha, testing_sha, candidate}; the manifest lives under candidate.manifest and the testing sha at the top level.	2026-11-02 09:17:09 +08:00
hector	4d81330479	ci(validation): resolve the newest nightly build on scheduled sweeps The first scheduled sweep failed at candidate resolution: a schedule carries no workflow inputs and resolve_functional_candidate.py only derives a build from a workflow_run payload, so the empty BUILD_RUN_ID crashed int(). Look up the newest successful nightly-gnu run explicitly when the inputs are empty and feed it to the resolver. Also downgrade the evidence artifact to if-no-files-found: warn so an early infra failure does not cascade into a second red step.	2026-11-02 09:09:20 +08:00