perf(storage): gate large foreground PUT pressure (#6751)

Add a default-on, size-aware foreground PUT admission policy so large or
unknown-size PutObject requests are backpressured before body ingest and
erasure/RPC fan-out. Preserve the explicit strict gate semantics, including
limit=0 as an opt-out, and keep small PUTs on the legacy fast path.

Closes rustfs/backlog#2038

Co-authored-by: heihutu <heihutu@gmail.com>

.agents/skills/rustfs-release-publish/SKILL.md

@@ -6,7 +6,7 @@ description: "Run the end-to-end RustFS console gate, version bump, preview vali
 
 This skill orchestrates a full release. It wraps `rustfs-release-version-bump` (which only edits version files and opens the PR) with a mandatory preview-tag validation loop before the final tag is published.
 
-Core design: **version files never carry a `-preview.N` suffix**. The preview suffix exists only in tag names. A preview tag creates a visible GitHub Release marked Prerelease and uploads versioned assets, but it never becomes GitHub Latest and never updates `*-latest`, `latest.json`, R2, Docker, or Helm channels. This works because the binary self-reports the git tag it was built from (`build::TAG` via shadow_rs, see `rustfs/src/config/cli.rs` `SHORT_VERSION`), and `build.yml` derives artifact names and preview classification from the tag name — Cargo.toml's version is only a no-tag fallback. Therefore the preview tag and the final tag can (and MUST) point at the exact same commit: what you validated is byte-for-byte the source that ships.
+Core design: **version files never carry a `-preview.N` suffix**. The preview suffix exists only in tag names. A preview tag creates a visible GitHub Release marked Prerelease and uploads versioned assets, but it never becomes GitHub Latest and never updates `*-latest`, `latest.json`, R2, Docker, or Helm channels. That Release is temporary: `build.yml` deletes it automatically once the final tag's Release is published, so the Releases page ends up carrying deliverables only while the `-preview.N` tags stay behind as the traceability record. This works because the binary self-reports the git tag it was built from (`build::TAG` via shadow_rs, see `rustfs/src/config/cli.rs` `SHORT_VERSION`), and `build.yml` derives artifact names and preview classification from the tag name — Cargo.toml's version is only a no-tag fallback. Therefore the preview tag and the final tag can (and MUST) point at the exact same commit: what you validated is byte-for-byte the source that ships.
 
 Pipeline shape:
 
@@ -19,6 +19,7 @@ check console main against its latest Release
   -> validate with latest rc client
   -> report preview acceptance results -> STOP for explicit human confirmation
   -> tag <target> at the SAME commit (zero delta) -> re-verify CI/release
+  -> CI deletes the <target>-preview.N Releases (tags kept)
 ```
 
 On validation failure: fix lands on main via normal PR (version files are already at `<target>`, no new bump PR), then tag `<preview-tag N+1>` at the new main commit and restart from Phase 2.
@@ -51,14 +52,16 @@ Rules:
 - Use `<target>-preview.N` for every target, e.g. `1.0.0-beta.10-preview.3` or `1.1.0-preview.1`.
 - The canonical suffix is exactly `-preview.<digits>`. `build.yml` recognizes it before alpha/beta/rc classification and routes it to the preview-only path; any other tag containing `-preview` fails closed instead of being treated as a release.
 - A preview Release MUST be published with `isPrerelease=true` and `isLatest=false`. Any `*-latest` preview asset or preview-triggered `latest.json`, R2, Docker, or Helm publication is a pipeline failure.
+- Preview Releases are cleaned up by the `cleanup-preview-releases` job after `publish-release` succeeds for the deliverable tag. It deletes every Release whose tag is exactly `<target>-preview.<digits>` and never passes `--cleanup-tag`, so the tags survive.
 
 ## Hard rules
 
 - Version files (Cargo.toml, Cargo.lock, README, flake.nix, Chart.yaml, rustfs.spec) are bumped ONCE, directly to `<target>`. Never write a `-preview.N` suffix into any version file. If `rustfs-release-version-bump` is ever asked for a `-preview` version, that is a pipeline bug — stop.
-- Preview Release assets are versioned and intentionally visible on the Releases page. Do not label them Latest or use them to update any latest distribution channel.
+- Preview Release assets are versioned and intentionally visible on the Releases page for the duration of validation. Do not label them Latest or use them to update any latest distribution channel.
+- Never delete a preview Release by hand before Phase 6 finishes — Phase 4 downloads its assets and the final Release notes are generated while it still exists. Cleanup is CI's job; only step in manually (`gh release delete "<preview-tag>" --yes`, never `--cleanup-tag`) if `cleanup-preview-releases` failed.
 - Tags have no `v` prefix. Always annotated: `git tag -a <tag> -m "Release <tag>"`.
 - The final tag MUST point at exactly `PREVIEW_HASH` — the commit the validated preview tag points at. Never tag current `main` HEAD (commits merged after validation are unvalidated), and never create an extra version-bump commit between preview and final.
-- When a previous deliverable exists, GitHub Release notes for the preview and final tags MUST use it as their shared comparison baseline: the most recently published non-preview Release before the target. Internal `-preview.N` Releases are explicitly excluded from that selection, even when they point at the same commit as the final tag. If no previous deliverable exists, omit `previous_tag_name` and record that GitHub's default baseline fallback was used.
+- When a previous deliverable exists, GitHub Release notes for the preview and final tags MUST use it as their shared comparison baseline: the most recently published non-preview Release before the target. Internal `-preview.N` Releases are explicitly excluded from that selection, even when they point at the same commit as the final tag — cleanup runs after the notes are generated, so the preview Release is still present and would otherwise be picked as the baseline. If no previous deliverable exists, omit `previous_tag_name` and record that GitHub's default baseline fallback was used.
 - Generated Release notes carry a workflow-management marker so retries can repair them. Before manually curating a generated body, remove that marker; unmarked non-placeholder notes are preserved by later workflow runs.
 - Phases run in order; a failure in any phase blocks everything after it. After the fix lands on main, restart from Phase 2 with the next preview iteration against the new `origin/main` hash — do not resume mid-pipeline against a stale hash.
 - Completing preview acceptance does not authorize the final tag. After Phases 3–5 pass, report the acceptance evidence and stop until the user explicitly confirms continuation. The original release request, an earlier confirmation, silence, or an automated follow-up does not satisfy this gate.
@@ -230,6 +233,7 @@ git push origin "<target>"
 - CI rebuilds from the same source; the only changed input is the tag name, so the binary now self-reports `<target>`.
 - Verify the final tag's complete publication path: all matrix and release jobs green; `gh release view "<target>"` shows the full versioned and `-latest` asset set plus checksums, SBOM, and provenance; Docker and Helm workflows succeed; `latest.json` points to `<target>`. A stable target must have `isPrerelease=false` and `isLatest=true`. An alpha/beta/rc target must have `isPrerelease=true`; GitHub does not permit prereleases to be Latest, but the project `latest.json` still advances to the final non-preview target.
 - Verify the final Release body contains `## What's Changed` and a Full Changelog link. When `PREVIOUS_DELIVERABLE` exists, the link MUST be `https://github.com/rustfs/rustfs/compare/<PREVIOUS_DELIVERABLE>...<target>` and the baseline MUST equal the preview Release baseline; for example, both `1.0.0-beta.12-preview.1` and `1.0.0-beta.12` compare from `1.0.0-beta.11`.
+- Verify the preview cleanup: `cleanup-preview-releases` must succeed, `gh release view "<preview-tag>"` must then report `release not found` for every preview iteration of this target, and `git rev-parse "<preview-tag>^{commit}"` must still resolve to `PREVIEW_HASH` (the tag is kept). If the job failed, delete the leftover Releases manually with `gh release delete "<preview-tag>" --yes` and report it.
 - Optionally spot-check `./rustfs --version` from a final-tag artifact — it must report `<target>`.
 
 ## Output contract
@@ -239,5 +243,5 @@ Always report:
 - Console gate result: previous/latest Console tags, whether merged changes required a release, `CONSOLE_HASH`, and Console run/Release URLs when a release was published.
 - Target version, preview tag(s) used, `PREVIEW_HASH` (which both tags point at).
 - Manual confirmation gate status (`WAITING_FOR_CONFIRMATION` or `CONFIRMED`) and its exact target, preview tag, and `PREVIEW_HASH`.
-- Per-phase result (PASS/FAIL/BLOCKED) with key evidence: preview and final Release URLs, preview `isPrerelease`/`isLatest` state, final latest-channel state, console check results, and the rc command matrix.
+- Per-phase result (PASS/FAIL/BLOCKED) with key evidence: preview and final Release URLs, preview `isPrerelease`/`isLatest` state, final latest-channel state, console check results, the rc command matrix, and the preview-Release cleanup result (deleted Releases plus surviving tags).
 - Any deviation from this pipeline and why the user approved it.

.config/nextest.toml

@@ -205,16 +205,6 @@ retries = 2
 filter = 'package(rustfs) & test(execute_get_object_resumes_from_relocated_pool_without_splicing_body)'
 test-group = 'ecstore-serial-flaky'
 
-# QUARANTINE: OPEN rustfs#6711 — the multipart fencing test's epoch helper
-# reads xl.meta back from EVERY disk, but a multipart commit only guarantees
-# quorum-many disks have persisted; a lagging disk under CI load panics the
-# read-back with "file not found" (observed on the rio-v2 leg of a
-# nextest-config-only PR; same all-disk-materialization assumption as the
-# relocated-pool fixture fixed by #6707).
-[[profile.ci.overrides]]
-filter = 'package(rustfs-ecstore) & test(object_transaction_fencing_persists_epoch_on_multipart_commit)'
-retries = 2
-
 # Serialize the 4-disk reliability / degraded-read e2e tests under the ci
 # profile too (see the e2e-reliability test-group note near the top). Not a
 # quarantine: no retries, just single-threaded so several 4-disk servers never

.github/actionlint.yaml

@@ -5,3 +5,4 @@ self-hosted-runner:
     - sm-standard-2
     - sm-standard-4
     - dind-sm-standard-2
+    - smoke-testing

.github/workflows/build.yml

@@ -1033,6 +1033,55 @@ jobs:
           echo "🎉 Released $TAG successfully!"
           echo "📄 Release URL: ${{ needs.create-release.outputs.release_url }}"
 
+  # Remove the internal preview releases once the deliverable release is live.
+  # Only the Releases are deleted; the -preview.N tags stay so the validated
+  # commit remains traceable.
+  cleanup-preview-releases:
+    name: Cleanup Preview Releases
+    needs: [ build-check, publish-release ]
+    if: startsWith(github.ref, 'refs/tags/') && (needs.build-check.outputs.build_type == 'release' || needs.build-check.outputs.build_type == 'prerelease')
+    runs-on: ubuntu-latest
+    timeout-minutes: 30
+    permissions:
+      contents: write
+    steps:
+      - name: Delete preview releases for this target
+        env:
+          GH_TOKEN: ${{ github.token }}
+        shell: bash
+        run: |
+          set -euo pipefail
+
+          TAG="${{ needs.build-check.outputs.version }}"
+          RELEASES_JSON="${RUNNER_TEMP}/releases.json"
+
+          # Fetch before filtering: a failed listing must abort here instead of
+          # looking like "nothing to clean up".
+          gh api --paginate "repos/${GITHUB_REPOSITORY}/releases?per_page=100" > "$RELEASES_JSON"
+
+          # Match only <target>-preview.<digits>. String operations, not a
+          # regex over the tag, so dots in the version cannot widen the match.
+          DELETED=0
+          while IFS= read -r preview_tag; do
+            [[ -n "$preview_tag" ]] || continue
+            echo "🧹 Deleting preview release $preview_tag (tag kept)"
+            gh release delete "$preview_tag" --yes
+            DELETED=$((DELETED + 1))
+          done < <(
+            jq -r --arg tag "$TAG" '
+              .[]
+              | select(.tag_name | startswith($tag + "-preview."))
+              | select(.tag_name | ltrimstr($tag + "-preview.") | test("^[0-9]+$"))
+              | .tag_name
+            ' "$RELEASES_JSON"
+          )
+
+          if [[ "$DELETED" -eq 0 ]]; then
+            echo "ℹ️  No preview releases to clean up for $TAG"
+          else
+            echo "✅ Removed $DELETED preview release(s) for $TAG"
+          fi
+
   alert-on-failure:
     name: Alert on scheduled failure
     needs: [build-check, prepare-platform-matrix, build-rustfs, build-summary]

.github/workflows/rustfs-heal-test.yml

@@ -0,0 +1,126 @@
+name: RustFS Heal Test
+
+on:
+  workflow_dispatch:
+    inputs:
+      package_url:
+        description: 'Direct .deb URL (nightly/R2). Defaults to the latest nightly deb.'
+        required: false
+        type: string
+      stop_node_gb:
+        description: 'Stop the outage node when surviving nodes reach N GiB'
+        required: false
+        default: '15'
+      warp_stop_gb:
+        description: 'Stop warp when surviving nodes reach N GiB'
+        required: false
+        default: '40'
+      heal_target_gb:
+        description: 'Outage node must reach N GiB after heal to pass'
+        required: false
+        default: '40'
+      cleanup_before:
+        description: 'Reset the nodes before the test (DESTROYS existing data/config)'
+        type: boolean
+        default: true
+      cleanup_after:
+        description: 'Reset the nodes after the test (DESTROYS test data/config)'
+        type: boolean
+        default: true
+
+permissions:
+  contents: read
+
+# Only one test at a time: both this and the pool-expansion workflow mutate
+# the same test environment, so they share one concurrency group.
+concurrency:
+  group: rustfs-pool-expansion-test
+  cancel-in-progress: false
+
+defaults:
+  run:
+    shell: bash
+
+env:
+  RUSTFS_ACCESS_KEY: ${{ secrets.RUSTFS_ACCESS_KEY }}
+  RUSTFS_SECRET_KEY: ${{ secrets.RUSTFS_SECRET_KEY }}
+  RUSTFS_API_ENDPOINT: ${{ secrets.RUSTFS_API_ENDPOINT || vars.RUSTFS_API_ENDPOINT || vars.RUSTFS_RC_ENDPOINT }}
+  RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES }}
+  RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
+  RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
+
+jobs:
+  heal-test:
+    runs-on: smoke-testing
+    timeout-minutes: 480
+    steps:
+      - name: Checkout
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+
+      - name: Show environment
+        run: |
+          uname -a
+          jq --version
+          openssl version
+          warp --version || true
+          df -h /data | tail -1
+
+      - name: Reset test environment (before)
+        if: ${{ inputs.cleanup_before != 'false' }}
+        run: |
+          chmod +x scripts/test/rustfs_heal_test.sh
+          ./scripts/test/rustfs_heal_test.sh --reset -y
+
+      - name: Install RustFS package & start cluster
+        run: |
+          ARGS=(--steps "1,2" -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_heal_test.sh "${ARGS[@]}"
+
+      - name: Preflight checks
+        run: |
+          ARGS=(--preflight --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_heal_test.sh "${ARGS[@]}"
+
+      - name: Run heal test (write -> outage -> heal -> verify)
+        run: |
+          ./scripts/test/rustfs_heal_test.sh \
+            --steps "3,4,5,6,7" -y \
+            --endpoint "${{ env.RUSTFS_API_ENDPOINT }}" \
+            --stop-node-gb "${{ inputs.stop_node_gb }}" \
+            --warp-stop-gb "${{ inputs.warp_stop_gb }}" \
+            --heal-target-gb "${{ inputs.heal_target_gb }}" \
+            --log-file /tmp/rustfs-heal-test.log
+
+      - name: Upload test logs
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: rustfs-heal-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-heal-test*.log
+            /tmp/rustfs-warp.*.log
+          if-no-files-found: warn
+
+      - name: Reset test environment (after)
+        if: ${{ always() && inputs.cleanup_after != 'false' }}
+        run: |
+          ./scripts/test/rustfs_heal_test.sh --reset -y
+
+      - name: Notify on failure
+        if: failure()
+        run: |
+          echo "RustFS heal test failed"
+          echo "Package source: ${{ inputs.package_url || 'nightly (R2 latest)' }}"
+          echo "See the uploaded log artifact for details."

.github/workflows/rustfs-pool-expand-test.yml

@@ -30,6 +30,18 @@ on:
         description: 'Run the pool decommission step (3-pool topology only)'
         type: boolean
         default: true
+      stop_node_gb:
+        description: 'Heal: stop the outage node when surviving nodes reach N GiB'
+        required: false
+        default: '15'
+      warp_stop_gb:
+        description: 'Heal: stop warp when surviving nodes reach N GiB'
+        required: false
+        default: '40'
+      heal_target_gb:
+        description: 'Heal: outage node must reach N GiB after heal'
+        required: false
+        default: '40'
       cleanup_before:
         description: 'Reset the nodes before the test (DESTROYS existing data/config)'
         type: boolean
@@ -38,9 +50,10 @@ on:
         description: 'Reset the nodes after the test (DESTROYS test data/config)'
         type: boolean
         default: true
-  schedule:
-    # Nightly regression run; remove if you do not want a schedule.
-    - cron: '0 21 * * *'
+  workflow_run:
+    # Run after the nightly build completes: pool expansion first, then heal.
+    workflows: ["Nightly GNU Build"]
+    types: [completed]
 
 permissions:
   contents: read
@@ -61,19 +74,23 @@ env:
   RUSTFS_API_ENDPOINT: ${{ secrets.RUSTFS_API_ENDPOINT || vars.RUSTFS_API_ENDPOINT || vars.RUSTFS_RC_ENDPOINT }}
   RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES }}
   RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
-  # Package used by the scheduled run (workflow_dispatch inputs are empty for
-  # schedule events), i.e. the latest nightly deb published by nightly-gnu.yml.
+  # Package used by the nightly run (workflow_dispatch inputs are empty for
+  # workflow_run events), i.e. the latest nightly deb published by nightly-gnu.yml.
   RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
 
 jobs:
   pool-expansion-test:
     runs-on: smoke-testing
     timeout-minutes: 360
+    # Run on manual dispatch, or when the nightly build completed successfully
+    # (its deb is what the tests install). Skipped when nightly failed.
+    if: ${{ github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success' }}
     steps:
       - name: Checkout
         uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
         with:
           persist-credentials: false
+          ref: ${{ github.event.workflow_run.head_sha || github.ref }}
 
       - name: Show environment
         run: |
@@ -91,7 +108,7 @@ jobs:
 
       - name: Install RustFS package & start first pool
         run: |
-          ARGS=(--steps 1,2,3 -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
+          ARGS=(--steps "1,2,3" -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
           if [ -n "${{ inputs.package_url }}" ]; then
             ARGS+=(--package-url "${{ inputs.package_url }}")
           elif [ -n "${{ inputs.rustfs_version }}" ]; then
@@ -137,8 +154,8 @@ jobs:
         with:
           name: rustfs-pool-test-${{ github.run_id }}
           path: |
-            /tmp/rustfs-pool-test.log
-            /tmp/rustfs-warp.log
+            /tmp/rustfs-pool-test*.log
+            /tmp/rustfs-warp.*.log
           if-no-files-found: warn
 
       - name: Reset test environment (after)
@@ -152,3 +169,75 @@ jobs:
           echo "RustFS pool expansion test failed"
           echo "Package source: ${{ inputs.package_url || inputs.rustfs_version || 'nightly (R2 latest)' }}"
           echo "See the uploaded log artifact for details."
+
+  # Heal regression runs after the pool test regardless of its outcome: a pool
+  # failure must be reported (it makes the run red) but must not block heal.
+  heal-test:
+    name: Heal test (after pool test)
+    runs-on: smoke-testing
+    timeout-minutes: 480
+    needs: pool-expansion-test
+    if: ${{ always() && (github.event_name == 'workflow_dispatch' || github.event.workflow_run.conclusion == 'success') }}
+    steps:
+      - name: Checkout
+        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
+        with:
+          persist-credentials: false
+          ref: ${{ github.event.workflow_run.head_sha || github.ref }}
+
+      - name: Reset test environment (before)
+        if: ${{ inputs.cleanup_before != 'false' }}
+        run: |
+          chmod +x scripts/test/rustfs_heal_test.sh
+          ./scripts/test/rustfs_heal_test.sh --reset -y
+
+      - name: Install RustFS package & start cluster
+        run: |
+          ARGS=(--steps "1,2" -y --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_heal_test.sh "${ARGS[@]}"
+
+      - name: Preflight checks
+        run: |
+          ARGS=(--preflight --endpoint "${{ env.RUSTFS_API_ENDPOINT }}")
+          if [ -n "${{ inputs.package_url }}" ]; then
+            ARGS+=(--package-url "${{ inputs.package_url }}")
+          else
+            ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
+          fi
+          ./scripts/test/rustfs_heal_test.sh "${ARGS[@]}"
+
+      - name: Run heal test (write -> outage -> heal -> verify)
+        run: |
+          ./scripts/test/rustfs_heal_test.sh \
+            --steps 3,4,5,6,7 -y \
+            --endpoint "${{ env.RUSTFS_API_ENDPOINT }}" \
+            --stop-node-gb "${{ inputs.stop_node_gb || '15' }}" \
+            --warp-stop-gb "${{ inputs.warp_stop_gb || '40' }}" \
+            --heal-target-gb "${{ inputs.heal_target_gb || '40' }}" \
+            --log-file /tmp/rustfs-heal-test.log
+
+      - name: Upload test logs
+        if: always()
+        uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
+        with:
+          name: rustfs-heal-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-heal-test.log
+            /tmp/rustfs-warp.*.log
+          if-no-files-found: warn
+
+      - name: Reset test environment (after)
+        if: ${{ always() && inputs.cleanup_after != 'false' }}
+        run: |
+          ./scripts/test/rustfs_heal_test.sh --reset -y
+
+      - name: Notify on failure
+        if: failure()
+        run: |
+          echo "RustFS heal test failed"
+          echo "See the uploaded log artifact for details."

Cargo.lock

@@ -1623,6 +1623,22 @@ dependencies = [
  "digest 0.11.3",
 ]
 
+[[package]]
+name = "blazesym"
+version = "0.2.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "847a0a95b041ad5aae1bdc44f2bd54743f76eb0065c4f86b139f81343c287eaf"
+dependencies = [
+ "cpp_demangle",
+ "crc32fast",
+ "flate2",
+ "gimli 0.33.0",
+ "libc",
+ "memmap2",
+ "rustc-demangle",
+ "tempfile",
+]
+
 [[package]]
 name = "block-buffer"
 version = "0.10.4"
@@ -1806,6 +1822,12 @@ dependencies = [
  "libbz2-rs-sys",
 ]
 
+[[package]]
+name = "c-enum"
+version = "0.2.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "cd17eb909a8c6a894926bfcc3400a4bb0e732f5a57d37b1f14e8b29e329bace8"
+
 [[package]]
 name = "camino"
 version = "1.2.5"
@@ -2300,6 +2322,15 @@ version = "0.8.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "773648b94d0e5d620f64f280777445740e61fe701025087ec8b57f45c791888b"
 
+[[package]]
+name = "cpp_demangle"
+version = "0.5.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "0667304c32ea56cb4cd6d2d7c0cfe9a2f8041229db8c033af7f8d69492429def"
+dependencies = [
+ "cfg-if",
+]
+
 [[package]]
 name = "cpubits"
 version = "0.1.1"
@@ -3453,25 +3484,23 @@ dependencies = [
 
 [[package]]
 name = "deadpool"
-version = "0.12.3"
+version = "0.13.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0be2b1d1d6ec8d846f05e137292d0b89133caf95ef33695424c09568bdd39b1b"
+checksum = "3e98a7e119cd347f4201e1159b19831029e203e2d8b790547708e8157b4acf1e"
 dependencies = [
  "deadpool-runtime",
- "lazy_static",
- "num_cpus",
  "tokio",
 ]
 
 [[package]]
 name = "deadpool-postgres"
-version = "0.14.1"
+version = "0.14.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3d697d376cbfa018c23eb4caab1fd1883dd9c906a8c034e8d9a3cb06a7e0bef9"
+checksum = "65a536565624b97fc19f758cd01b15d12908d3344425066efc8162236fbd3749"
 dependencies = [
  "async-trait",
  "deadpool",
- "getrandom 0.2.17",
+ "getrandom 0.4.3",
  "tokio",
  "tokio-postgres",
  "tracing",
@@ -3479,9 +3508,9 @@ dependencies = [
 
 [[package]]
 name = "deadpool-runtime"
-version = "0.1.4"
+version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "092966b41edc516079bdf31ec78a2e0588d1d0c08f78b91d8307215928642b2b"
+checksum = "2657f61fb1dd8bf37a8d51093cc7cee4e77125b22f7753f49b289f831bec2bae"
 dependencies = [
  "tokio",
 ]
@@ -3685,35 +3714,61 @@ dependencies = [
 ]
 
 [[package]]
-name = "dial9-macro"
-version = "0.3.7"
+name = "dial9-core"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a7e31f073f2e14e5a9d338c543a0601aeaf7c43fc428cd59ce417230d0db37d"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
-
-[[package]]
-name = "dial9-tokio-telemetry"
-version = "0.3.15"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b511dfd54f5191f7eb86856fe19d8e3c8f71673bd256e1bfa1c8128fbbc0cdb"
+checksum = "9e8cbbc8955394be626249a3b52ddd6bf664373661eaeae70d87eb12bf6f20b6"
 dependencies = [
  "arc-swap",
  "bon",
  "bytes",
  "crossbeam-queue",
- "dial9-macro",
+ "dial9-trace-format",
+ "flate2",
+ "futures-util",
+ "libc",
+ "metrique",
+ "metrique-timesource",
+ "tokio",
+ "tokio-util",
+ "tracing",
+ "ulid",
+]
+
+[[package]]
+name = "dial9-perf-self-profile"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a8f65948455c504bf08576c7b5cfe93bfcaa486d661dc4ee85c2a6309ab89629"
+dependencies = [
+ "blazesym",
+ "bon",
+ "bytes",
+ "crossbeam-utils",
+ "dial9-core",
+ "dial9-trace-format",
+ "libc",
+ "perf-event-data",
+ "perf-event-open-sys2",
+ "tracing",
+]
+
+[[package]]
+name = "dial9-tokio-telemetry"
+version = "0.5.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af1244091367c805b5d98a6a590f8ab992efda06e6d2560a6967ef898ffd30a2"
+dependencies = [
+ "bon",
+ "bytes",
+ "dial9-core",
+ "dial9-perf-self-profile",
  "dial9-trace-format",
  "flate2",
  "futures-util",
  "hostname",
  "libc",
- "metrique",
  "metrique-timesource",
- "metrique-writer",
  "pin-project-lite",
  "serde",
  "serde_json",
@@ -3725,20 +3780,22 @@ dependencies = [
 
 [[package]]
 name = "dial9-trace-format"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b3636d6ec60d94840cc414dcd6a95c77b3ba0a7b86d43fd035b89662eeb5cfa7"
+checksum = "86083b7240114b2d0da4a7e9d041571d80ca3b1f92ae0ec794f1be21709daa6a"
 dependencies = [
  "dial9-trace-format-derive",
  "serde",
+ "typeid",
 ]
 
 [[package]]
 name = "dial9-trace-format-derive"
-version = "0.4.1"
+version = "0.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fff7c2855b73d0de34bc31d6dc7afbf0f6ce230a668403ac2b57b21d1ffe3928"
+checksum = "9309248f12e414d88bcc9505f78b0c9ed47f79c5b62db611493e19a612cdc9d6"
 dependencies = [
+ "proc-macro-crate",
  "proc-macro2",
  "quote",
  "syn 2.0.119",
@@ -4558,6 +4615,9 @@ version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0bf7f043f89559805f8c7cacc432749b2fa0d0a0a9ee46ce47164ed5ba7f126c"
 dependencies = [
+ "fnv",
+ "hashbrown 0.16.1",
+ "indexmap 2.14.0",
  "stable_deref_trait",
 ]
 
@@ -4581,20 +4641,20 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-auth"
-version = "1.15.0"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f54aab44c16b8463ae11b165a87c3d484780231f157bb1ed65843d591beb5abd"
+checksum = "ff461519b1a948200f163574be072753bcfb462a323f0eb426629d89872dd685"
 dependencies = [
  "async-trait",
  "aws-lc-rs",
- "base64 0.22.1",
+ "base64 0.23.1",
  "bytes",
- "chrono",
  "google-cloud-gax",
  "hex",
  "hmac 0.13.0",
  "http 1.5.0",
- "jsonwebtoken 10.4.0",
+ "jiff",
+ "jsonwebtoken",
  "reqwest",
  "rustc_version",
  "rustls",
@@ -4610,9 +4670,9 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-gax"
-version = "1.13.0"
+version = "1.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b9a46dd0fd026bbc4a5d84e6ab0c941cee6e3b057976a0bb107fdb5238ce598f"
+checksum = "c5615cff28ee59cfe52fbb4c11b8b1e77f650296e2ea4f4c2b7757ac6b19e752"
 dependencies = [
  "bytes",
  "futures",
@@ -4625,13 +4685,14 @@ dependencies = [
  "serde_json",
  "thiserror 2.0.20",
  "tokio",
+ "tokio-stream",
 ]
 
 [[package]]
 name = "google-cloud-gax-internal"
-version = "0.7.16"
+version = "0.7.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb04c54317ace06d489213f761797240b3046142a9b7ce6b9a82a9d134e193d1"
+checksum = "d2766757d877a7a8ac23da9884cb0e3f10ed9b75a0ce59801ce6b19bf9d5819e"
 dependencies = [
  "bytes",
  "futures",
@@ -4668,9 +4729,9 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-iam-v1"
-version = "1.11.0"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34cdf5acc7ef946ee2db7a7f62bd436d8395a6543b4beef110cdc061fcf578bb"
+checksum = "5f962b40234b1531e6ef73f7558871c96e117231e962086c98804323fb8d2c82"
 dependencies = [
  "async-trait",
  "bytes",
@@ -4686,9 +4747,9 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-longrunning"
-version = "1.12.0"
+version = "1.13.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e6ce05df0aea2c08472983ce2bbbed9483cbb637b89ff69a7c4ef94371fe4f2"
+checksum = "1c0363c5389ffda2b55cd8a86eef4b19a3481a48467c91dc5d77f626a9572766"
 dependencies = [
  "async-trait",
  "bytes",
@@ -4704,9 +4765,9 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-lro"
-version = "1.9.0"
+version = "1.10.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd7cca2b991d619525d72a170ca7f413cb520872702442da22ac9af650a8e786"
+checksum = "47af3deef75c14a2983c430898d960c765bddbcc9f9188ca0563108e9227cfe7"
 dependencies = [
  "google-cloud-gax",
  "google-cloud-gax-internal",
@@ -4733,14 +4794,13 @@ dependencies = [
 
 [[package]]
 name = "google-cloud-storage"
-version = "1.17.0"
+version = "1.18.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9227f65175fa91a6e41f246797917697efdadfe09dd8ea84ad8b737a71efbd28"
+checksum = "973399251b245c63f1d02d0768772833dcf1372ee358f593fb9159de8fe9c7d4"
 dependencies = [
  "async-trait",
- "base64 0.22.1",
+ "base64 0.23.1",
  "bytes",
- "chrono",
  "crc32c",
  "futures",
  "google-cloud-auth",
@@ -4755,6 +4815,7 @@ dependencies = [
  "hex",
  "http 1.5.0",
  "http-body 1.1.0",
+ "jiff",
  "md5",
  "percent-encoding",
  "prost 0.14.4",
@@ -5784,22 +5845,6 @@ dependencies = [
  "wasm-bindgen",
 ]
 
-[[package]]
-name = "jsonwebtoken"
-version = "10.4.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "eba32bfb4ffdeaca3e34431072faf01745c9b26d25504aa7a6cf5684334fc4fc"
-dependencies = [
- "aws-lc-rs",
- "base64 0.22.1",
- "getrandom 0.2.17",
- "js-sys",
- "serde",
- "serde_json",
- "signature 2.2.0",
- "zeroize",
-]
-
 [[package]]
 name = "jsonwebtoken"
 version = "11.0.0"
@@ -6357,7 +6402,6 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "2b55bfa39e6f5e44a37a59a794915ce36d04371471cf62ae365cd9703f58a5e0"
 dependencies = [
  "itoa",
- "jiff",
  "metrique-core",
  "metrique-macro",
  "metrique-service-metrics",
@@ -6366,7 +6410,6 @@ dependencies = [
  "metrique-writer-core",
  "metrique-writer-macro",
  "ryu",
- "serde_json",
  "tokio",
 ]
 
@@ -7611,6 +7654,27 @@ version = "2.3.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9b4f627cb1b25917193a259e49bdad08f671f8d9708acfd5fe0a8c1455d87220"
 
+[[package]]
+name = "perf-event-data"
+version = "0.1.8"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "575828d9d7d205188048eb1508560607a03d21eafdbba47b8cade1736c1c28e1"
+dependencies = [
+ "bitflags 2.13.1",
+ "c-enum",
+ "perf-event-open-sys2",
+]
+
+[[package]]
+name = "perf-event-open-sys2"
+version = "5.0.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d9c25955321465255e437600b54296983fab1feac2cd0c38958adeb26dbae49e"
+dependencies = [
+ "libc",
+ "memoffset",
+]
+
 [[package]]
 name = "petgraph"
 version = "0.7.1"
@@ -9443,7 +9507,6 @@ dependencies = [
 name = "rustfs-audit"
 version = "1.0.0-rc.4"
 dependencies = [
- "async-trait",
  "const-str",
  "futures",
  "hashbrown 0.17.1",
@@ -9537,7 +9600,7 @@ dependencies = [
  "base64-simd",
  "chacha20poly1305",
  "hotpath",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
  "pbkdf2 0.13.0",
  "rand 0.10.2",
  "rsa 0.10.0-rc.18",
@@ -9582,7 +9645,6 @@ dependencies = [
  "flatbuffers",
  "futures",
  "futures-util",
- "glob",
  "google-cloud-auth",
  "google-cloud-storage",
  "hex-simd",
@@ -9789,7 +9851,7 @@ dependencies = [
  "hmac 0.13.0",
  "hotpath",
  "http 1.5.0",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
  "moka",
  "openidconnect",
  "pollster",
@@ -10220,7 +10282,7 @@ dependencies = [
  "hotpath",
  "ipnetwork",
  "jiff",
- "jsonwebtoken 11.0.0",
+ "jsonwebtoken",
  "moka",
  "pollster",
  "proptest",
@@ -10441,7 +10503,6 @@ dependencies = [
  "s3s",
  "serde",
  "serde_json",
- "sha1 0.11.0",
  "sha2 0.11.0",
  "thiserror 2.0.20",
  "time",
@@ -10981,7 +11042,7 @@ checksum = "9774ba4a74de5f7b1c1451ed6cd5285a32eddb5cccb8cc655a4e50009e06477f"
 [[package]]
 name = "s3s"
 version = "0.15.0"
-source = "git+https://github.com/rustfs/s3s.git?rev=f4dedc905ec621fa85a4686df6304190b55375f6#f4dedc905ec621fa85a4686df6304190b55375f6"
+source = "git+https://github.com/rustfs/s3s.git?rev=0f6f83d98b37fd9edcaa3be573db4aa8f568e088#0f6f83d98b37fd9edcaa3be573db4aa8f568e088"
 dependencies = [
  "arc-swap",
  "arrayvec",
@@ -12813,12 +12874,28 @@ version = "0.12.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8e28f89b80c87b8fb0cf04ab448d5dd0dd0ade2f8891bae878de66a75a28600e"
 
+[[package]]
+name = "typeid"
+version = "1.0.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bc7d623258602320d5c55d1bc22793b57daff0ec7efc270ea7d55ce1d5f5471c"
+
 [[package]]
 name = "typenum"
 version = "1.20.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6f5e870be6c3b371b77fe0ee0bafb859fa4964b4404c27de1d380043c4dda20"
 
+[[package]]
+name = "ulid"
+version = "1.2.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "470dbf6591da1b39d43c14523b2b469c86879a53e8b758c8e090a470fe7b1fbe"
+dependencies = [
+ "rand 0.9.5",
+ "web-time",
+]
+
 [[package]]
 name = "unarray"
 version = "0.1.4"
@@ -12953,9 +13030,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
 [[package]]
 name = "uuid"
-version = "1.25.0"
+version = "1.26.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "f053576934f05a761a402421fbbe3d425d9366f75f978806a037b3ca481abecc"
+checksum = "b5772d71c9be8a8a6ac2117d949c5b224c1b72241bb611d9a3012edcf8af7812"
 dependencies = [
  "getrandom 0.4.3",
  "js-sys",

Cargo.toml

@@ -259,8 +259,8 @@ enumset = "1.1.14"
 faster-hex = "0.10.0"
 flate2 = "1.1.9"
 glob = "0.3.4"
-google-cloud-storage = "1.17.0"
-google-cloud-auth = "1.15.0"
+google-cloud-storage = "1.18.0"
+google-cloud-auth = "1.16.0"
 hashbrown = { version = "0.17.1" }
 # Base32 for RFC 6238 TOTP shared secrets (RFC 4648 unpadded, the alphabet
 # every authenticator app expects). Already in the graph transitively.
@@ -304,7 +304,7 @@ rustify = { version = "0.7", default-features = false }
 rustix = { version = "1.1.4" }
 rust-embed = { version = "8.12.0" }
 rustc-hash = { version = "2.1.3" }
-s3s = { git = "https://github.com/rustfs/s3s.git", rev = "f4dedc905ec621fa85a4686df6304190b55375f6", version = "0.15.0", features = ["minio"] }
+s3s = { git = "https://github.com/rustfs/s3s.git", rev = "0f6f83d98b37fd9edcaa3be573db4aa8f568e088", version = "0.15.0", features = ["minio"] }
 serial_test = "4.0.1"
 shadow-rs = { default-features = false, version = "2.0.0" }
 siphasher = "1.0.3"
@@ -327,7 +327,7 @@ tracing-subscriber = { version = "0.3.23" }
 transform-stream = "0.3.1"
 url = "2.5.8"
 urlencoding = "2.1.3"
-uuid = { version = "1.25.0" }
+uuid = { version = "1.26.0" }
 vaultrs = { version = "0.8.0" }
 tar = "0.4.46"
 walkdir = "2.5.0"
@@ -341,7 +341,7 @@ zstd = "0.13.3"
 # Observability and Metrics
 metrics = "0.24.6"
 metrics-util = "0.20"
-dial9-tokio-telemetry = "0.3"
+dial9-tokio-telemetry = "0.5.0"
 opentelemetry = { version = "0.32.0" }
 opentelemetry-appender-tracing = { version = "0.32.0" }
 opentelemetry-otlp = { version = "0.32.0" }

crates/audit/Cargo.toml

@@ -68,7 +68,6 @@ tracing = { workspace = true, features = ["std", "attributes"] }
 
 [dev-dependencies]
 rustfs-targets = { workspace = true, features = ["test-support"] }
-async-trait = { workspace = true }
 temp-env = { workspace = true }
 url = { workspace = true }
 

crates/config/src/constants/health.rs

@@ -40,6 +40,14 @@ pub const HEALTH_OBJECT_PROGRESS_LOCK_MARGIN_MS: u64 = 5_000;
 pub const ENV_HEALTH_CLUSTER_TIMEOUT_MS: &str = "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS";
 pub const DEFAULT_HEALTH_CLUSTER_TIMEOUT_MS: u64 = 2000;
 
+/// Timeout for one remote lock-client online check used by readiness (milliseconds).
+///
+/// This is intentionally shorter than the generic lock RPC timeout so
+/// `/health/ready` can report degradation instead of riding a dead peer's
+/// connect or HTTP/2 keepalive budget.
+pub const ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS: &str = "RUSTFS_HEALTH_LOCK_ONLINE_TIMEOUT_MS";
+pub const DEFAULT_HEALTH_LOCK_ONLINE_TIMEOUT_MS: u64 = 1000;
+
 /// Maximum time to wait for local node runtime readiness (storage / IAM / lock
 /// quorum) during startup before failing fast (seconds).
 ///

crates/config/src/constants/object.rs

@@ -288,6 +288,39 @@ pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0;
 
 const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE);
 
+/// Enable automatic foreground admission for large or unknown-size PutObject requests.
+///
+/// Unlike the strict experimental gate above, this default-on path only applies
+/// to requests that are large enough to create sustained erasure/RPC pressure.
+/// Small PUTs continue on the legacy path unless the strict gate is explicitly
+/// enabled.
+pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE";
+pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE: bool = true;
+
+/// Maximum large foreground PutObject requests admitted concurrently per process.
+///
+/// `0` derives a conservative default from the local disk-read scheduler cap,
+/// currently clamped to protect the commit path without making ordinary high
+/// throughput uploads single-file.
+pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT";
+pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT: usize = 0;
+
+/// Minimum object size that enters automatic large PutObject admission.
+///
+/// Requests with an unknown size are treated as large because the write pressure
+/// cannot be bounded from headers.
+pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES";
+pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES: usize = 32 * 1024 * 1024;
+
+/// Time in milliseconds a large foreground PutObject waits for a permit.
+///
+/// A short wait smooths transient bursts while still returning S3
+/// `SlowDown`/503 before body ingest when the node is already saturated.
+pub const ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
+pub const DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 250;
+
+const _: () = assert!(DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE);
+
 /// Environment variable for minimum GetObject timeout in seconds.
 ///
 /// When dynamic timeout calculation is enabled, this is the minimum timeout

crates/e2e_test/src/kms/kms_anonymous_enforcement_test.rs

@@ -0,0 +1,220 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Anonymous access to SSE-KMS objects under per-key authorization.
+//!
+//! Locks both halves of the anonymous contract decided in backlog#2028 (D4):
+//!
+//! - **Enforcement on**: anonymous requests hold no `kms` grants, so a public
+//!   bucket policy does not let them read SSE-KMS objects or write through an
+//!   SSE-KMS default-encryption rule. Both fail with `AccessDenied`.
+//! - **Enforcement off** (the default): bucket policy alone governs anonymous
+//!   access, matching the pre-enforcement behavior — public SSE-KMS objects are
+//!   decrypted and served, and anonymous writes are encrypted under the default
+//!   key.
+//!
+//! The denial today is emergent — an empty-account principal falling through to
+//! the IAM default deny — so without this file a refactor of principal
+//! construction or policy evaluation could silently flip it. Each test carries a
+//! plaintext-object positive control: a denial proves nothing while the bucket
+//! policy has not propagated.
+
+use super::common::{LocalKMSTestEnvironment, create_key_with_specific_id};
+use crate::common::{init_logging, local_http_client};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_sdk_s3::types::{
+    ServerSideEncryption, ServerSideEncryptionByDefault, ServerSideEncryptionConfiguration, ServerSideEncryptionRule,
+};
+use std::time::Duration;
+
+type TestResult = Result<(), Box<dyn std::error::Error + Send + Sync>>;
+
+const DEFAULT_KEY: &str = "kms-anon-default-key";
+const BUCKET: &str = "kms-anon-enforcement";
+const PLAIN_OBJECT: &str = "plain.txt";
+const ENCRYPTED_OBJECT: &str = "encrypted.txt";
+const PAYLOAD: &[u8] = b"kms anonymous enforcement payload";
+
+/// How long a bucket policy change may take to reach the request path.
+const POLICY_PROPAGATION: Duration = Duration::from_secs(20);
+
+/// Start a local-KMS server and build the public-bucket fixture.
+///
+/// The bucket holds a plaintext object (the positive control), an SSE-KMS
+/// object, an SSE-KMS default-encryption rule, and a bucket policy opening
+/// `GetObject`/`PutObject` to everyone. The enforcement switch defaults to off,
+/// so the enforcing case has to set it explicitly.
+async fn start_public_sse_kms_bucket(env: &mut LocalKMSTestEnvironment, enforce: bool) -> TestResult {
+    create_key_with_specific_id(&env.kms_keys_dir, DEFAULT_KEY).await?;
+
+    let key_dir = env.kms_keys_dir.clone();
+    let args = vec![
+        "--kms-enable",
+        "--kms-backend",
+        "local",
+        "--kms-key-dir",
+        key_dir.as_str(),
+        "--kms-default-key-id",
+        DEFAULT_KEY,
+    ];
+    let mut envs = vec![("RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS", "true")];
+    if enforce {
+        envs.push(("RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY", "true"));
+    }
+    env.base_env.start_rustfs_server_with_env(args, &envs).await?;
+    env.base_env.create_test_bucket(BUCKET).await?;
+
+    let owner = env.base_env.create_s3_client();
+
+    owner
+        .put_object()
+        .bucket(BUCKET)
+        .key(PLAIN_OBJECT)
+        .body(ByteStream::from_static(PAYLOAD))
+        .send()
+        .await?;
+    owner
+        .put_object()
+        .bucket(BUCKET)
+        .key(ENCRYPTED_OBJECT)
+        .body(ByteStream::from_static(PAYLOAD))
+        .server_side_encryption(ServerSideEncryption::AwsKms)
+        .ssekms_key_id(DEFAULT_KEY)
+        .send()
+        .await?;
+
+    let encryption_config = ServerSideEncryptionConfiguration::builder()
+        .rules(
+            ServerSideEncryptionRule::builder()
+                .apply_server_side_encryption_by_default(
+                    ServerSideEncryptionByDefault::builder()
+                        .sse_algorithm(ServerSideEncryption::AwsKms)
+                        .kms_master_key_id(DEFAULT_KEY)
+                        .build()?,
+                )
+                .build(),
+        )
+        .build()?;
+    owner
+        .put_bucket_encryption()
+        .bucket(BUCKET)
+        .server_side_encryption_configuration(encryption_config)
+        .send()
+        .await?;
+
+    let policy = serde_json::json!({
+        "Version": "2012-10-17",
+        "Statement": [{
+            "Sid": "PublicReadWrite",
+            "Effect": "Allow",
+            "Principal": "*",
+            "Action": ["s3:GetObject", "s3:PutObject"],
+            "Resource": [format!("arn:aws:s3:::{BUCKET}/*")]
+        }]
+    })
+    .to_string();
+    owner.put_bucket_policy().bucket(BUCKET).policy(&policy).send().await?;
+    let _ = owner.delete_public_access_block().bucket(BUCKET).send().await;
+
+    Ok(())
+}
+
+fn object_url(env: &LocalKMSTestEnvironment, key: &str) -> String {
+    format!("{}/{BUCKET}/{key}", env.base_env.url)
+}
+
+async fn anonymous_get(env: &LocalKMSTestEnvironment, key: &str) -> Result<reqwest::Response, reqwest::Error> {
+    local_http_client().get(object_url(env, key)).send().await
+}
+
+async fn anonymous_put(env: &LocalKMSTestEnvironment, key: &str) -> Result<reqwest::Response, reqwest::Error> {
+    local_http_client().put(object_url(env, key)).body(PAYLOAD).send().await
+}
+
+/// Retry the plaintext read until the public bucket policy is live.
+async fn wait_for_public_read(env: &LocalKMSTestEnvironment) -> TestResult {
+    let deadline = tokio::time::Instant::now() + POLICY_PROPAGATION;
+    loop {
+        let status = anonymous_get(env, PLAIN_OBJECT).await?.status();
+        if status.as_u16() == 200 {
+            return Ok(());
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(format!("positive control never became readable: anonymous GET {PLAIN_OBJECT} -> {status}").into());
+        }
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+}
+
+async fn assert_anonymous_denied(response: reqwest::Response, what: &str) -> TestResult {
+    let status = response.status().as_u16();
+    let body = response.text().await?;
+    assert_eq!(status, 403, "{what} must be denied, got {status}: {body}");
+    assert!(body.contains("AccessDenied"), "{what} must carry AccessDenied: {body}");
+    Ok(())
+}
+
+/// Enforcement on: a public bucket policy does not exempt anonymous requests
+/// from per-key authorization, on either the read or the default-encryption
+/// write path.
+#[tokio::test(flavor = "multi_thread")]
+async fn anonymous_sse_kms_denied_under_enforcement() -> TestResult {
+    init_logging();
+
+    let mut env = LocalKMSTestEnvironment::new().await?;
+    start_public_sse_kms_bucket(&mut env, true).await?;
+    wait_for_public_read(&env).await?;
+
+    let read = anonymous_get(&env, ENCRYPTED_OBJECT).await?;
+    assert_anonymous_denied(read, "anonymous GET of an SSE-KMS object").await?;
+
+    let write = anonymous_put(&env, "anon-write.txt").await?;
+    assert_anonymous_denied(write, "anonymous PUT through an SSE-KMS default-encryption rule").await?;
+
+    Ok(())
+}
+
+/// Enforcement off (the default): bucket policy alone governs anonymous access,
+/// and the default-encryption rule still encrypts anonymous writes.
+#[tokio::test(flavor = "multi_thread")]
+async fn anonymous_sse_kms_governed_by_bucket_policy_without_enforcement() -> TestResult {
+    init_logging();
+
+    let mut env = LocalKMSTestEnvironment::new().await?;
+    start_public_sse_kms_bucket(&mut env, false).await?;
+    wait_for_public_read(&env).await?;
+
+    let read = anonymous_get(&env, ENCRYPTED_OBJECT).await?;
+    assert_eq!(read.status().as_u16(), 200, "anonymous GET of a public SSE-KMS object must succeed");
+    assert_eq!(read.bytes().await?.as_ref(), PAYLOAD, "the object must be served decrypted");
+
+    let write = anonymous_put(&env, "anon-write.txt").await?;
+    assert_eq!(write.status().as_u16(), 200, "anonymous PUT to a public bucket must succeed");
+
+    let stored = env
+        .base_env
+        .create_s3_client()
+        .head_object()
+        .bucket(BUCKET)
+        .key("anon-write.txt")
+        .send()
+        .await?;
+    assert_eq!(
+        stored.server_side_encryption(),
+        Some(&ServerSideEncryption::AwsKms),
+        "the anonymous write must be encrypted by the bucket default rule"
+    );
+
+    Ok(())
+}

crates/e2e_test/src/kms/mod.rs

@@ -57,6 +57,9 @@ mod copy_object_version_restore_sse_test;
 #[cfg(test)]
 mod configured_roundtrip_test;
 
+#[cfg(test)]
+mod kms_anonymous_enforcement_test;
+
 #[cfg(test)]
 mod kms_authorization_negative_matrix_test;
 

crates/e2e_test/src/presigned_negative_test.rs

@@ -87,7 +87,9 @@ fn valid_config() -> PresigningConfig {
 }
 
 /// Flip bytes inside the `X-Amz-Signature=` query value without changing its
-/// length, producing a structurally valid but incorrect signature.
+/// length, producing a structurally valid but incorrect signature. Every hex
+/// digit is replaced by its complement (15 - v), which has no fixed point, so
+/// the tamper changes the value no matter which digits the signature contains.
 fn tamper_signature(uri: &str) -> String {
     let marker = "X-Amz-Signature=";
     let idx = uri.find(marker).expect("presigned uri must carry X-Amz-Signature") + marker.len();
@@ -96,10 +98,9 @@ fn tamper_signature(uri: &str) -> String {
     let (sig, tail) = rest.split_at(end);
     let tampered: String = sig
         .chars()
-        .map(|c| match c {
-            '0' => 'f',
-            'a' => '0',
-            other => other,
+        .map(|c| {
+            let v = c.to_digit(16).expect("X-Amz-Signature value must be hex");
+            char::from_digit(15 - v, 16).expect("complement of a hex digit is a hex digit")
         })
         .collect();
     assert_ne!(sig, tampered, "tamper must actually change the signature hex");

crates/ecstore/Cargo.toml

@@ -146,7 +146,6 @@ bytes = { workspace = true, features = ["serde"] }
 byteorder = { workspace = true }
 chrono = { workspace = true, features = ["serde"] }
 jiff = { workspace = true, features = ["serde"] }
-glob = { workspace = true }
 thiserror.workspace = true
 flatbuffers.workspace = true
 futures.workspace = true

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

@@ -27,7 +27,7 @@ use rustfs_protos::{
     ConnectionEvictionLogLevel, evict_failed_connection_with_log_level, models::PingBodyBuilder,
     proto_gen::node_service::node_service_client::NodeServiceClient,
 };
-use std::time::Duration;
+use std::{sync::OnceLock, time::Duration};
 use tokio::time::timeout;
 use tonic::Request;
 use tonic::service::interceptor::InterceptedService;
@@ -44,11 +44,35 @@ pub struct RemoteClient {
 }
 
 impl RemoteClient {
+    const ONLINE_CHECK_RESOURCE: &'static str = "health-lock-online";
+
     pub fn new(endpoint: String) -> Self {
         Self { addr: endpoint }
     }
 
+    fn ping_body() -> Bytes {
+        static BODY: OnceLock<Bytes> = OnceLock::new();
+        BODY.get_or_init(|| {
+            let mut fbb = flatbuffers::FlatBufferBuilder::new();
+            let payload = fbb.create_vector(b"health-check");
+            let mut builder = PingBodyBuilder::new(&mut fbb);
+            builder.add_payload(payload);
+            let root = builder.finish();
+            fbb.finish(root, None);
+            Bytes::copy_from_slice(fbb.finished_data())
+        })
+        .clone()
+    }
+
     fn build_ping_request() -> PingRequest {
+        PingRequest {
+            version: 1,
+            body: Self::ping_body(),
+        }
+    }
+
+    #[cfg(test)]
+    fn build_fresh_ping_request_for_test() -> PingRequest {
         let mut fbb = flatbuffers::FlatBufferBuilder::new();
         let payload = fbb.create_vector(b"health-check");
         let mut builder = PingBodyBuilder::new(&mut fbb);
@@ -164,6 +188,16 @@ impl RemoteClient {
         )
     }
 
+    fn online_check_timeout() -> Duration {
+        Duration::from_millis(
+            rustfs_utils::get_env_u64(
+                rustfs_config::ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS,
+                rustfs_config::DEFAULT_HEALTH_LOCK_ONLINE_TIMEOUT_MS,
+            )
+            .max(1),
+        )
+    }
+
     async fn execute_rpc<T, F>(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result<T, LockError>
     where
         F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
@@ -547,24 +581,37 @@ impl LockClient for RemoteClient {
     }
 
     async fn is_online(&self) -> bool {
-        // Use Ping interface to test if remote service is online
-        let mut client = match self.get_client().await {
-            Ok(client) => client,
-            Err(_) => {
-                info!("remote client {} connection failed", self.addr);
-                return false;
-            }
-        };
-
-        let ping_req = Request::new(Self::build_ping_request());
-
-        match client.ping(ping_req).await {
-            Ok(_) => {
-                info!("remote client {} is online", self.addr);
+        let online_timeout = Self::online_check_timeout();
+        match timeout(online_timeout, async {
+            let mut client = self.get_client().await?;
+            let ping_req = Request::new(Self::build_ping_request());
+            self.execute_rpc("ping", Self::ONLINE_CHECK_RESOURCE, client.ping(ping_req))
+                .await?;
+            Ok::<(), LockError>(())
+        })
+        .await
+        {
+            Ok(Ok(())) => {
+                debug!(addr = %self.addr, timeout_ms = online_timeout.as_millis(), "remote lock client is online");
                 true
             }
+            Ok(Err(err)) => {
+                debug!(
+                    addr = %self.addr,
+                    timeout_ms = online_timeout.as_millis(),
+                    error = %err,
+                    "remote lock client online check failed"
+                );
+                false
+            }
             Err(_) => {
-                info!("remote client {} ping failed", self.addr);
+                let reason = format!("online check timed out after {:?}", online_timeout);
+                warn!(
+                    addr = %self.addr,
+                    timeout_ms = online_timeout.as_millis(),
+                    "remote lock client online check timed out"
+                );
+                self.evict_connection("ping", &reason, Self::ONLINE_CHECK_RESOURCE).await;
                 false
             }
         }
@@ -651,6 +698,15 @@ mod tests {
         );
     }
 
+    #[test]
+    fn cached_ping_request_matches_fresh_flatbuffer_payload() {
+        let cached = RemoteClient::build_ping_request();
+        let fresh = RemoteClient::build_fresh_ping_request_for_test();
+
+        assert_eq!(cached.version, fresh.version);
+        assert_eq!(cached.body, fresh.body);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn test_remote_client_acquire_lock_uses_rpc_timeout_and_evicts_connection() {
@@ -779,6 +835,48 @@ mod tests {
         accept_task.abort();
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn test_remote_client_is_online_uses_health_timeout_and_evicts_connection() {
+        ensure_test_rpc_secret();
+        let Some((addr, accept_task)) = spawn_hanging_listener().await else {
+            return;
+        };
+        cache_lazy_channel(&addr).await;
+        assert!(runtime_sources::test_node_channel_is_cached(&addr).await);
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS, Some("50")),
+                (rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("1000")),
+            ],
+            async {
+                let client = RemoteClient::new(addr.clone());
+                let started_at = tokio::time::Instant::now();
+
+                let online = client.is_online().await;
+                let elapsed = started_at.elapsed();
+
+                assert!(!online, "hanging remote lock peer must not be reported online");
+                assert!(
+                    elapsed >= Duration::from_millis(40),
+                    "remote online check should honor configured health timeout, got {elapsed:?}"
+                );
+                assert!(
+                    elapsed < Duration::from_secs(1),
+                    "health timeout should keep readiness probes bounded, got {elapsed:?}"
+                );
+                assert!(
+                    !runtime_sources::test_node_channel_is_cached(&addr).await,
+                    "online-check timeout should evict cached connection"
+                );
+            },
+        )
+        .await;
+
+        accept_task.abort();
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn test_remote_client_refresh_tonic_error_evicts_connection() {
@@ -906,4 +1004,21 @@ mod tests {
             assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(1));
         });
     }
+
+    #[test]
+    #[serial_test::serial]
+    fn test_remote_client_online_timeout_honors_configured_deadline() {
+        temp_env::with_var(rustfs_config::ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS, None::<&str>, || {
+            assert_eq!(
+                RemoteClient::online_check_timeout(),
+                Duration::from_millis(rustfs_config::DEFAULT_HEALTH_LOCK_ONLINE_TIMEOUT_MS)
+            );
+        });
+        temp_env::with_var(rustfs_config::ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS, Some("50"), || {
+            assert_eq!(RemoteClient::online_check_timeout(), Duration::from_millis(50));
+        });
+        temp_env::with_var(rustfs_config::ENV_HEALTH_LOCK_ONLINE_TIMEOUT_MS, Some("0"), || {
+            assert_eq!(RemoteClient::online_check_timeout(), Duration::from_millis(1));
+        });
+    }
 }

crates/ecstore/src/config/com.rs

@@ -2793,12 +2793,13 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        SERVER_CONFIG_LOCK, ServerConfigSnapshot, apply_dynamic_config_for_sub_sys_with, config_task_join_error,
-        configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config,
-        lookup_configs, new_and_save_server_config, read_config, read_config_no_lock_preserve_empty_with_metadata,
-        read_config_preserve_empty, read_config_with_metadata, read_config_without_migrate, read_server_config_snapshot,
-        save_server_config, save_server_config_snapshot, save_server_config_snapshot_with_generation,
-        server_config_transaction_lock_path, should_warn_ignored_scalar_section, storage_class_kvs_mut,
+        SERVER_CONFIG_LOCK, ServerConfigSnapshot, apply_dynamic_config_for_sub_sys_with, build_scalar_config_object,
+        config_task_join_error, configs_semantically_equal, decode_server_config_blob, encode_server_config_blob,
+        heal_config_descriptor, is_standard_object_server_config, lookup_configs, new_and_save_server_config, read_config,
+        read_config_no_lock_preserve_empty_with_metadata, read_config_preserve_empty, read_config_with_metadata,
+        read_config_without_migrate, read_server_config_snapshot, save_server_config, save_server_config_snapshot,
+        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, should_warn_ignored_scalar_section,
+        storage_class_kvs_mut,
     };
     use crate::config::{audit, heal, notify, oidc, scanner};
     use crate::disk::endpoint::Endpoint;
@@ -3541,6 +3542,31 @@ mod tests {
         cfg
     }
 
+    /// `Config::new()` (and every decode built on it) reads the process-global
+    /// `rustfs_config::server_config::DEFAULT_KVS` OnceLock at call time, and
+    /// other tests in this binary register it via `crate::config::init()`
+    /// mid-run. Equality assertions must therefore normalize both sides with
+    /// one snapshot taken after both configs exist, never against a later
+    /// `Config::new()`.
+    fn default_kvs_snapshot() -> Option<&'static std::collections::HashMap<String, KVS>> {
+        rustfs_config::server_config::DEFAULT_KVS.get()
+    }
+
+    /// Fills the default sections `cfg` is missing from an explicit
+    /// [`default_kvs_snapshot`], mirroring `Config::set_defaults`.
+    fn filled_with_default_kvs(mut cfg: Config, snapshot: Option<&std::collections::HashMap<String, KVS>>) -> Config {
+        if let Some(defaults) = snapshot {
+            for (sub_sys, kvs) in defaults {
+                cfg.0
+                    .entry(sub_sys.clone())
+                    .or_default()
+                    .entry(DEFAULT_DELIMITER.to_string())
+                    .or_insert_with(|| kvs.clone());
+            }
+        }
+        cfg
+    }
+
     #[test]
     fn test_external_scanner_config_decodes_with_defaults() {
         let cfg =
@@ -3630,7 +3656,9 @@ mod tests {
         }"#;
 
         let cfg = decode_server_config_blob(seed).expect("root heal null should mean no persisted override");
-        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none());
+        // The heal section may hold registered defaults, so assert on the
+        // semantic diff instead of the section's presence.
+        assert!(build_scalar_config_object(&cfg, heal_config_descriptor()).is_empty());
         assert!(!is_standard_object_server_config(seed));
 
         let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("legacy seed should canonicalize on an authorized save");
@@ -3665,7 +3693,12 @@ mod tests {
             let input = format!(r#"{{"version":"33","storageclass":{{"standard":"","rrs":""}},{section}}}"#);
             let cfg = decode_server_config_blob(input.as_bytes())
                 .unwrap_or_else(|err| panic!("legacy scalar section {section} should be ignored, got: {err}"));
-            assert_eq!(cfg, base, "ignored section {section} must contribute no overrides");
+            let snapshot = default_kvs_snapshot();
+            assert_eq!(
+                filled_with_default_kvs(cfg.clone(), snapshot),
+                filled_with_default_kvs(base.clone(), snapshot),
+                "ignored section {section} must contribute no overrides"
+            );
             assert!(
                 !is_standard_object_server_config(input.as_bytes()),
                 "seed with {section} must not count as standard so a save rewrites it"
@@ -3743,12 +3776,19 @@ mod tests {
     fn valid_heal_object_and_kvs_array_shapes_remain_accepted() {
         let empty_object = br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":{}}"#;
         let cfg = decode_server_config_blob(empty_object).expect("empty heal object should decode as no override");
-        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none());
+        // The heal section may hold registered defaults, so assert on the
+        // semantic diff instead of the section's presence.
+        assert!(build_scalar_config_object(&cfg, heal_config_descriptor()).is_empty());
 
         let kvs_array =
             br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":[{"key":"bitrot_cycle","value":"off"}]}"#;
         let cfg = decode_server_config_blob(kvs_array).expect("heal KVS array should decode");
-        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_some());
+        assert_eq!(
+            build_scalar_config_object(&cfg, heal_config_descriptor())
+                .get(HEAL_BITROT_CYCLE)
+                .and_then(Value::as_str),
+            Some("off")
+        );
     }
 
     #[test]
@@ -4900,8 +4940,12 @@ mod tests {
     fn test_fallback_returns_default_config_when_recovery_enabled() {
         let cfg = fallback_server_config_after_corruption(corrupt_config_error(), "config/config.json", true)
             .expect("recovery enabled must fall back to the default config");
+        let snapshot = default_kvs_snapshot();
         assert!(
-            configs_semantically_equal(&cfg, &Config::new()),
+            configs_semantically_equal(
+                &filled_with_default_kvs(cfg, snapshot),
+                &filled_with_default_kvs(Config(std::collections::HashMap::new()), snapshot)
+            ),
             "fallback config should be the default server config"
         );
     }
@@ -5220,7 +5264,7 @@ mod tests {
     #[tokio::test]
     async fn server_config_snapshot_serializes_read_modify_write_transactions() {
         let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode");
-        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None));
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline.clone()), None));
         let first = read_server_config_snapshot(store.clone())
             .await
             .expect("first config snapshot");
@@ -5234,7 +5278,11 @@ mod tests {
             .await
             .expect("second transaction should acquire after the first snapshot is dropped")
             .expect("second config snapshot");
-        assert!(configs_semantically_equal(&second.config, &Config::new()));
+        // Compare raw bytes against the baseline blob rather than a fresh
+        // Config::new(): the process-global DEFAULT_KVS can be registered by a
+        // sibling test mid-run, which would make a Config::new() evaluated here
+        // diverge from the baseline encoded above.
+        assert_eq!(second.raw.as_deref(), Some(baseline.as_slice()));
     }
 
     #[tokio::test]
@@ -5514,8 +5562,12 @@ mod tests {
             .expect("unrecoverable corruption should fall back to the default config");
 
         assert_eq!(store.heal_calls.load(Ordering::SeqCst), 1, "heal should be attempted before falling back");
+        let snapshot = default_kvs_snapshot();
         assert!(
-            configs_semantically_equal(&cfg, &Config::new()),
+            configs_semantically_equal(
+                &filled_with_default_kvs(cfg, snapshot),
+                &filled_with_default_kvs(Config(std::collections::HashMap::new()), snapshot)
+            ),
             "fallback config should be the default server config"
         );
     }

crates/ecstore/src/core/pools.rs

@@ -96,6 +96,8 @@ const LOG_SUBSYSTEM_POOLS: &str = "pools";
 const EVENT_DECOMMISSION_STATE: &str = "decommission_state";
 const EVENT_DECOMMISSION_BUCKET: &str = "decommission_bucket";
 const EVENT_DECOMMISSION_ENTRY: &str = "decommission_entry";
+const POOL_ACTIVATION_FLEET_PROOF_REQUIRED: &str = "pool activation requires a live fleet capability proof";
+const POOL_ACTIVATION_FLEET_PROOF_EXPIRED: &str = "pool activation fleet capability proof expired before commit";
 const DECOMMISSION_STAGE_MIGRATE_OBJECT: &str = "migrate_object";
 const DECOMMISSION_STAGE_CLEANUP_PREFLIGHT: &str = "cleanup_preflight";
 const DECOMMISSION_STAGE_SOURCE_CLEANUP: &str = "source_cleanup";
@@ -1832,6 +1834,13 @@ pub(crate) struct PoolRebalanceActivationFence {
 }
 
 impl PoolRebalanceActivationFence {
+    pub(crate) fn set_fleet_proof(
+        &mut self,
+        fleet_proof: Option<crate::services::notification_sys::CrossPoolFenceFleetProofToken>,
+    ) {
+        self.fleet_proof = fleet_proof;
+    }
+
     pub(crate) fn ensure_held(&self) -> Result<()> {
         #[cfg(test)]
         let forced_lost = self.forced_lost.load(Ordering::Acquire);
@@ -1845,7 +1854,7 @@ impl PoolRebalanceActivationFence {
             .as_ref()
             .is_some_and(|proof| !crate::services::notification_sys::cross_pool_fence_fleet_proof_matches(proof))
         {
-            return Err(Error::other("pool activation fleet capability proof expired before commit"));
+            return Err(Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED));
         }
 
         Ok(())
@@ -1901,7 +1910,17 @@ pub(crate) async fn acquire_pool_activation_fleet_proof(
     }
     crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof()
         .map(Some)
-        .ok_or_else(|| Error::other("pool activation requires a live fleet capability proof"))
+        .ok_or_else(|| Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED))
+}
+
+pub(crate) fn is_pool_activation_fleet_proof_error(err: &Error) -> bool {
+    // Save-stage helpers add context by formatting the original error, so the
+    // marker may be nested in the display string. Restrict matching to the
+    // `Error::other` I/O shape used by this activation path.
+    matches!(err, Error::Io(io_error) if io_error.kind() == std::io::ErrorKind::Other && {
+        let message = io_error.to_string();
+        message.contains(POOL_ACTIVATION_FLEET_PROOF_REQUIRED) || message.contains(POOL_ACTIVATION_FLEET_PROOF_EXPIRED)
+    })
 }
 
 #[cfg(test)]
@@ -10828,6 +10847,15 @@ mod tests {
     use crate::bucket::replication::{ReplicationState, ReplicationStatusType};
     use serde::Serialize;
 
+    #[test]
+    fn pool_activation_fleet_proof_error_classifier_matches_only_retryable_proof_failures() {
+        assert!(is_pool_activation_fleet_proof_error(&Error::other(POOL_ACTIVATION_FLEET_PROOF_REQUIRED)));
+        assert!(is_pool_activation_fleet_proof_error(&Error::other(POOL_ACTIVATION_FLEET_PROOF_EXPIRED)));
+        let wrapped = format!("rebalance meta save failed during start_rebalance: {POOL_ACTIVATION_FLEET_PROOF_EXPIRED}");
+        assert!(is_pool_activation_fleet_proof_error(&Error::other(wrapped)));
+        assert!(!is_pool_activation_fleet_proof_error(&Error::ConfigNotFound));
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn decommission_activation_fence_loss_after_durable_save_blocks_publication() {

crates/ecstore/src/error/mod.rs

@@ -157,6 +157,8 @@ pub enum StorageError {
     InvalidPartNumber(usize),
     #[error("Your proposed upload is smaller than the minimum allowed size. Part {0} size {1} is less than minimum {2}")]
     EntityTooSmall(usize, i64, i64),
+    #[error("multipart upload size {0} exceeds the configured limit {1}")]
+    EntityTooLarge(u64, u64),
 
     // ── Erasure / Quorum ─────────────────────────────────────────────
     #[error("erasure read quorum")]
@@ -554,6 +556,7 @@ impl Clone for StorageError {
             StorageError::DecommissionNotStarted => StorageError::DecommissionNotStarted,
             StorageError::InvalidPart(a, b, c) => StorageError::InvalidPart(*a, b.clone(), c.clone()),
             StorageError::EntityTooSmall(a, b, c) => StorageError::EntityTooSmall(*a, *b, *c),
+            StorageError::EntityTooLarge(a, b) => StorageError::EntityTooLarge(*a, *b),
             StorageError::DoneForNow => StorageError::DoneForNow,
             StorageError::DecommissionAlreadyRunning => StorageError::DecommissionAlreadyRunning,
             StorageError::RebalanceAlreadyRunning => StorageError::RebalanceAlreadyRunning,
@@ -673,6 +676,7 @@ impl StorageError {
             StorageError::InsufficientWriteQuorum(_, _) => StorageErrorCode::InsufficientWriteQuorum,
             StorageError::PreconditionFailed => StorageErrorCode::PreconditionFailed,
             StorageError::EntityTooSmall(_, _, _) => StorageErrorCode::EntityTooSmall,
+            StorageError::EntityTooLarge(_, _) => StorageErrorCode::EntityTooLarge,
             StorageError::InvalidRangeSpec(_) => StorageErrorCode::InvalidRangeSpec,
             StorageError::NotModified => StorageErrorCode::NotModified,
             StorageError::InvalidPartNumber(_) => StorageErrorCode::InvalidPartNumber,
@@ -795,6 +799,7 @@ impl StorageError {
             StorageErrorCode::EntityTooSmall => {
                 Some(StorageError::EntityTooSmall(Default::default(), Default::default(), Default::default()))
             }
+            StorageErrorCode::EntityTooLarge => Some(StorageError::EntityTooLarge(Default::default(), Default::default())),
             StorageErrorCode::InvalidRangeSpec => Some(StorageError::InvalidRangeSpec(Default::default())),
             StorageErrorCode::NotModified => Some(StorageError::NotModified),
             StorageErrorCode::InvalidPartNumber => Some(StorageError::InvalidPartNumber(Default::default())),

crates/ecstore/src/services/notification_sys.rs

@@ -234,6 +234,39 @@ pub(crate) fn install_cross_pool_fence_fleet_proof_for_test() {
     });
 }
 
+#[cfg(test)]
+pub(crate) struct CrossPoolFenceFleetProofGuard {
+    previous_proof: Option<FleetCapabilityProof>,
+    previous_topology_conflict: bool,
+}
+
+#[cfg(test)]
+impl Drop for CrossPoolFenceFleetProofGuard {
+    fn drop(&mut self) {
+        let mut state = cross_pool_fence_fleet_proof_slot()
+            .write()
+            .unwrap_or_else(std::sync::PoisonError::into_inner);
+        state.proof = self.previous_proof.take();
+        state.topology_conflict = self.previous_topology_conflict;
+    }
+}
+
+/// Temporarily revoke the test proof so activation paths can exercise their
+/// fail-closed behavior without changing the process-wide topology binding.
+#[cfg(test)]
+pub(crate) fn without_cross_pool_fence_fleet_proof_for_test() -> CrossPoolFenceFleetProofGuard {
+    let mut state = cross_pool_fence_fleet_proof_slot()
+        .write()
+        .unwrap_or_else(std::sync::PoisonError::into_inner);
+    let guard = CrossPoolFenceFleetProofGuard {
+        previous_proof: state.proof.clone(),
+        previous_topology_conflict: state.topology_conflict,
+    };
+    state.proof = None;
+    state.topology_conflict = true;
+    guard
+}
+
 #[cfg(any(test, feature = "test-util"))]
 pub fn rotate_cross_pool_fence_fleet_proof_for_test() -> bool {
     let mut state = cross_pool_fence_fleet_proof_slot()

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

@@ -570,10 +570,13 @@ impl ECStore {
     where
         S: EcstoreObjectIO + StorageNamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
     {
-        let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
+        // Lock order: pool_meta_save_gate -> pool.bin -> rebalance.bin.
         let mut pool_meta_guard = self.pool_meta_save_gate.lock().await;
         pool_meta_guard.ensure_write_safe("rebalance worker activation")?;
-        let activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), fleet_proof).await?;
+        // Classify the durable rebalance record while holding both namespace
+        // fences. A terminal record is a no-op and must not depend on the
+        // notification subsystem having published a fleet proof yet.
+        let mut activation_fence = acquire_pool_rebalance_activation_locks(pool.clone(), None).await?;
         let pool_meta = self
             .load_runtime_pool_meta_under_activation_fence(&mut pool_meta_guard, &activation_fence, "rebalance worker activation")
             .await?;
@@ -597,10 +600,17 @@ impl ECStore {
         }
 
         activation_fence.ensure_held()?;
-        if !is_rebalance_conflicting_with_decommission(&persisted) {
+        if !crate::services::rebalance::rebalance_requires_worker_activation(&persisted) {
             return Ok(RebalanceWorkerActivationFence::NotStartedTerminal);
         }
 
+        // Active worker admission still requires the fail-closed fleet proof.
+        // Attach it immediately before the final fence validation so expiry or
+        // topology changes are checked again at every later commit boundary.
+        let fleet_proof = acquire_pool_activation_fleet_proof(&self.ctx).await?;
+        activation_fence.set_fleet_proof(fleet_proof);
+        activation_fence.ensure_held()?;
+
         Ok(RebalanceWorkerActivationFence::Ready(Box::new(activation_fence)))
     }
 
@@ -1476,6 +1486,64 @@ mod tests {
         assert_activation_locks_released(&store).await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_worker_skips_terminal_metadata_without_fleet_proof() {
+        let rebalance_id = "terminal-metadata-without-proof";
+        let completed = RebalanceMeta {
+            id: rebalance_id.to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Completed,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(Some(completed)).await;
+        let _proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test();
+
+        let activation = store
+            .fence_rebalance_worker_activation(store.pools[0].clone(), rebalance_id)
+            .await
+            .expect("terminal metadata should not require a fleet proof");
+        assert!(matches!(activation, RebalanceWorkerActivationFence::NotStartedTerminal));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn rebalance_worker_still_requires_fleet_proof_for_active_metadata() {
+        let rebalance_id = "active-metadata-without-proof";
+        let active = RebalanceMeta {
+            id: rebalance_id.to_string(),
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        let (_temp_dirs, store, _other_store) = crate::services::rebalance::test_two_pool_stores(Some(active)).await;
+        let _proof_guard = crate::services::notification_sys::without_cross_pool_fence_fleet_proof_for_test();
+
+        let err = match store
+            .fence_rebalance_worker_activation(store.pools[0].clone(), rebalance_id)
+            .await
+        {
+            Ok(_) => panic!("active metadata must not be admitted without a fleet proof"),
+            Err(err) => err,
+        };
+        assert!(
+            err.to_string()
+                .contains("pool activation requires a live fleet capability proof")
+        );
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn rebalance_activation_adopts_commit_after_post_save_fence_loss() {

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

@@ -214,6 +214,14 @@ pub(super) fn is_rebalance_in_progress(meta: &RebalanceMeta) -> bool {
     meta.pool_stats.iter().any(is_rebalance_pool_active)
 }
 
+/// Persisted rebalance metadata requires worker activation only while it has
+/// not reached a durable terminal marker and at least one pool is still marked
+/// active. Merely finding `rebalance.bin` is not evidence that admission is
+/// required: terminal metadata is retained for status reporting.
+pub(crate) fn rebalance_requires_worker_activation(meta: &RebalanceMeta) -> bool {
+    meta.stopped_at.is_none() && is_rebalance_in_progress(meta)
+}
+
 pub(crate) fn is_rebalance_conflicting_with_decommission(meta: &RebalanceMeta) -> bool {
     is_rebalance_in_progress(meta)
 }

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

@@ -49,8 +49,8 @@ mod worker;
 
 #[cfg(feature = "test-util")]
 pub use entry::test_util::PausedRebalanceEntryTestFixture;
-pub(crate) use meta::is_rebalance_conflicting_with_decommission;
 pub use meta::{decode_rebalance_stop_propagation_record, encode_rebalance_stop_propagation_record};
+pub(crate) use meta::{is_rebalance_conflicting_with_decommission, rebalance_requires_worker_activation};
 pub use types::{
     DiskStat, RebalSaveOpt, RebalStatus, RebalanceCleanupWarningEntry, RebalanceCleanupWarnings, RebalanceInfo, RebalanceMeta,
     RebalanceStats, RebalanceStopPropagationRecord,

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

@@ -22,11 +22,11 @@ use super::meta::{
     is_rebalance_in_progress, is_rebalance_meta_replaceable_for_new_id, is_rebalance_stopped_terminal_event,
     mark_rebalance_bucket_done, merge_rebalance_bucket_lists, merge_rebalance_meta, next_rebal_bucket_from_stat,
     percent_free_ratio, rebalance_goal_reached, rebalance_meta_load_no_data_error, rebalance_meta_load_unknown_format_error,
-    rebalance_meta_load_unknown_version_error, record_rebalance_cleanup_warning_in_meta, remove_rebalanced_buckets_from_queue,
-    resolve_next_rebalance_bucket, resolve_rebalance_participants, should_accept_rebalance_stats_update,
-    should_ignore_rebalance_data_usage_cache, should_pool_participate, should_preserve_rebalance_stopped_state,
-    should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state, take_bucket_from_rebalance_queue,
-    validate_init_rebalance_state, validate_start_rebalance_state,
+    rebalance_meta_load_unknown_version_error, rebalance_requires_worker_activation, record_rebalance_cleanup_warning_in_meta,
+    remove_rebalanced_buckets_from_queue, resolve_next_rebalance_bucket, resolve_rebalance_participants,
+    should_accept_rebalance_stats_update, should_ignore_rebalance_data_usage_cache, should_pool_participate,
+    should_preserve_rebalance_stopped_state, should_skip_start_rebalance, stop_rebalance_meta_snapshot, stop_rebalance_state,
+    take_bucket_from_rebalance_queue, validate_init_rebalance_state, validate_start_rebalance_state,
 };
 use super::migration::{
     MigrationBackend, MigrationVersionResult, migrate_entry_version, migrate_entry_version_with_retry_wait,
@@ -3386,6 +3386,52 @@ fn test_is_rebalance_in_progress_only_started_participants() {
     assert!(is_rebalance_in_progress(&meta));
 }
 
+#[test]
+fn test_rebalance_requires_worker_activation_only_for_active_non_stopped_metadata() {
+    let now = OffsetDateTime::now_utc();
+    let active = RebalanceMeta {
+        pool_stats: vec![RebalanceStats {
+            participating: true,
+            info: RebalanceInfo {
+                status: RebalStatus::Started,
+                ..Default::default()
+            },
+            ..Default::default()
+        }],
+        ..Default::default()
+    };
+    let stopped_active = RebalanceMeta {
+        stopped_at: Some(now),
+        pool_stats: active.pool_stats.clone(),
+        ..Default::default()
+    };
+
+    assert!(rebalance_requires_worker_activation(&active));
+    for status in [
+        RebalStatus::Completed,
+        RebalStatus::Stopped,
+        RebalStatus::Failed,
+        RebalStatus::None,
+    ] {
+        let terminal = RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+        assert!(
+            !rebalance_requires_worker_activation(&terminal),
+            "terminal status {status:?} must not resume"
+        );
+    }
+    assert!(!rebalance_requires_worker_activation(&stopped_active));
+}
+
 #[test]
 fn test_is_rebalance_conflicting_with_decommission_true_when_in_progress() {
     let now = OffsetDateTime::now_utc();

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

@@ -84,11 +84,13 @@ use rustfs_rio::TryGetIndex;
 use rustfs_utils::http::SSEC_ALGORITHM_HEADER;
 #[cfg(test)]
 use rustfs_utils::http::SUFFIX_COMPRESSION;
+use rustfs_utils::http::{SUFFIX_MAX_TOTAL_OBJECT_SIZE, get_consistent_str};
 use std::future::Future;
 #[cfg(test)]
 use std::sync::atomic::AtomicBool;
 #[cfg(any(test, feature = "test-util"))]
 use std::sync::atomic::{AtomicUsize, Ordering};
+use std::sync::{Mutex, OnceLock};
 #[cfg(any(test, feature = "test-util"))]
 use std::time::Duration;
 #[cfg(test)]
@@ -97,6 +99,83 @@ use tokio::task::JoinSet;
 
 const MULTIPART_LIST_IO_CONCURRENCY: usize = 16;
 
+static CAPPED_MULTIPART_STAGING: OnceLock<Mutex<HashMap<String, Arc<tokio::sync::Semaphore>>>> = OnceLock::new();
+
+struct CappedMultipartStagingGuard {
+    upload_id_path: String,
+    permit: Option<tokio::sync::OwnedSemaphorePermit>,
+}
+
+impl Drop for CappedMultipartStagingGuard {
+    fn drop(&mut self) {
+        // Release the permit before checking the Arc count so a concurrent
+        // Abort/Complete cleanup can remove the now-unused map entry.
+        self.permit.take();
+        remove_capped_multipart_staging_semaphore(&self.upload_id_path);
+    }
+}
+
+fn capped_multipart_staging_semaphore(upload_id_path: &str) -> Arc<tokio::sync::Semaphore> {
+    CAPPED_MULTIPART_STAGING
+        .get_or_init(|| Mutex::new(HashMap::new()))
+        .lock()
+        .expect("capped multipart staging semaphore map should not be poisoned")
+        .entry(upload_id_path.to_owned())
+        .or_insert_with(|| Arc::new(tokio::sync::Semaphore::new(1)))
+        .clone()
+}
+
+fn remove_capped_multipart_staging_semaphore(upload_id_path: &str) {
+    if let Some(map) = CAPPED_MULTIPART_STAGING.get() {
+        let mut map = map
+            .lock()
+            .expect("capped multipart staging semaphore map should not be poisoned");
+        let removable = map
+            .get(upload_id_path)
+            .is_some_and(|semaphore| Arc::strong_count(semaphore) == 1);
+        if removable {
+            map.remove(upload_id_path);
+        }
+    }
+}
+
+fn multipart_size_limit_from_metadata(metadata: &HashMap<String, String>) -> Result<Option<u64>> {
+    if !contains_key_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) {
+        return Ok(None);
+    }
+
+    let Some(value) = get_consistent_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) else {
+        return Err(Error::InvalidArgument(
+            "multipart upload".to_string(),
+            SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+            "missing or conflicting internal size limit".to_string(),
+        ));
+    };
+
+    let limit = value.parse::<u64>().map_err(|_| {
+        Error::InvalidArgument(
+            "multipart upload".to_string(),
+            SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+            "invalid internal size limit".to_string(),
+        )
+    })?;
+    Ok(Some(limit))
+}
+
+fn admitted_multipart_size(current: u64, candidate: u64, limit: u64) -> Result<u64> {
+    let total = current.checked_add(candidate).ok_or_else(|| {
+        Error::InvalidArgument(
+            "multipart upload".to_string(),
+            SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+            "logical size overflow".to_string(),
+        )
+    })?;
+    if total > limit {
+        return Err(Error::EntityTooLarge(total, limit));
+    }
+    Ok(total)
+}
+
 pub(crate) struct StaleMultipartCleanupGuard {
     file_info: FileInfo,
     upload_path: String,
@@ -115,8 +194,13 @@ impl StaleMultipartCleanupGuard {
 
     pub(crate) async fn delete(self, set: &SetDisks) -> Result<()> {
         fence_commit_on_lock_loss(Some(&self.lock_guard), "stale_multipart_cleanup", &self.upload_path)?;
-        set.delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &self.upload_path, self.write_quorum)
-            .await
+        let result = set
+            .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &self.upload_path, self.write_quorum)
+            .await;
+        if result.is_ok() {
+            remove_capped_multipart_staging_semaphore(&self.upload_path);
+        }
+        result
     }
 }
 
@@ -635,6 +719,61 @@ async fn multipart_upload_paths_on_disk(disk: DiskStore, bucket: &str) -> disk::
 }
 
 impl SetDisks {
+    async fn current_multipart_logical_size(
+        &self,
+        bucket: &str,
+        object: &str,
+        upload_id: &str,
+        upload_id_path: &str,
+        fi: &FileInfo,
+        replacing_part: usize,
+    ) -> Result<u64> {
+        let online_disks = self.get_disks_internal().await;
+        let read_quorum = fi.read_quorum(self.default_read_quorum());
+        let part_path = format!(
+            "{}{}",
+            path_join_buf(&[
+                upload_id_path,
+                fi.data_dir.map(|v| v.to_string()).unwrap_or_default().as_str(),
+            ]),
+            SLASH_SEPARATOR
+        );
+        let part_numbers = match Self::list_parts(&online_disks, &part_path, read_quorum).await {
+            Ok(parts) => parts,
+            Err(DiskError::FileNotFound) => return Ok(0),
+            Err(err) => return Err(to_object_err(err.into(), vec![bucket, object, upload_id])),
+        };
+        if part_numbers.is_empty() {
+            return Ok(0);
+        }
+        let part_meta_paths = part_numbers
+            .iter()
+            .map(|number| format!("{part_path}part.{number}.meta"))
+            .collect::<Vec<_>>();
+        let existing_parts =
+            Self::read_parts(&online_disks, RUSTFS_META_MULTIPART_BUCKET, &part_meta_paths, &part_numbers, read_quorum)
+                .await
+                .map_err(|err| to_object_err(err.into(), vec![bucket, object, upload_id]))?;
+
+        existing_parts.into_iter().try_fold(0_u64, |total, part| {
+            if part.error.is_some() || part.number == replacing_part {
+                return if part.error.is_some() {
+                    Err(Error::PartMissingOrCorrupt)
+                } else {
+                    Ok(total)
+                };
+            }
+            let part_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
+            total.checked_add(part_size).ok_or_else(|| {
+                Error::InvalidArgument(
+                    "multipart upload".to_string(),
+                    SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+                    "logical size overflow".to_string(),
+                )
+            })
+        })
+    }
+
     async fn discover_multipart_upload_paths(
         &self,
         orig_bucket: &str,
@@ -1130,9 +1269,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         crate::hp_guard!("SetDisks::put_object_part");
         let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement);
 
-        let (fi, _) = self
+        let (fi, _) = match self
             .check_upload_id_exists_with_opts(bucket, object, upload_id, true, opts)
-            .await?;
+            .await
+        {
+            Ok(value) => value,
+            Err(err @ Error::InvalidUploadID(..)) => {
+                remove_capped_multipart_staging_semaphore(&upload_id_path);
+                return Err(err);
+            }
+            Err(err) => return Err(err),
+        };
+        let multipart_size_limit = multipart_size_limit_from_metadata(&fi.metadata)?;
         ensure_data_movement_upload_access(&fi, bucket, object, upload_id, opts)?;
         ensure_multipart_bucket_incarnation(&self.ctx, &fi, bucket, object, upload_id, opts.expected_bucket_incarnation_id)
             .await?;
@@ -1165,6 +1313,44 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         let part_suffix = format!("part.{part_id}");
         let tmp_part = format!("{}x{}", Uuid::new_v4(), OffsetDateTime::now_utc().unix_timestamp());
         let tmp_part_path = Arc::new(format!("{tmp_part}/{part_suffix}"));
+        let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix);
+        let part_lock_path = format!("{upload_id_path}/{part_suffix}");
+
+        // Keep at most one capped part staging locally per upload. The
+        // distributed lock below is held only for the durable admission check;
+        // it is reacquired for the short final rename, so Complete/Abort are
+        // not blocked behind a slow body upload.
+        let _capped_staging_guard = if multipart_size_limit.is_some() {
+            Some(CappedMultipartStagingGuard {
+                upload_id_path: upload_id_path.clone(),
+                permit: Some(
+                    capped_multipart_staging_semaphore(&upload_id_path)
+                        .acquire_owned()
+                        .await
+                        .map_err(|_| Error::other("capped multipart staging semaphore closed"))?,
+                ),
+            })
+        } else {
+            None
+        };
+
+        if let Some(limit) = multipart_size_limit {
+            let admission_guard = self
+                .acquire_write_lock_diag("put_object_part_admission", RUSTFS_META_MULTIPART_BUCKET, &upload_id_path)
+                .await?;
+            let declared_size = if data.size() >= 0 {
+                u64::try_from(data.size()).map_err(|_| Error::PartMissingOrCorrupt)?
+            } else if data.actual_size() >= 0 {
+                u64::try_from(data.actual_size()).map_err(|_| Error::PartMissingOrCorrupt)?
+            } else {
+                return Err(Error::PartMissingOrCorrupt);
+            };
+            let current_size = self
+                .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &fi, part_id)
+                .await?;
+            admitted_multipart_size(current_size, declared_size, limit)?;
+            drop(admission_guard);
+        }
 
         let result: Result<PartInfo> = async {
             let erasure =
@@ -1365,30 +1551,21 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 .await?;
             }
 
-            let part_path = format!("{}/{}/{}", upload_id_path, fi.data_dir.unwrap_or_default(), part_suffix);
-            let part_lock_path = format!("{upload_id_path}/{part_suffix}");
-
             #[cfg(test)]
             pause_multipart_commit(bucket, object, MultipartCommitPause::PutPartBeforeLockAcquire).await;
-            // Serialize only same-part commits (rename_part), not the whole upload.
-            // Each concurrent stream writes to its own unique temp dir (see
-            // `tmp_part` above), so the encode/stream phase never conflicts and must
-            // stay lock-free — holding a lock across it would serialize slow
-            // re-transmits of the same part and defeat the S3 "last finisher wins"
-            // semantics. The mixed-generation hazard is confined to rename_part,
-            // where two temp parts are moved cross-disk onto the SAME final
-            // part_path: interleaving there can leave shards from two generations,
-            // each individually bitrot-valid, that only surface as silent corruption
-            // at read time (backlog#853). A write lock scoped to this part number
-            // makes each same-part commit atomic across disks, so the last committer
-            // wins consistently, while different part numbers commit onto disjoint
-            // part paths and stay concurrent (issue#5961 — an uploadId-wide write
-            // lock serialized them into 503 lock-acquire timeouts). The shared
-            // uploadId read lock keeps completion/abort (which take the uploadId
-            // write lock) from racing any in-flight part commit; a guarded
-            // completion takes the object lock before the upload lock to preserve
-            // global ordering.
-            let (_upload_commit_guard, _part_commit_guard) = if opts.no_lock {
+            // Capped uploads reacquire the upload-wide write lock for the
+            // final durable check and rename. Uncapped uploads retain the
+            // concurrent encode path and only serialize the final same-part
+            // rename; completion/abort use the upload-wide write lock.
+            let (_upload_commit_guard, _part_commit_guard) = if multipart_size_limit.is_some() {
+                let upload_guard = self
+                    .acquire_write_lock_diag("put_object_part_commit", RUSTFS_META_MULTIPART_BUCKET, &upload_id_path)
+                    .await?;
+                let part_guard = self
+                    .acquire_write_lock_diag("put_object_part_commit", RUSTFS_META_MULTIPART_BUCKET, &part_lock_path)
+                    .await?;
+                (Some(upload_guard), Some(part_guard))
+            } else if opts.no_lock {
                 (None, None)
             } else {
                 let upload_guard = self
@@ -1400,8 +1577,16 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 (Some(upload_guard), Some(part_guard))
             };
             let (commit_fi, _) = self
-                .check_upload_id_exists_with_opts(bucket, object, upload_id, false, opts)
+                .check_upload_id_exists_with_opts(bucket, object, upload_id, multipart_size_limit.is_some(), opts)
                 .await?;
+            let commit_size_limit = multipart_size_limit_from_metadata(&commit_fi.metadata)?;
+            if commit_size_limit != multipart_size_limit {
+                return Err(Error::InvalidArgument(
+                    "multipart upload".to_string(),
+                    SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+                    "size limit metadata changed or is missing".to_string(),
+                ));
+            }
             ensure_data_movement_upload_access(&commit_fi, bucket, object, upload_id, opts)?;
             ensure_multipart_bucket_incarnation(
                 &self.ctx,
@@ -1431,6 +1616,14 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             }
             ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
+            if let Some(limit) = commit_size_limit {
+                let current_size = self
+                    .current_multipart_logical_size(bucket, object, upload_id, &upload_id_path, &commit_fi, part_id)
+                    .await?;
+                let candidate_size = u64::try_from(actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
+                admitted_multipart_size(current_size, candidate_size, limit)?;
+            }
+
             let _ = self
                 .rename_part(
                     &shuffle_disks,
@@ -1891,12 +2084,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
         let upload_id_path = Self::get_multipart_upload_dir(bucket, object, upload_id, opts.data_movement);
 
-        self.delete_all_with_quorum(
-            RUSTFS_META_MULTIPART_BUCKET,
-            &upload_id_path,
-            fi.write_quorum(self.default_write_quorum()),
-        )
-        .await
+        let result = self
+            .delete_all_with_quorum(
+                RUSTFS_META_MULTIPART_BUCKET,
+                &upload_id_path,
+                fi.write_quorum(self.default_write_quorum()),
+            )
+            .await;
+        if result.is_ok() {
+            remove_capped_multipart_staging_semaphore(&upload_id_path);
+        }
+        result
     }
     // complete_multipart_upload finished
     #[tracing::instrument(skip(self))]
@@ -2016,6 +2214,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             return Err(Error::other("part result number err"));
         }
 
+        if let Some(limit) = multipart_size_limit_from_metadata(&fi.metadata)? {
+            let mut total = 0_u64;
+            for part in &object_parts {
+                if part.error.is_some() {
+                    return Err(Error::PartMissingOrCorrupt);
+                }
+                let part_size = u64::try_from(part.actual_size).map_err(|_| Error::PartMissingOrCorrupt)?;
+                total = total.checked_add(part_size).ok_or_else(|| {
+                    Error::InvalidArgument(
+                        "multipart upload".to_string(),
+                        SUFFIX_MAX_TOTAL_OBJECT_SIZE.to_string(),
+                        "logical size overflow".to_string(),
+                    )
+                })?;
+            }
+            if total > limit {
+                return Err(Error::EntityTooLarge(total, limit));
+            }
+            rustfs_utils::http::metadata_compat::remove_str(&mut fi.metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE);
+        }
+
         let mut checksum_type = rustfs_rio::ChecksumType::NONE;
 
         if let Some(cs) = fi.metadata.get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM) {
@@ -3024,13 +3243,17 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
         };
 
-        if detach_commit_owner {
+        let result = if detach_commit_owner {
             tokio::spawn(commit)
                 .await
                 .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
         } else {
             commit.await
+        };
+        if result.is_ok() {
+            remove_capped_multipart_staging_semaphore(&upload_id_path);
         }
+        result
     }
 }
 
@@ -3096,6 +3319,34 @@ mod tests {
         assert!(multipart_bucket_incarnation_id(&nil_metadata).is_err());
     }
 
+    #[test]
+    fn multipart_size_limit_metadata_is_dual_key_and_fail_closed() {
+        let mut metadata = HashMap::new();
+        insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "100".to_string());
+        assert_eq!(multipart_size_limit_from_metadata(&metadata).unwrap(), Some(100));
+
+        metadata.insert("x-minio-internal-max-total-object-size".to_string(), "101".to_string());
+        assert!(multipart_size_limit_from_metadata(&metadata).is_err());
+
+        let mut invalid = HashMap::new();
+        insert_str(&mut invalid, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "-1".to_string());
+        assert!(multipart_size_limit_from_metadata(&invalid).is_err());
+    }
+
+    #[test]
+    fn multipart_size_admission_handles_boundaries_and_overflow() {
+        assert_eq!(admitted_multipart_size(90, 10, 100).unwrap(), 100);
+        assert!(matches!(
+            admitted_multipart_size(90, 11, 100),
+            Err(StorageError::EntityTooLarge(101, 100))
+        ));
+        assert!(admitted_multipart_size(u64::MAX - 1, 1, u64::MAX).is_ok());
+        assert!(matches!(
+            admitted_multipart_size(u64::MAX, 1, u64::MAX),
+            Err(StorageError::InvalidArgument(_, _, _))
+        ));
+    }
+
     #[test]
     fn multipart_bucket_incarnation_gate_accepts_only_current_or_same_lifetime_legacy_uploads() {
         let expected = Uuid::new_v4();
@@ -3695,11 +3946,17 @@ mod tests {
 
     async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec<Option<Uuid>> {
         let mut epochs = Vec::with_capacity(disks.len());
+        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
         for (disk_index, disk) in disks.iter().enumerate() {
-            let file_info = disk
-                .read_version("", bucket, object, "", &ReadOptions::default())
-                .await
-                .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}"));
+            let file_info = loop {
+                match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
+                    Ok(file_info) => break file_info,
+                    Err(DiskError::FileNotFound) if tokio::time::Instant::now() < deadline => {
+                        tokio::time::sleep(Duration::from_millis(25)).await;
+                    }
+                    Err(err) => panic!("disk {disk_index} should persist object metadata: {err}"),
+                }
+            };
             epochs.push(
                 file_info
                     .object_transaction_epoch()
@@ -3762,22 +4019,34 @@ mod tests {
         let (upload_id, parts) =
             stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await;
 
-        temp_env::async_with_vars(
+        let epochs = temp_env::async_with_vars(
             [
                 (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
                 (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+                (ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true")),
             ],
             async {
+                let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
                 set_disks
                     .clone()
                     .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
                     .await
                     .expect("fenced multipart completion should commit with a live proof");
+
+                tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
+                    .await
+                    .expect("multipart completion should leave one rename tail in flight after quorum ACK");
+                let disks = disk_stores.clone();
+                let mut epochs = tokio::spawn(async move { object_transaction_epochs(&disks, bucket, object).await });
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(100), &mut epochs).await.is_err(),
+                    "epoch read-back should wait for the lagging rename tail"
+                );
+                rename_barrier.release();
+                epochs.await.expect("epoch read-back should finish after the rename tail")
             },
         )
         .await;
-
-        let epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
         let first = epochs[0].expect("fenced multipart completion should persist an epoch");
         assert!(!first.is_nil());
         assert!(epochs.into_iter().all(|epoch| epoch == Some(first)));

crates/ecstore/src/store/init.rs

@@ -99,6 +99,8 @@ fn preflight_startup_rpc_secret_with(
 const LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES: usize = 6;
 const LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(60 * 3);
 const LOCAL_DECOMMISSION_RESUME_RETRY_DELAY: Duration = Duration::from_secs(30);
+const REBALANCE_INITIAL_RESUME_DELAY: Duration = Duration::from_secs(10);
+const REBALANCE_RESUME_RETRY_DELAY: Duration = Duration::from_secs(10);
 
 fn should_retry_local_decommission_resume(err: &Error, attempt: usize) -> bool {
     matches!(err, Error::ConfigNotFound) && attempt < LOCAL_DECOMMISSION_RESUME_MAX_CONFIG_RETRIES
@@ -108,8 +110,12 @@ fn should_retry_format_load(err: &Error) -> bool {
     !matches!(err, Error::CorruptedFormat)
 }
 
-fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_meta_loaded: bool) -> bool {
-    rebalance_meta_loaded && !decommission_running
+fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_resume_required: bool) -> bool {
+    rebalance_resume_required && !decommission_running
+}
+
+fn should_defer_rebalance_auto_start(distributed: bool, fleet_proof_available: bool) -> bool {
+    distributed && !fleet_proof_available
 }
 
 fn should_schedule_local_decommission_resume(
@@ -127,6 +133,17 @@ async fn wait_for_local_decommission_resume_delay(rx: &CancellationToken, delay:
     }
 }
 
+async fn wait_for_rebalance_resume_delay(rx: &CancellationToken, delay: Duration) -> bool {
+    tokio::select! {
+        _ = rx.cancelled() => false,
+        _ = tokio::time::sleep(delay) => true,
+    }
+}
+
+async fn wait_for_rebalance_resume_retry(rx: &CancellationToken) -> bool {
+    wait_for_rebalance_resume_delay(rx, REBALANCE_RESUME_RETRY_DELAY).await
+}
+
 fn resolve_store_init_stage_result(result: Result<()>, stage: &str) -> Result<()> {
     result.map_err(|err| Error::other(format!("store init failed during {stage}: {err}")))
 }
@@ -283,6 +300,71 @@ async fn resume_local_decommission_after_init(store: Arc<ECStore>, rx: Cancellat
     }
 }
 
+async fn resume_rebalance_after_init(store: Arc<ECStore>, rx: CancellationToken) {
+    if !wait_for_rebalance_resume_delay(&rx, REBALANCE_INITIAL_RESUME_DELAY).await {
+        return;
+    }
+
+    loop {
+        if rx.is_cancelled() {
+            return;
+        }
+
+        let resume_required = store
+            .rebalance_meta
+            .read()
+            .await
+            .as_ref()
+            .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation);
+        if !resume_required {
+            return;
+        }
+
+        if should_defer_rebalance_auto_start(
+            store.ctx.is_dist_erasure().await,
+            crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().is_some(),
+        ) {
+            if !wait_for_rebalance_resume_retry(&rx).await {
+                return;
+            }
+            continue;
+        }
+
+        match store.start_rebalance().await {
+            Ok(()) => return,
+            Err(err) if crate::core::pools::is_pool_activation_fleet_proof_error(&err) => {
+                warn!(
+                    event = EVENT_ECSTORE_INIT_STATUS,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_STORE_INIT,
+                    stage = "start_rebalance",
+                    state = "retrying",
+                    reason = "fleet_capability_proof_unavailable",
+                    error = %err,
+                    retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(),
+                    "Retrying deferred rebalance auto-start"
+                );
+                if !wait_for_rebalance_resume_retry(&rx).await {
+                    return;
+                }
+            }
+            Err(err) => {
+                error!(
+                    event = EVENT_ECSTORE_INIT_STATUS,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_STORE_INIT,
+                    stage = "start_rebalance",
+                    state = "failed",
+                    reason = "deferred_resume_failed",
+                    error = %err,
+                    "Failed to resume rebalance after store initialization"
+                );
+                return;
+            }
+        }
+    }
+}
+
 impl ECStore {
     /// Validate topology and process storage-class overrides before any disk is opened.
     pub fn validate_startup_storage_class(endpoint_pools: &EndpointServerPools) -> Result<()> {
@@ -574,12 +656,49 @@ impl ECStore {
         }
 
         resolve_store_init_stage_result(self.load_rebalance_meta().await, "load_rebalance_meta")?;
-        let rebalance_meta_loaded = self.rebalance_meta.read().await.is_some();
+        let rebalance_resume_required = {
+            let rebalance_meta = self.rebalance_meta.read().await;
+            rebalance_meta
+                .as_ref()
+                .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation)
+        };
         let decommission_running =
             pool_meta_has_active_decommission(&installed_pool_meta) || self.is_decommission_running().await;
-        if should_auto_start_rebalance_after_init(decommission_running, rebalance_meta_loaded) {
-            resolve_store_init_stage_result(self.start_rebalance().await, "start_rebalance")?;
-        } else if decommission_running && rebalance_meta_loaded {
+        let distributed = self.ctx.is_dist_erasure().await;
+        let fleet_proof_available = crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().is_some();
+        let mut rebalance_auto_start_deferred = false;
+        if should_auto_start_rebalance_after_init(decommission_running, rebalance_resume_required) {
+            if should_defer_rebalance_auto_start(distributed, fleet_proof_available) {
+                rebalance_auto_start_deferred = true;
+                warn!(
+                    event = EVENT_ECSTORE_INIT_STATUS,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_STORE_INIT,
+                    stage = "start_rebalance",
+                    state = "deferred",
+                    reason = "fleet_capability_proof_unavailable",
+                    retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(),
+                    "Deferred rebalance auto-start until a live fleet capability proof is available"
+                );
+            } else if let Err(err) = self.start_rebalance().await {
+                if crate::core::pools::is_pool_activation_fleet_proof_error(&err) {
+                    rebalance_auto_start_deferred = true;
+                    warn!(
+                        event = EVENT_ECSTORE_INIT_STATUS,
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_STORE_INIT,
+                        stage = "start_rebalance",
+                        state = "deferred",
+                        reason = "fleet_capability_proof_changed",
+                        error = %err,
+                        retry_delay_secs = REBALANCE_RESUME_RETRY_DELAY.as_secs(),
+                        "Deferred rebalance auto-start after the fleet capability proof changed"
+                    );
+                } else {
+                    return resolve_store_init_stage_result(Err(err), "start_rebalance");
+                }
+            }
+        } else if decommission_running && rebalance_resume_required {
             warn!(
                 event = EVENT_ECSTORE_INIT_STATUS,
                 component = LOG_COMPONENT_ECSTORE,
@@ -616,12 +735,13 @@ impl ECStore {
             .is_ok();
         if should_schedule_local_decommission_resume(&local_pool_indices, pool_meta_replica_state, pool_meta_write_safe) {
             let store = self.clone();
+            let decommission_rx = rx.clone();
 
             tokio::spawn(async move {
-                if !wait_for_local_decommission_resume_delay(&rx, LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY).await {
+                if !wait_for_local_decommission_resume_delay(&decommission_rx, LOCAL_DECOMMISSION_INITIAL_RESUME_DELAY).await {
                     return;
                 }
-                resume_local_decommission_after_init(store, rx, local_pool_indices).await;
+                resume_local_decommission_after_init(store, decommission_rx, local_pool_indices).await;
             });
         } else if !local_pool_indices.is_empty() {
             error!(
@@ -648,6 +768,11 @@ impl ECStore {
             info!("TierConfigMgr init error: {}", err);
         }
 
+        if rebalance_auto_start_deferred {
+            let store = self.clone();
+            tokio::spawn(resume_rebalance_after_init(store, rx));
+        }
+
         Ok(())
     }
 
@@ -665,7 +790,8 @@ mod tests {
         load_pool_meta_for_startup, persist_pool_meta_for_startup_if_safe, pool_first_endpoint_is_local,
         pool_meta_has_active_decommission, preflight_startup_rpc_secret_with, resolve_startup_pool_defaults_with,
         resolve_store_init_stage_result, save_validated_pool_meta_for_startup, should_auto_start_rebalance_after_init,
-        should_retry_format_load, should_retry_local_decommission_resume, wait_for_local_decommission_resume_delay,
+        should_defer_rebalance_auto_start, should_retry_format_load, should_retry_local_decommission_resume,
+        wait_for_local_decommission_resume_delay,
     };
     #[cfg(feature = "test-util")]
     use crate::disk::DiskAPI;
@@ -1450,7 +1576,7 @@ mod tests {
     }
 
     #[test]
-    fn test_should_auto_start_rebalance_after_init_allows_loaded_rebalance_without_decommission() {
+    fn test_should_auto_start_rebalance_after_init_allows_active_rebalance_without_decommission() {
         assert!(should_auto_start_rebalance_after_init(false, true));
     }
 
@@ -1460,10 +1586,17 @@ mod tests {
     }
 
     #[test]
-    fn test_should_auto_start_rebalance_after_init_rejects_missing_rebalance_meta() {
+    fn test_should_auto_start_rebalance_after_init_rejects_terminal_or_missing_rebalance() {
         assert!(!should_auto_start_rebalance_after_init(false, false));
     }
 
+    #[test]
+    fn test_should_defer_rebalance_auto_start_only_without_distributed_fleet_proof() {
+        assert!(should_defer_rebalance_auto_start(true, false));
+        assert!(!should_defer_rebalance_auto_start(true, true));
+        assert!(!should_defer_rebalance_auto_start(false, false));
+    }
+
     #[test]
     fn test_store_init_recovery_skips_rebalance_when_decommission_metadata_is_active() {
         let pool_meta = init_test_pool_meta(Some(PoolDecommissionInfo {
@@ -1473,22 +1606,69 @@ mod tests {
             canceled: false,
             ..Default::default()
         }));
-        let rebalance_meta = Some(RebalanceMeta::default());
+        let rebalance_meta = Some(RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        });
 
         assert!(!should_auto_start_rebalance_after_init(
             pool_meta_has_active_decommission(&pool_meta),
-            rebalance_meta.is_some()
+            rebalance_meta
+                .as_ref()
+                .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation)
         ));
     }
 
     #[test]
-    fn test_store_init_recovery_allows_rebalance_when_only_rebalance_metadata_exists() {
+    fn test_store_init_recovery_allows_active_rebalance_without_decommission() {
         let pool_meta = init_test_pool_meta(None);
-        let rebalance_meta = Some(RebalanceMeta::default());
+        let rebalance_meta = Some(RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Started,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        });
 
         assert!(should_auto_start_rebalance_after_init(
             pool_meta_has_active_decommission(&pool_meta),
-            rebalance_meta.is_some()
+            rebalance_meta
+                .as_ref()
+                .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation)
+        ));
+    }
+
+    #[test]
+    fn test_store_init_recovery_skips_completed_rebalance_metadata() {
+        let pool_meta = init_test_pool_meta(None);
+        let rebalance_meta = Some(RebalanceMeta {
+            pool_stats: vec![RebalanceStats {
+                participating: true,
+                info: RebalanceInfo {
+                    status: RebalStatus::Completed,
+                    ..Default::default()
+                },
+                ..Default::default()
+            }],
+            ..Default::default()
+        });
+
+        assert!(!should_auto_start_rebalance_after_init(
+            pool_meta_has_active_decommission(&pool_meta),
+            rebalance_meta
+                .as_ref()
+                .is_some_and(crate::services::rebalance::rebalance_requires_worker_activation)
         ));
     }
 

crates/obs/Cargo.toml

@@ -74,7 +74,7 @@ hotpath-cpu = [
 # Tokio runtime-level telemetry. Requires a `--cfg tokio_unstable` build; the
 # build script fails the compile when that flag is missing. Off by default so
 # ordinary builds neither pay for nor depend on Tokio's unstable API.
-dial9 = ["dep:dial9-tokio-telemetry"]
+dial9 = ["dep:dial9-tokio-telemetry", "dial9-tokio-telemetry/process-resource"]
 #
 # NOTE: there is deliberately no `dial9-taskdump` feature. dial9 only captures a
 # task dump for futures it wrapped itself, i.e. those spawned via

crates/obs/src/telemetry/dial9/config.rs

@@ -76,7 +76,7 @@ pub struct Dial9Config {
     /// Directory where trace files are written
     pub output_dir: String,
 
-    /// Prefix for trace file names
+    /// Trace family name under the output directory
     pub file_prefix: String,
 
     /// Maximum size of each trace file in bytes
@@ -158,7 +158,7 @@ impl Dial9Config {
         }
     }
 
-    /// Get the base path for trace files.
+    /// Get the trace family directory for rotating trace segments.
     pub fn base_path(&self) -> PathBuf {
         PathBuf::from(&self.output_dir).join(&self.file_prefix)
     }

crates/obs/src/telemetry/dial9/enabled.rs

@@ -23,15 +23,22 @@ use super::config::Dial9Config;
 use super::state::{dial9_runtime_state, measure_disk_usage_bytes};
 use super::{EVENT_DIAL9_STATE, LOG_COMPONENT_OBS, LOG_SUBSYSTEM_DIAL9};
 use crate::TelemetryError;
-use dial9_tokio_telemetry::telemetry::{ProcessResourceUsageConfig, RotatingWriter, TracedRuntime};
+use dial9_tokio_telemetry::telemetry::{
+    Dial9Handle, Dial9HandleTokioExt, DiskBuffer, ProcessResourceUsageConfig, RecorderPerfExt, TokioAttachOptions, recorder,
+};
 use std::time::Duration;
 use tracing::{info, warn};
 
-pub use dial9_tokio_telemetry::telemetry::TelemetryGuard;
+pub type TelemetryGuard = Dial9Handle;
+
+type ShutdownRecorder = Box<dyn FnOnce() + Send + 'static>;
 
 /// How often the background refresher restates trace-file disk usage.
 const DISK_USAGE_REFRESH_INTERVAL: Duration = Duration::from_secs(60);
 
+/// Maximum time spent flushing the recorder during graceful shutdown.
+const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(5);
+
 /// Name recorded in segment metadata so the trace viewer can label workers.
 const RUNTIME_NAME: &str = "rustfs-worker";
 
@@ -43,13 +50,14 @@ const RUNTIME_NAME: &str = "rustfs-worker";
 /// are lost.
 pub struct Dial9SessionGuard {
     guard: TelemetryGuard,
+    shutdown: Option<ShutdownRecorder>,
     config: Dial9Config,
 }
 
 impl Dial9SessionGuard {
     /// Whether the underlying telemetry session is recording.
     pub fn is_active(&self) -> bool {
-        self.guard.is_enabled()
+        self.guard.is_enabled() && self.guard.is_connected() && !self.guard.is_stopped()
     }
 }
 
@@ -72,8 +80,10 @@ impl Drop for Dial9SessionGuard {
             state = "flushed",
             "dial9 state changed"
         );
-        // `TelemetryGuard`'s own `Drop` flushes buffered events and seals the
-        // active segment; it runs immediately after this body.
+
+        if let Some(shutdown) = self.shutdown.take() {
+            shutdown();
+        }
     }
 }
 
@@ -97,53 +107,58 @@ pub fn build_traced_runtime(
         TelemetryError::Io(format!("Failed to create dial9 output directory '{}': {e}", config.output_dir))
     })?;
 
-    let writer = RotatingWriter::new(config.base_path(), config.max_file_size, config.total_disk_budget()).map_err(|e| {
-        dial9_runtime_state().record_runtime_error(&config);
-        TelemetryError::Io(format!("Failed to create dial9 RotatingWriter: {e}"))
-    })?;
+    let writer = DiskBuffer::builder()
+        .base_path(config.base_path())
+        .max_file_size(config.max_file_size)
+        .max_total_size(config.total_disk_budget())
+        .build()
+        .map_err(|e| {
+            dial9_runtime_state().record_runtime_error(&config);
+            TelemetryError::Io(format!("Failed to create dial9 DiskBuffer: {e}"))
+        })?;
 
-    // `with_trace_path` transitions the builder into the state that spawns the
-    // background worker, which drives the segment pipeline.
-    let traced = TracedRuntime::builder()
-        .with_trace_path(&config.output_dir)
-        .with_task_tracking(true)
-        .with_runtime_name(RUNTIME_NAME)
-        .with_process_resource_usage(ProcessResourceUsageConfig::default());
+    let recorder = recorder(writer)
+        .with_process_resource_usage(ProcessResourceUsageConfig::default())
+        .build();
+    let guard = recorder.handle().clone();
+    let shutdown: ShutdownRecorder = Box::new(move || recorder.graceful_shutdown(SHUTDOWN_TIMEOUT));
 
-    // `build_and_start` rather than `build`: `build` returns a live guard that
-    // never records, writing segments that contain only a header.
-    //
-    // No `with_task_dumps` here. dial9 captures a task dump only for futures it
+    let attached = guard
+        .attach_tokio_runtime(
+            builder,
+            TokioAttachOptions::builder()
+                .runtime_name(RUNTIME_NAME)
+                .task_tracking_enabled(true)
+                .build(),
+        )
+        .map(|runtime| (runtime, guard, shutdown));
+
+    // No task dumps here. dial9 captures a task dump only for futures it
     // wrapped itself, i.e. those spawned via `dial9_tokio_telemetry::spawn`;
     // `tokio::spawn` gets no wrapper. RustFS spawns with `tokio::spawn`
-    // throughout, so calling `with_task_dumps` records nothing. Measured on an
+    // throughout, so enabling task dumps records nothing. Measured on an
     // identical workload: 0 dumps via `tokio::spawn`, 14709 via `dial9::spawn`.
     // See rustfs/backlog#1157 (D9-16) and dial9-rs/dial9#477.
     //
     // No `with_s3_uploader` here: dial9's `worker-s3` feature carries a
     // vulnerable TLS stack. See the note in `crates/obs/Cargo.toml`.
-    finish_traced_runtime(traced.build_and_start(builder, writer), config)
+    finish_traced_runtime(attached, config)
 }
 
 /// Publish the outcome of a traced-runtime build and start the background
 /// disk-usage refresher.
 fn finish_traced_runtime(
-    started: std::io::Result<(tokio::runtime::Runtime, TelemetryGuard)>,
+    started: std::io::Result<(tokio::runtime::Runtime, TelemetryGuard, ShutdownRecorder)>,
     config: Dial9Config,
 ) -> Result<(tokio::runtime::Runtime, Dial9SessionGuard), TelemetryError> {
-    let (runtime, guard) = started.map_err(|e| {
+    let (runtime, guard, shutdown) = started.map_err(|e| {
         dial9_runtime_state().record_runtime_error(&config);
-        TelemetryError::Io(format!("Failed to build dial9 TracedRuntime: {e}"))
+        TelemetryError::Io(format!("Failed to attach dial9 runtime telemetry: {e}"))
     })?;
 
-    // `is_enabled` distinguishes a live guard from the inert one a lenient
-    // config produces after a build failure. It does NOT mean recording has
-    // started — a guard from `build` (rather than `build_and_start`) reports
-    // `true` while writing segments that contain only a header. Recording is
-    // guaranteed by the `build_and_start` call above, not by this check.
     if !guard.is_enabled() {
         dial9_runtime_state().record_runtime_error(&config);
-        return Err(TelemetryError::Io("dial9 TracedRuntime built with telemetry disabled".to_string()));
+        return Err(TelemetryError::Io("dial9 runtime telemetry attached with recording disabled".to_string()));
     }
 
     dial9_runtime_state().record_runtime_started(&config);
@@ -160,7 +175,14 @@ fn finish_traced_runtime(
         "dial9 state changed"
     );
 
-    Ok((runtime, Dial9SessionGuard { guard, config }))
+    Ok((
+        runtime,
+        Dial9SessionGuard {
+            guard,
+            shutdown: Some(shutdown),
+            config,
+        },
+    ))
 }
 
 /// Periodically restate trace-file disk usage so the metrics collector can read

crates/obs/src/telemetry/dial9/mod.rs

@@ -49,13 +49,13 @@
 //!
 //! # Known observability gap
 //!
-//! `dial9`'s `RotatingWriter` stops accepting writes (its internal `Finished`
-//! state) when the output directory disappears or a segment cannot be sealed,
-//! and it exposes no way to observe that from outside. `TelemetryGuard::is_enabled`
-//! reports how the session was *built*, not whether it is still writing. There
-//! is therefore no `writer_healthy` metric: it could only ever be hard-coded to
-//! `1`. Watch `rustfs_dial9_disk_usage_bytes` — a session that is recording but
-//! whose disk usage stops growing has most likely hit this state.
+//! `dial9`'s `DiskBuffer` stops accepting writes when the output directory
+//! disappears or a segment cannot be sealed, and it exposes no way to observe
+//! that from outside. `Dial9Handle::is_enabled` reports whether the recorder is
+//! connected and unpaused, not whether the disk writer is still making progress.
+//! There is therefore no `writer_healthy` metric: it could only ever be
+//! hard-coded to `1`. Watch `rustfs_dial9_disk_usage_bytes` — a session that is
+//! recording but whose disk usage stops growing has most likely hit this state.
 //! Reported upstream as dial9-rs/dial9#658.
 
 mod config;

crates/obs/src/telemetry/dial9/state.rs

@@ -27,6 +27,9 @@ use std::sync::OnceLock;
 use std::sync::RwLock;
 use std::sync::atomic::{AtomicU64, Ordering};
 
+/// Segment filename stem used by `dial9` rotating disk buffers.
+const DIAL9_SEGMENT_STEM: &str = "trace";
+
 /// Point-in-time view of dial9 runtime state.
 #[derive(Debug, Clone, Default)]
 pub(crate) struct Dial9RuntimeSnapshot {
@@ -66,8 +69,8 @@ impl Dial9RuntimeState {
 
     pub(super) fn record_config(&self, config: &Dial9Config) {
         *self.trace_dir.write().expect("dial9 trace_dir lock should not be poisoned") = Some(TraceLocation {
-            output_dir: PathBuf::from(&config.output_dir),
-            file_prefix: config.file_prefix.clone(),
+            output_dir: config.base_path(),
+            file_prefix: DIAL9_SEGMENT_STEM.to_string(),
         });
         if !config.enabled {
             self.active_sessions.store(0, Ordering::Relaxed);
@@ -168,11 +171,11 @@ mod tests {
     #[test]
     fn measure_disk_usage_sums_only_matching_prefix() {
         let dir = tempdir().expect("create temp dir");
-        std::fs::write(dir.path().join("rustfs-tokio.0.bin"), vec![0_u8; 128]).expect("write segment");
-        std::fs::write(dir.path().join("rustfs-tokio.1.bin"), vec![0_u8; 64]).expect("write segment");
+        std::fs::write(dir.path().join("trace.0.bin"), vec![0_u8; 128]).expect("write segment");
+        std::fs::write(dir.path().join("trace.1.bin"), vec![0_u8; 64]).expect("write segment");
         std::fs::write(dir.path().join("unrelated.log"), vec![0_u8; 4096]).expect("write unrelated");
 
-        assert_eq!(measure_disk_usage_bytes(dir.path(), "rustfs-tokio"), 192);
+        assert_eq!(measure_disk_usage_bytes(dir.path(), DIAL9_SEGMENT_STEM), 192);
     }
 
     #[test]

crates/s3-client/Cargo.toml

@@ -54,7 +54,6 @@ rustls-pki-types.workspace = true
 s3s = { workspace = true, features = ["minio"] }
 serde = { workspace = true, features = ["derive"] }
 serde_json = { workspace = true, features = ["raw_value"] }
-sha1 = { workspace = true }
 sha2 = { workspace = true }
 thiserror.workspace = true
 time = { workspace = true, features = ["parsing", "formatting", "macros", "serde"] }

crates/storage-api/src/error.rs

@@ -81,6 +81,7 @@ pub enum StorageErrorCode {
     InsufficientWriteQuorum,
     PreconditionFailed,
     EntityTooSmall,
+    EntityTooLarge,
     InvalidRangeSpec,
     NotModified,
     InvalidPartNumber,
@@ -169,6 +170,7 @@ impl StorageErrorCode {
             Self::InsufficientWriteQuorum => 0x3A,
             Self::PreconditionFailed => 0x3B,
             Self::EntityTooSmall => 0x3C,
+            Self::EntityTooLarge => 0x56,
             Self::InvalidRangeSpec => 0x3D,
             Self::NotModified => 0x3E,
             Self::InvalidPartNumber => 0x3F,
@@ -257,6 +259,7 @@ impl StorageErrorCode {
             0x3A => Some(Self::InsufficientWriteQuorum),
             0x3B => Some(Self::PreconditionFailed),
             0x3C => Some(Self::EntityTooSmall),
+            0x56 => Some(Self::EntityTooLarge),
             0x3D => Some(Self::InvalidRangeSpec),
             0x3E => Some(Self::NotModified),
             0x3F => Some(Self::InvalidPartNumber),
@@ -350,6 +353,7 @@ mod tests {
         (StorageErrorCode::InsufficientWriteQuorum, 0x3A),
         (StorageErrorCode::PreconditionFailed, 0x3B),
         (StorageErrorCode::EntityTooSmall, 0x3C),
+        (StorageErrorCode::EntityTooLarge, 0x56),
         (StorageErrorCode::InvalidRangeSpec, 0x3D),
         (StorageErrorCode::NotModified, 0x3E),
         (StorageErrorCode::InvalidPartNumber, 0x3F),

crates/utils/src/http/metadata_compat.rs

@@ -44,6 +44,8 @@ pub const SUFFIX_COMPRESSION: &str = "compression";
 pub const SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT: &str = "replication-preserve-ciphertext";
 pub const SUFFIX_COMPRESSION_SIZE: &str = "compression-size";
 pub const SUFFIX_ACTUAL_SIZE: &str = "actual-size";
+/// Maximum logical object size for a capability-bound multipart upload.
+pub const SUFFIX_MAX_TOTAL_OBJECT_SIZE: &str = "max-total-object-size";
 pub const SUFFIX_ACTUAL_OBJECT_SIZE: &str = "actual-object-size";
 /// Used by replication; key stored with capital A
 pub const SUFFIX_ACTUAL_OBJECT_SIZE_CAP: &str = "Actual-Object-Size";

docs/operations/kms-per-key-authorization.md

@@ -81,7 +81,7 @@ Scope and exemptions:
 
 - **SSE-KMS only.** SSE-S3 wraps its data key with a server-owned key the caller never names, and SSE-C never reaches KMS; both are exempt, matching AWS.
 - **The resolved key**, not the header. A bucket default encryption rule naming a KMS key is authorized the same way an explicit `x-amz-server-side-encryption-aws-kms-key-id` header is.
-- **Anonymous requests are exempt.** They have no identity policy to evaluate, and denying them would break public buckets holding SSE-KMS objects. They remain governed by bucket policy.
+- **Anonymous requests are denied.** An anonymous caller has no identity policy and therefore holds no `kms` grants, so under enforcement every anonymous read or write of an SSE-KMS object fails with `AccessDenied` — even when a bucket policy makes the bucket public. This matches AWS, where anonymous requests cannot use SSE-KMS objects at all, and it keeps the per-key gate meaningful: were anonymous requests exempt, any denied identity could bypass the gate on a public bucket by simply dropping its credentials. **A public bucket serving SSE-KMS objects is incompatible with enforcement** — serve public content unencrypted or under SSE-S3 instead. With enforcement off (the default), anonymous access to SSE-KMS objects remains governed by bucket policy alone. The server warns once per process when it first denies an anonymous request; per-request denials appear on audit entries (`kmsOutcome=failure`, `kmsErrorClass=access_denied`, empty requester identity) and at debug level.
 - **Internal work is exempt.** Replication, lifecycle transitions, healing and the scanner run as the system principal.
 - **Authorization runs before key state is checked**, so a denial cannot be used to probe whether a key exists, is disabled, or is pending deletion. The response is always `AccessDenied`.
 - **Multipart uploads are authorized at create time**, where the session data key is generated. Part uploads and completion reuse that envelope and are not re-authorized against the destination key.
@@ -89,13 +89,13 @@ Scope and exemptions:
 
 ## Migration
 
-Data-path enforcement is **off by default in this release** because it changes the outcome of requests that succeed today: an identity holding only `s3:PutObject` can currently encrypt under any key. Turning it on without preparing policies will produce `AccessDenied` on working workloads.
+Data-path enforcement is **off by default** because it changes the outcome of requests that succeed today: an identity holding only `s3:PutObject` can currently encrypt under any key. Turning it on without preparing policies will produce `AccessDenied` on working workloads.
 
 ```bash
 RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY=true
 ```
 
-The server logs the configured mode once at startup, and warns while enforcement is off. A later release defaults it to enabled.
+The server logs the configured mode once at startup, and warns while enforcement is off. Enforcement stays opt-in: there is no roadmap to flip the default.
 
 Recommended sequence:
 

docs/operations/presigned-multipart-size-limit.md

@@ -0,0 +1,55 @@
+# Presigned multipart total-size limit
+
+RustFS V2 supports an optional capability on a signed or SigV4-presigned
+`CreateMultipartUpload` request:
+
+```text
+x-rustfs-max-total-object-size=<unsigned 64-bit integer>
+```
+
+The backend must include the parameter before calculating the SigV4
+signature. It is part of the canonical query and cannot be added, removed, or
+changed by the browser. RustFS stores the verified limit in the multipart
+upload session and applies it to every `UploadPart` and to
+`CompleteMultipartUpload`.
+
+Backend pseudocode (the custom query must be present before signing):
+
+```text
+uri = "/photos/archive.zip?uploads"
+uri += "&x-rustfs-max-total-object-size=104857600"
+presigned_url = sigv4_presign("POST", uri, credentials)
+# Return presigned_url to the browser. Never append the parameter afterwards.
+```
+
+The resulting flow is:
+
+1. The backend signs `CreateMultipartUpload?...&x-rustfs-max-total-object-size=104857600`.
+2. RustFS verifies the SigV4 request and persists the limit with the upload ID.
+3. The browser uploads parts using the returned upload ID.
+4. RustFS rejects a part whose declared logical size would exceed the remaining
+   budget and rejects completion if the server-side part metadata exceeds the
+   limit.
+
+The limit is measured in logical object bytes (`actual_size`), not erasure,
+encryption, or compression bytes. Replacing an existing part uses replacement
+semantics: the old part size is removed before the new part size is admitted.
+Unknown-length parts are rejected for capped sessions rather than buffered
+without a bound. Capped parts are admitted under an upload-wide write lock
+before temporary shards are created and use a per-upload staging permit to
+bound local in-flight data. The distributed lock is released while the body is
+read and reacquired for the final check/rename, so Complete and Abort are not
+blocked behind a slow upload. The normal request-body stall timeout releases
+the staging permit when a client stops sending.
+
+The parameter is accepted only on `CreateMultipartUpload`. Supplying it on
+`UploadPart`, `CompleteMultipartUpload`, `AbortMultipartUpload`, listing, or
+copy operations returns `InvalidRequest`; those requests use the persisted
+session state. A multipart upload created without this parameter remains
+unlimited for backward compatibility. The V1 single-request capability
+(`x-rustfs-max-content-length`) is independent and is not a multipart limit.
+
+Because enforcement happens in the multipart data plane, every node that may
+receive requests for a capped upload must run the V2 implementation. During a
+rolling upgrade, route capped uploads only to upgraded nodes; older nodes treat
+the internal metadata as unknown and cannot enforce the limit.

docs/operations/presigned-put-size-limit.md

@@ -0,0 +1,35 @@
+# Presigned PutObject size limit
+
+RustFS V1 supports an optional, RustFS-specific capability on a SigV4
+presigned `PutObject` URL:
+
+```text
+x-rustfs-max-content-length=<unsigned 64-bit integer>
+```
+
+The backend that creates the URL must add this query parameter to the request
+URI before calculating the SigV4 presign. It is part of the canonical query;
+adding, removing, or changing it after signing invalidates the signature. A
+browser can then upload with a plain `PUT` and does not need a custom size
+header.
+
+RustFS validates the capability after SigV4 authentication and enforces it on
+the decoded request body. A declared `Content-Length` above the limit is
+rejected before storage. If the body produces more bytes than the limit while
+streaming, RustFS returns `EntityTooLarge` and does not publish the object.
+
+The V1 contract is deliberately narrow:
+
+- The parameter is accepted only on a SigV4 presigned `PutObject` request.
+- Duplicate, case-variant, malformed, negative, or overflowing values return
+  `InvalidRequest`.
+- Requests without the parameter, including ordinary authenticated or
+  anonymous `PUT`, keep the existing behavior.
+- The parameter on `CopyObject`, multipart, `GET`, `HEAD`, `DELETE`, bucket, or
+  other operations returns `InvalidRequest`.
+- Unknown-length and SigV4 streaming-chunked uploads remain unsupported by the
+  existing PutObject admission contract and are not enabled by this feature.
+
+This capability is per request; it is not a cumulative multipart-upload cap.
+Multipart session limits are planned for V2 under a separate query/API
+contract.

docs/testing/ci-gates.md

@@ -119,6 +119,11 @@ pull-request gate.
 Use an exact preview tag for end-to-end release rehearsal. Manual dispatches
 are backfill/debug paths and do not prove the automatic `workflow_run` chain.
 
+A preview Release is internal validation state, not a deliverable: after the
+final tag's release is published, `cleanup-preview-releases` deletes every
+`<target>-preview.<N>` Release for that target. The tags themselves are kept, so
+the validated commit stays traceable.
+
 ## Evidence requirements
 
 A green check is useful only when it proves the intended behavior ran:

rustfs/src/app/multipart_usecase.rs

@@ -67,6 +67,7 @@ use super::storage_api::multipart_usecase::sse::{
 use super::storage_api::multipart_usecase::{
     StorageObjectInfo as ObjectInfo, StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader,
 };
+use crate::app::object::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
 use crate::app::object_data_cache::{
     ObjectDataCacheAdapter, invalidate_object_data_cache_after_complete_multipart_success,
     invalidate_object_data_cache_before_mutation,
@@ -78,6 +79,11 @@ use crate::app::object_usecase::{
 use crate::app::runtime_sources::{
     AppContext, current_app_context, current_object_data_cache_for_context, current_object_store_handle_for_context,
 };
+use crate::auth::{
+    VerifiedPresignedRequest, VerifiedSigV4Request, parse_presigned_multipart_max_total_object_size,
+    reject_presigned_multipart_max_total_object_size_for_other_operation,
+    reject_presigned_put_max_content_length_for_other_operation,
+};
 use crate::capacity::record_capacity_write;
 use crate::error::ApiError;
 use crate::table_catalog;
@@ -91,8 +97,9 @@ use rustfs_utils::CompressionAlgorithm;
 #[cfg(test)]
 use rustfs_utils::http::insert_header;
 use rustfs_utils::http::{
-    SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP,
-    SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_header, get_source_scheme,
+    SUFFIX_MAX_TOTAL_OBJECT_SIZE, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, SUFFIX_REPLICATION_STATUS,
+    SUFFIX_REPLICATION_TIMESTAMP, SUFFIX_SOURCE_REPLICATION_REQUEST, contains_key_str, get_consistent_str, get_header,
+    get_source_scheme,
     headers::{AMZ_DECODED_CONTENT_LENGTH, AMZ_OBJECT_TAGGING, AMZ_STORAGE_CLASS},
     insert_str,
 };
@@ -107,6 +114,7 @@ use s3s::{S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error};
 use std::collections::{HashMap, HashSet};
 use std::str::FromStr;
 use std::sync::Arc;
+use std::time::Duration;
 use tokio::sync::RwLock;
 use tokio_util::io::StreamReader;
 use tracing::{instrument, warn};
@@ -225,6 +233,22 @@ fn create_multipart_upload_metadata(
     metadata
 }
 
+fn multipart_max_total_object_size(metadata: &HashMap<String, String>) -> S3Result<Option<u64>> {
+    if !contains_key_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE) {
+        return Ok(None);
+    }
+
+    let value = get_consistent_str(metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE).ok_or_else(|| {
+        S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            "multipart size capability metadata is missing or inconsistent".to_string(),
+        )
+    })?;
+    value.parse::<u64>().map(Some).map_err(|_| {
+        S3Error::with_message(S3ErrorCode::InvalidRequest, "multipart size capability metadata is invalid".to_string())
+    })
+}
+
 /// A multipart session advertises disk compression only when the staged-rollout
 /// switch (`RUSTFS_COMPRESSION_MULTIPART_ENABLED`) is on, the object key/headers
 /// qualify, AND the session is not an SSE-C ciphertext-passthrough replication
@@ -397,6 +421,16 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<AbortMultipartUploadInput>,
     ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         record_s3_op(S3Operation::AbortMultipartUpload);
         let mut opts = ObjectOptions::default();
         apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
@@ -438,6 +472,16 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<CompleteMultipartUploadInput>,
     ) -> S3Result<S3Response<CompleteMultipartUploadOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         let mut helper = OperationHelper::new(
             &req,
             EventName::ObjectCreatedCompleteMultipartUpload,
@@ -741,6 +785,16 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<CreateMultipartUploadInput>,
     ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        let multipart_max_total_object_size = parse_presigned_multipart_max_total_object_size(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         let helper =
             OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
                 .suppress_event();
@@ -791,6 +845,9 @@ impl DefaultMultipartUsecase {
         )?;
 
         let mut metadata = create_multipart_upload_metadata(input_metadata, &req.headers, tagging, storage_class.as_ref());
+        if let Some(limit) = multipart_max_total_object_size {
+            insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, limit.to_string());
+        }
 
         let has_explicit_object_lock_retention = object_lock_mode.is_some()
             || object_lock_retain_until_date.is_some()
@@ -962,6 +1019,16 @@ impl DefaultMultipartUsecase {
     #[instrument(level = "debug", skip(self, req))]
     #[hotpath::measure(impl_type = "MultipartUsecase")]
     pub async fn execute_upload_part(&self, req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         let mut opts = ObjectOptions::default();
         apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
         let input = req.input;
@@ -985,6 +1052,40 @@ impl DefaultMultipartUsecase {
 
         let mut size = resolve_upload_part_size(&req.headers, content_length)?;
         let mut body_stream = body.ok_or_else(|| s3_error!(IncompleteBody))?;
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        let fi = store
+            .get_multipart_info(&bucket, &key, &upload_id, &opts)
+            .await
+            .map_err(ApiError::from)?;
+        let max_total_object_size = multipart_max_total_object_size(&fi.user_defined)?;
+        if max_total_object_size.is_some() && size.is_some_and(|size| size < 0) {
+            return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
+        }
+        if max_total_object_size.is_some() && size.is_none() {
+            return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
+        }
+        if let (Some(limit), Some(size)) = (max_total_object_size, size)
+            && u64::try_from(size).is_ok_and(|size| size > limit)
+        {
+            return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
+        }
+        if max_total_object_size.is_some() {
+            let request_id = req
+                .extensions
+                .get::<super::storage_api::multipart_usecase::request_context::RequestContext>()
+                .map(|ctx| ctx.request_id.clone())
+                .unwrap_or_default();
+            body_stream = guard_put_object_body_read_timeout(
+                body_stream,
+                &bucket,
+                &key,
+                &request_id,
+                content_length,
+                put_object_body_read_timeout().max(Duration::from_secs(rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT)),
+            );
+        }
 
         if size.is_none() {
             let mut total = 0i64;
@@ -1005,16 +1106,6 @@ impl DefaultMultipartUsecase {
             body_stream = StreamingBlob::wrap(stream);
         }
 
-        // Get multipart info early to check if managed encryption will be applied
-        let Some(store) = self.object_store() else {
-            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
-        };
-
-        let fi = store
-            .get_multipart_info(&bucket, &key, &upload_id, &opts)
-            .await
-            .map_err(ApiError::from)?;
-
         let mut size = size.ok_or_else(|| s3_error!(UnexpectedContent))?;
         let ingress_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(std::time::Instant::now);
 
@@ -1229,6 +1320,16 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<ListMultipartUploadsInput>,
     ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         let mut opts = ObjectOptions::default();
         apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
         let ListMultipartUploadsInput {
@@ -1276,6 +1377,16 @@ impl DefaultMultipartUsecase {
     }
 
     pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         let mut opts = ObjectOptions::default();
         apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
         let ListPartsInput {
@@ -1307,6 +1418,16 @@ impl DefaultMultipartUsecase {
         &self,
         req: S3Request<UploadPartCopyInput>,
     ) -> S3Result<S3Response<UploadPartCopyOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         // Captured before `req.input` is destructured below.
         let copy_principal = SseKmsPrincipal::from_request(&req);
         let source_bucket = match &req.input.copy_source {
@@ -1405,6 +1526,7 @@ impl DefaultMultipartUsecase {
             .get_multipart_info(&bucket, &key, &upload_id, &dst_opts)
             .await
             .map_err(ApiError::from)?;
+        let destination_size_limit = multipart_max_total_object_size(&mp_info.user_defined)?;
         EncryptionRequest {
             bucket: &bucket,
             key: &key,
@@ -1487,19 +1609,25 @@ impl DefaultMultipartUsecase {
             return Err(s3_error!(PreconditionFailed));
         }
 
+        let source_logical_size = match src_info.get_actual_size() {
+            Ok(size) if size >= 0 => size,
+            Ok(_) | Err(_) if destination_size_limit.is_some() => {
+                return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
+            }
+            Ok(_) | Err(_) => src_info.size,
+        };
+
         let (_start_offset, length) = if let Some(ref range_spec) = rs {
             // Copy-source ranges are expressed over the logical plaintext object.
             // Encrypted (and compressed) objects have a larger or smaller physical
             // representation, so validating against `size` rejects valid later parts.
-            let validation_size = src_info.get_actual_size().unwrap_or(src_info.size);
-
-            validate_copy_source_range_not_exceeds(range_spec, validation_size)?;
+            validate_copy_source_range_not_exceeds(range_spec, source_logical_size)?;
 
             range_spec
-                .get_offset_length(validation_size)
+                .get_offset_length(source_logical_size)
                 .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRange, e.to_string()))?
         } else {
-            (0, src_info.size)
+            (0, source_logical_size)
         };
 
         let is_disk_compressed =
@@ -2101,6 +2229,16 @@ mod tests {
         assert_eq!(metadata.get(AMZ_OBJECT_TAGGING), Some(&"project=rustfs".to_string()));
     }
 
+    #[test]
+    fn multipart_max_total_object_size_reads_compatible_internal_metadata() {
+        let mut metadata = HashMap::new();
+        insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "104857600".to_string());
+        assert_eq!(multipart_max_total_object_size(&metadata).unwrap(), Some(104_857_600));
+
+        metadata.insert("x-minio-internal-max-total-object-size".to_string(), "1".to_string());
+        assert!(multipart_max_total_object_size(&metadata).is_err());
+    }
+
     #[tokio::test]
     async fn execute_complete_multipart_upload_rejects_missing_parts_payload() {
         let input = CompleteMultipartUploadInput::builder()

rustfs/src/app/object/copy.rs

@@ -16,6 +16,8 @@
 
 use super::*;
 
+use crate::auth::{VerifiedPresignedRequest, reject_presigned_put_max_content_length_for_other_operation};
+
 fn copy_namespace_lock_error(bucket: &str, object: &str, mode: &'static str, err: rustfs_lock::LockError) -> StorageError {
     match err {
         rustfs_lock::LockError::QuorumNotReached { required, achieved } => StorageError::NamespaceLockQuorumUnavailable {
@@ -92,6 +94,11 @@ impl DefaultObjectUsecase {
 
     #[instrument(name = "execute_copy_object", level = "debug", skip(self, req))]
     async fn execute_copy_object_inner(&self, req: S3Request<CopyObjectInput>) -> S3Result<S3Response<CopyObjectOutput>> {
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         if let Some(context) = &self.context {
             let _ = context.object_store();
         }

rustfs/src/app/object/get.rs

@@ -7242,6 +7242,12 @@ mod tests {
         let mut staged_targets = Vec::with_capacity(pool_disk_paths[target_pool].len());
         for (source_disk, target_disk) in pool_disk_paths[source_pool].iter().zip(&pool_disk_paths[target_pool]) {
             let source_dir = source_disk.join(&bucket).join(object);
+            // The same write-quorum minority gap tolerated above (#6701) can
+            // leave a lagging source-pool disk without the object; skip it and
+            // stage the replicas that exist — the reader tolerates the gap.
+            if !source_dir.join("xl.meta").is_file() {
+                continue;
+            }
             let target_dir = target_disk.join(&bucket).join(object);
             let staging_dir = temp_dir.path().join(format!("resume-relocate-{}", Uuid::new_v4()));
             std::fs::create_dir_all(&staging_dir).expect("create relocated target staging directory");
@@ -7258,15 +7264,18 @@ mod tests {
                 }
             }
             std::fs::copy(source_dir.join("xl.meta"), staging_dir.join("xl.meta")).expect("stage relocated object metadata");
-            staged_targets.push((staging_dir, target_dir));
+            staged_targets.push((staging_dir, target_dir, source_dir.join("xl.meta")));
         }
+        assert!(
+            staged_targets.len() > pool_disk_paths[source_pool].len() / 2,
+            "a write-quorum majority of the source pool's disks must hold the object to stage the relocation"
+        );
         let (version_dirs, deleted) = delete_object_part_shards(&pool_disk_paths[source_pool], &bucket, object, &[2, 3]);
         assert!(version_dirs > 0, "the source pool must have at least one version data directory");
         assert_eq!(deleted, version_dirs * 2);
 
-        for ((staging_dir, target_dir), source_disk) in staged_targets.into_iter().zip(&pool_disk_paths[source_pool]) {
+        for (staging_dir, target_dir, source_meta) in staged_targets {
             std::fs::rename(staging_dir, target_dir).expect("publish relocated target object");
-            let source_meta = source_disk.join(&bucket).join(object).join("xl.meta");
             std::fs::remove_file(source_meta).expect("remove relocated source object metadata");
         }
         store

rustfs/src/app/object/mod.rs

@@ -195,6 +195,7 @@ pub(crate) use self::delete::*;
 pub(crate) use self::extract::*;
 pub(crate) use self::get::*;
 use self::put::*;
+pub(crate) use self::put::{guard_put_object_body_read_timeout, put_object_body_read_timeout};
 pub(crate) use self::shared::*;
 #[cfg(test)]
 use self::test_support::*;

rustfs/src/app/object/put.rs

@@ -16,6 +16,9 @@
 
 use super::*;
 
+use crate::auth::{RUSTFS_MAX_CONTENT_LENGTH_QUERY, VerifiedPresignedRequest, parse_presigned_put_max_content_length};
+use crate::error::UploadLimitExceeded;
+
 const DEFAULT_PUT_LARGE_CONCURRENCY_TUNING_MIN_SIZE_BYTES: i64 = 32 * 1024 * 1024;
 
 const ENV_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES: &str = "RUSTFS_ZERO_COPY_EAGER_PUT_MAX_SIZE_BYTES";
@@ -87,7 +90,7 @@ fn resolve_put_object_authoritative_size(headers: &HeaderMap, content_length: Op
 /// Returns `Duration::ZERO` when disabled (`RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT=0`),
 /// in which case [`guard_put_object_body_read_timeout`] passes the body through
 /// untouched.
-fn put_object_body_read_timeout() -> Duration {
+pub(crate) fn put_object_body_read_timeout() -> Duration {
     Duration::from_secs(rustfs_utils::get_env_u64(
         rustfs_config::ENV_HTTP_REQUEST_BODY_READ_TIMEOUT,
         rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT,
@@ -124,6 +127,58 @@ struct RequestBodyReadTimeout {
     timed_out: bool,
 }
 
+/// Enforces a maximum size on the decoded request entity while preserving the
+/// streaming behavior of the underlying S3 body.
+struct MaxContentLengthStream {
+    inner: StreamingBlob,
+    limit: u64,
+    received: u64,
+    exceeded: bool,
+}
+
+impl Stream for MaxContentLengthStream {
+    type Item = Result<Bytes, StdError>;
+
+    fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
+        let this = self.as_mut().get_mut();
+        if this.exceeded {
+            return Poll::Ready(None);
+        }
+
+        match Pin::new(&mut this.inner).poll_next(cx) {
+            Poll::Ready(Some(Ok(chunk))) => {
+                let chunk_len = u64::try_from(chunk.len()).unwrap_or(u64::MAX);
+                let exceeds = this.received > this.limit || chunk_len > this.limit.saturating_sub(this.received);
+                if exceeds {
+                    this.exceeded = true;
+                    return Poll::Ready(Some(Err(Box::new(UploadLimitExceeded { limit: this.limit }))));
+                }
+
+                this.received = this.received.saturating_add(chunk_len);
+                Poll::Ready(Some(Ok(chunk)))
+            }
+            other => other,
+        }
+    }
+
+    fn size_hint(&self) -> (usize, Option<usize>) {
+        let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
+        let (lower, upper) = self.inner.size_hint();
+        (lower.min(remaining), upper.map(|upper| upper.min(remaining)))
+    }
+}
+
+impl ByteStream for MaxContentLengthStream {
+    fn remaining_length(&self) -> RemainingLength {
+        let remaining = usize::try_from(self.limit.saturating_sub(self.received)).unwrap_or(usize::MAX);
+        let inner = self.inner.remaining_length();
+        inner
+            .exact()
+            .map(|exact| RemainingLength::new_exact(exact.min(remaining)))
+            .unwrap_or_else(RemainingLength::unknown)
+    }
+}
+
 impl Stream for RequestBodyReadTimeout {
     type Item = Result<Bytes, StdError>;
 
@@ -205,7 +260,7 @@ impl ByteStream for RequestBodyReadTimeout {
 /// Wrap an incoming request body with [`RequestBodyReadTimeout`] unless the
 /// feature is disabled (`timeout == 0`), in which case the body is returned
 /// untouched. `remaining_length` is preserved via [`StreamingBlob::new`].
-fn guard_put_object_body_read_timeout(
+pub(crate) fn guard_put_object_body_read_timeout(
     body: StreamingBlob,
     bucket: &str,
     key: &str,
@@ -752,6 +807,11 @@ impl DefaultObjectUsecase {
         }
 
         let (event_name, quota_operation, request_method_name) = Self::put_object_execution_context(&req);
+        let max_content_length = parse_presigned_put_max_content_length(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
         if req.extensions.get::<PostObjectRequestMarker>().is_some() && is_post_object_sse_kms_requested(&req.input, &req.headers)
         {
             return Err(s3_error!(NotImplemented, "SSE-KMS is not supported for POST object uploads"));
@@ -769,6 +829,12 @@ impl DefaultObjectUsecase {
         // member) instead of writing the replica.
         let inbound_replication_put = replication_request_authorized(&req)
             && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true");
+        if max_content_length.is_some() && is_put_object_extract_requested(&req.headers) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is not supported for archive extraction"),
+            ));
+        }
         if is_put_object_extract_requested(&req.headers) && !inbound_replication_put {
             return Box::pin(self.execute_put_object_extract(req)).await;
         }
@@ -842,9 +908,25 @@ impl DefaultObjectUsecase {
             guard_put_object_body_read_timeout(body, &bucket, &key, &request_id, content_length, put_object_body_read_timeout())
         };
 
+        let body = match max_content_length {
+            Some(limit) => StreamingBlob::new(MaxContentLengthStream {
+                inner: body,
+                limit,
+                received: 0,
+                exceeded: false,
+            }),
+            None => body,
+        };
+
         // Resolve the authoritative decoded/plain object length (rejecting negative/unknown) before anything else consumes it.
         let mut size = resolve_put_object_authoritative_size(&req.headers, content_length)?;
 
+        if let Some(limit) = max_content_length
+            && u64::try_from(size).is_ok_and(|size| size > limit)
+        {
+            return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
+        }
+
         // The app check preserves the existing S3 error contract; the storage
         // commit path reserves the exact net logical growth under its locks.
         let quota_check = self
@@ -880,7 +962,7 @@ impl DefaultObjectUsecase {
         rustfs_io_metrics::record_put_object_stage_duration_from("app_bucket_validate", bucket_validate_stage_start);
 
         let put_admission = match get_concurrency_manager()
-            .admit_put_object()
+            .admit_put_object(size)
             .await
             .map_err(|_| s3_error!(InternalError, "foreground write admission closed"))?
         {
@@ -1555,6 +1637,7 @@ pub(super) fn previous_current_size_from_backfill(backfill: Option<OldCurrentSiz
 #[cfg(test)]
 mod tests {
     use super::*;
+    use futures::StreamExt;
     use http::{HeaderMap, HeaderName, HeaderValue, Method};
     use s3s::dto::{DefaultRetention, ObjectLockConfiguration, ObjectLockEnabled, ObjectLockRule};
     use std::pin::Pin;
@@ -1594,6 +1677,39 @@ mod tests {
         .expect("cancelled owner must abort and reap the stalled storage task");
     }
 
+    #[tokio::test]
+    async fn max_content_length_stream_rejects_the_first_chunk_over_limit() {
+        let inner = StreamingBlob::wrap(futures::stream::iter([
+            Ok::<Bytes, std::io::Error>(Bytes::from_static(b"1234")),
+            Ok::<Bytes, std::io::Error>(Bytes::from_static(b"56")),
+        ]));
+        let mut limited = MaxContentLengthStream {
+            inner,
+            limit: 5,
+            received: 0,
+            exceeded: false,
+        };
+
+        assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"1234"));
+        let error = limited.next().await.unwrap().unwrap_err();
+        assert!(error.downcast_ref::<UploadLimitExceeded>().is_some());
+        assert!(limited.next().await.is_none());
+    }
+
+    #[tokio::test]
+    async fn max_content_length_stream_allows_exact_limit() {
+        let inner = StreamingBlob::from_bytes(Bytes::from_static(b"12345"));
+        let mut limited = MaxContentLengthStream {
+            inner,
+            limit: 5,
+            received: 0,
+            exceeded: false,
+        };
+
+        assert_eq!(limited.next().await.unwrap().unwrap(), Bytes::from_static(b"12345"));
+        assert!(limited.next().await.is_none());
+    }
+
     #[test]
     fn put_request_user_metadata_cannot_suppress_bucket_default_retention() {
         let mut metadata =

rustfs/src/auth.rs

@@ -38,6 +38,7 @@ use subtle::ConstantTimeEq;
 use time::OffsetDateTime;
 use time::format_description::well_known::Rfc3339;
 use tracing::{debug, trace, warn};
+use url::form_urlencoded;
 
 const LOG_COMPONENT_AUTH: &str = "auth";
 const LOG_SUBSYSTEM_CREDENTIALS: &str = "credentials";
@@ -50,6 +51,19 @@ const EVENT_KEYSTONE_CREDENTIALS_VALIDATED: &str = "keystone_credentials_validat
 const EVENT_KEYSTONE_CONTEXT_MISSING: &str = "keystone_context_missing";
 const EVENT_SESSION_TOKEN_EXTRACTION: &str = "session_token_extraction";
 
+/// RustFS-specific query capability for a single presigned PutObject request.
+pub(crate) const RUSTFS_MAX_CONTENT_LENGTH_QUERY: &str = "x-rustfs-max-content-length";
+pub(crate) const RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY: &str = "x-rustfs-max-total-object-size";
+
+/// Inserted by the S3 access boundary after the upstream verifier accepts a
+/// request as SigV4 presigned. Downstream capability parsing must require this
+/// marker instead of treating query syntax as proof of authentication.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct VerifiedPresignedRequest;
+
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct VerifiedSigV4Request;
+
 /// Performs constant-time string comparison to prevent timing attacks.
 ///
 /// This function should be used when comparing sensitive values like passwords,
@@ -913,9 +927,11 @@ pub(crate) fn is_request_presigned_signature_v4_with_query(header: &HeaderMap, q
     if let Some(credential) = header.get(AMZ_CREDENTIAL) {
         return !credential.to_str().unwrap_or("").is_empty();
     }
-    query
-        .and_then(|query| get_query_param(query, "x-amz-credential"))
-        .is_some_and(|credential| !credential.is_empty())
+    query.is_some_and(|query| {
+        form_urlencoded::parse(query.as_bytes())
+            .find(|(name, _)| name.eq_ignore_ascii_case("x-amz-credential"))
+            .is_some_and(|(_, credential)| !credential.is_empty())
+    })
 }
 
 /// Verify request has AWS PreSign Version '2'
@@ -1007,6 +1023,184 @@ pub fn get_query_param<'a>(query: &'a str, param_name: &str) -> Option<&'a str>
     None
 }
 
+/// Parse the RustFS presigned PutObject size capability after authentication.
+///
+/// The query value is covered by SigV4 when it is present before presigning, but
+/// the signature does not assign any semantics to the extension. Keep parsing
+/// strict and only enable the capability for a verified SigV4 presigned request.
+pub(crate) fn parse_presigned_put_max_content_length(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_presigned: bool,
+) -> S3Result<Option<u64>> {
+    let Some(query) = query else {
+        return Ok(None);
+    };
+
+    let mut value = None;
+    let mut decoded_query = Vec::new();
+    for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
+        decoded_query.push((name.to_string(), candidate.to_string()));
+        if name == RUSTFS_MAX_CONTENT_LENGTH_QUERY {
+            if value.is_some() {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidRequest,
+                    format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must appear exactly once"),
+                ));
+            }
+            value = Some(candidate.into_owned());
+        } else if name.eq_ignore_ascii_case(RUSTFS_MAX_CONTENT_LENGTH_QUERY) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("query parameter name must be exactly {RUSTFS_MAX_CONTENT_LENGTH_QUERY}"),
+            ));
+        }
+    }
+
+    let Some(value) = value else {
+        return Ok(None);
+    };
+
+    let query_value = |wanted: &str| {
+        decoded_query
+            .iter()
+            .find(|(name, _)| name.eq_ignore_ascii_case(wanted))
+            .map(|(_, value)| value.as_str())
+    };
+    let is_complete_sigv4_query = [
+        ("x-amz-algorithm", "AWS4-HMAC-SHA256"),
+        ("x-amz-date", ""),
+        ("x-amz-expires", ""),
+        ("x-amz-signedheaders", ""),
+        ("x-amz-credential", ""),
+        ("x-amz-signature", ""),
+    ]
+    .into_iter()
+    .all(|(name, expected)| {
+        query_value(name).is_some_and(|value| !value.is_empty() && (expected.is_empty() || value == expected))
+    });
+    if !verified_presigned
+        || !is_complete_sigv4_query
+        || !matches!(get_request_auth_type_with_query(header, Some(query)), AuthType::Presigned)
+    {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} requires a SigV4 presigned request"),
+        ));
+    }
+
+    let limit = value.parse::<u64>().map_err(|_| {
+        S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} must be a non-negative 64-bit integer"),
+        )
+    })?;
+
+    Ok(Some(limit))
+}
+
+/// Reject the PutObject-only size capability when it appears on another
+/// operation. Callers must invoke this after request authentication has run.
+pub(crate) fn reject_presigned_put_max_content_length_for_other_operation(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_presigned: bool,
+) -> S3Result<()> {
+    if parse_presigned_put_max_content_length(header, query, verified_presigned)?.is_some() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
+        ));
+    }
+    Ok(())
+}
+
+/// Parse the V2 multipart total-size capability after SigV4 authentication.
+/// Header-authenticated CreateMultipartUpload requests are accepted because the
+/// custom query is covered by the SigV4 canonical request; later multipart
+/// operations read the immutable value from the upload session metadata.
+pub(crate) fn parse_presigned_multipart_max_total_object_size(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_sigv4: bool,
+) -> S3Result<Option<u64>> {
+    let Some(query) = query else {
+        return Ok(None);
+    };
+
+    let mut value = None;
+    let mut decoded_query = Vec::new();
+    for (name, candidate) in form_urlencoded::parse(query.as_bytes()) {
+        decoded_query.push((name.to_string(), candidate.to_string()));
+        if name == RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY {
+            if value.is_some() {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidRequest,
+                    format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must appear exactly once"),
+                ));
+            }
+            value = Some(candidate.into_owned());
+        } else if name.eq_ignore_ascii_case(RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY) {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("query parameter name must be exactly {RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY}"),
+            ));
+        }
+    }
+
+    let Some(value) = value else {
+        return Ok(None);
+    };
+
+    let auth_type = get_request_auth_type_with_query(header, Some(query));
+    let is_presigned = matches!(auth_type, AuthType::Presigned);
+    let is_header_signed = matches!(auth_type, AuthType::Signed);
+    let complete_presigned_query = [
+        ("x-amz-algorithm", "AWS4-HMAC-SHA256"),
+        ("x-amz-date", ""),
+        ("x-amz-expires", ""),
+        ("x-amz-signedheaders", ""),
+        ("x-amz-credential", ""),
+        ("x-amz-signature", ""),
+    ]
+    .into_iter()
+    .all(|(name, expected)| {
+        decoded_query
+            .iter()
+            .find(|(candidate, _)| candidate.eq_ignore_ascii_case(name))
+            .is_some_and(|(_, candidate)| !candidate.is_empty() && (expected.is_empty() || candidate == expected))
+    });
+
+    let authenticated = verified_sigv4 && (is_header_signed || (is_presigned && complete_presigned_query));
+    if !authenticated {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} requires a verified SigV4 request"),
+        ));
+    }
+
+    value.parse::<u64>().map(Some).map_err(|_| {
+        S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} must be a non-negative 64-bit integer"),
+        )
+    })
+}
+
+pub(crate) fn reject_presigned_multipart_max_total_object_size_for_other_operation(
+    header: &HeaderMap,
+    query: Option<&str>,
+    verified_sigv4: bool,
+) -> S3Result<()> {
+    if parse_presigned_multipart_max_total_object_size(header, query, verified_sigv4)?.is_some() {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRequest,
+            format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"),
+        ));
+    }
+    Ok(())
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -1651,6 +1845,113 @@ mod tests {
         assert_eq!(result, Some("value=with=equals"));
     }
 
+    #[test]
+    fn presigned_put_max_content_length_requires_exactly_one_signed_query_value() {
+        let headers = HeaderMap::new();
+        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
+
+        let query = format!("{signed_prefix}&x-rustfs-max-content-length=104857600");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&query), true).unwrap(),
+            Some(104_857_600)
+        );
+
+        let encoded_credential = query.replacen("X-Amz-Credential", "X%2DAmz-Credential", 1);
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&encoded_credential), true).unwrap(),
+            Some(104_857_600)
+        );
+
+        let duplicate = format!("{query}&x-rustfs-max-content-length=1");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&duplicate), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+
+        let wrong_case = format!("{signed_prefix}&X-RustFS-Max-Content-Length=1");
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(&wrong_case), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+
+        assert_eq!(
+            reject_presigned_put_max_content_length_for_other_operation(&headers, Some(&query), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+    }
+
+    #[test]
+    fn presigned_put_max_content_length_rejects_unsigned_or_invalid_values() {
+        let headers = HeaderMap::new();
+        let forged = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/credential&X-Amz-Signature=fake&x-rustfs-max-content-length=1";
+        assert_eq!(
+            parse_presigned_put_max_content_length(&headers, Some(forged), false)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+        for query in [
+            "x-rustfs-max-content-length=1",
+            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=-1",
+            "X-Amz-Credential=test/credential&x-rustfs-max-content-length=18446744073709551616",
+        ] {
+            let error = parse_presigned_put_max_content_length(&headers, Some(query), true).unwrap_err();
+            assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+        }
+    }
+
+    #[test]
+    fn multipart_max_total_object_size_requires_signed_create_request() {
+        let headers = HeaderMap::new();
+        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
+        let query = format!("{signed_prefix}&x-rustfs-max-total-object-size=104857600");
+
+        assert_eq!(
+            parse_presigned_multipart_max_total_object_size(&headers, Some(&query), true).unwrap(),
+            Some(104_857_600)
+        );
+        assert_eq!(
+            reject_presigned_multipart_max_total_object_size_for_other_operation(&headers, Some(&query), true)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+    }
+
+    #[test]
+    fn multipart_max_total_object_size_rejects_tampering_and_invalid_values() {
+        let headers = HeaderMap::new();
+        let signed_prefix = "X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Date=20260827T000000Z&X-Amz-Expires=900&X-Amz-SignedHeaders=host&X-Amz-Credential=test/20260827/us-east-1/s3/aws4_request&X-Amz-Signature=signature";
+        for query in [
+            "x-rustfs-max-total-object-size=1",
+            "X-RustFS-Max-Total-Object-Size=1",
+            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=1&x-rustfs-max-total-object-size=2",
+            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=-1",
+            "X-Amz-Algorithm=AWS4-HMAC-SHA256&x-rustfs-max-total-object-size=18446744073709551616",
+        ] {
+            assert_eq!(
+                parse_presigned_multipart_max_total_object_size(&headers, Some(query), true)
+                    .unwrap_err()
+                    .code(),
+                &S3ErrorCode::InvalidRequest
+            );
+        }
+
+        let forged = format!("{signed_prefix}&x-rustfs-max-total-object-size=1");
+        assert_eq!(
+            parse_presigned_multipart_max_total_object_size(&headers, Some(&forged), false)
+                .unwrap_err()
+                .code(),
+            &S3ErrorCode::InvalidRequest
+        );
+    }
+
     #[test]
     fn test_credentials_is_expired() {
         let mut cred = create_test_credentials();

rustfs/src/config/info.rs

@@ -591,12 +591,19 @@ struct FeatureSpec {
     default_enabled: bool,
 }
 
-fn feature_specs() -> [FeatureSpec; 7] {
-    [
+fn feature_specs() -> &'static [FeatureSpec] {
+    &[
+        FeatureSpec {
+            name: "default",
+            enabled: cfg!(feature = "default"),
+            description: "Default feature set",
+            dependencies: "ftps + webdav",
+            default_enabled: true,
+        },
         FeatureSpec {
             name: "metrics-gpu",
             enabled: cfg!(feature = "metrics-gpu"),
-            description: "Metrics GPU support",
+            description: "GPU metrics support",
             dependencies: "rustfs-obs/gpu",
             default_enabled: false,
         },
@@ -610,7 +617,7 @@ fn feature_specs() -> [FeatureSpec; 7] {
         FeatureSpec {
             name: "swift",
             enabled: cfg!(feature = "swift"),
-            description: "Swift storage backend",
+            description: "OpenStack Swift protocol support",
             dependencies: "rustfs-protocols/swift",
             default_enabled: false,
         },
@@ -621,6 +628,13 @@ fn feature_specs() -> [FeatureSpec; 7] {
             dependencies: "rustfs-protocols/webdav",
             default_enabled: true,
         },
+        FeatureSpec {
+            name: "sftp",
+            enabled: cfg!(feature = "sftp"),
+            description: "SFTP protocol support",
+            dependencies: "rustfs-protocols/sftp",
+            default_enabled: false,
+        },
         FeatureSpec {
             name: "license",
             enabled: cfg!(feature = "license"),
@@ -635,11 +649,81 @@ fn feature_specs() -> [FeatureSpec; 7] {
             dependencies: "(none)",
             default_enabled: false,
         },
+        FeatureSpec {
+            name: "tracing-chunk-debug",
+            enabled: cfg!(feature = "tracing-chunk-debug"),
+            description: "Per-chunk data-plane tracing",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
         FeatureSpec {
             name: "full",
             enabled: cfg!(feature = "full"),
-            description: "All features enabled",
-            dependencies: "metrics-gpu + ftps + swift + webdav",
+            description: "Full protocol and observability bundle",
+            dependencies: "metrics-gpu + ftps + swift + webdav + sftp + pyroscope",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "e2e-test-hooks",
+            enabled: cfg!(feature = "e2e-test-hooks"),
+            description: "End-to-end test hooks",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "connect-e2e-short-credentials",
+            enabled: cfg!(feature = "connect-e2e-short-credentials"),
+            description: "Short-lived Connect credentials for debug E2E builds",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "offline-enrollment-e2e-root",
+            enabled: cfg!(feature = "offline-enrollment-e2e-root"),
+            description: "Dedicated offline enrollment E2E root",
+            dependencies: "(none)",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "rio-v2",
+            enabled: cfg!(feature = "rio-v2"),
+            description: "RIO v2 storage path support",
+            dependencies: "rustfs-ecstore/rio-v2",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "pyroscope",
+            enabled: cfg!(feature = "pyroscope"),
+            description: "Pyroscope profiling support",
+            dependencies: "rustfs-obs/pyroscope",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "dial9",
+            enabled: cfg!(feature = "dial9"),
+            description: "Tokio runtime telemetry",
+            dependencies: "rustfs-obs/dial9",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath",
+            enabled: cfg!(feature = "hotpath"),
+            description: "Hotpath instrumentation",
+            dependencies: "hotpath + RustFS crate hotpath features",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath-alloc",
+            enabled: cfg!(feature = "hotpath-alloc"),
+            description: "Hotpath allocation diagnostics",
+            dependencies: "hotpath + hotpath/hotpath-alloc + RustFS crate hotpath-alloc features",
+            default_enabled: false,
+        },
+        FeatureSpec {
+            name: "hotpath-cpu",
+            enabled: cfg!(feature = "hotpath-cpu"),
+            description: "Hotpath CPU attribution",
+            dependencies: "hotpath + hotpath/hotpath-cpu + RustFS crate hotpath-cpu features",
             default_enabled: false,
         },
     ]
@@ -655,7 +739,7 @@ struct DepsInfoJson {
 
 fn collect_deps_info_json() -> DepsInfoJson {
     let features: Vec<FeatureInfoJson> = feature_specs()
-        .into_iter()
+        .iter()
         .map(|feature| FeatureInfoJson {
             name: feature.name,
             enabled: feature.enabled,
@@ -901,7 +985,7 @@ fn format_deps_info() -> String {
     output.push_str("### Feature Status\n\n");
     output.push_str("| Feature | Status | Description |\n");
     output.push_str("|---------|--------|-------------|\n");
-    for feature in &features {
+    for feature in features {
         let status = if feature.enabled { "✓" } else { "✗" };
         output.push_str(&format!("| {} | {} | {} |\n", feature.name, status, feature.description));
     }
@@ -916,7 +1000,7 @@ fn format_deps_info() -> String {
     output.push_str("\n### Feature Dependencies\n\n");
     output.push_str("| Feature | Dependencies |\n");
     output.push_str("|---------|-------------|\n");
-    for feature in &features {
+    for feature in features {
         output.push_str(&format!("| {} | {} |\n", feature.name, feature.dependencies));
     }
 
@@ -1001,10 +1085,22 @@ mod tests {
         let info = collect_deps_info_json();
         let feature_names: Vec<_> = info.features.iter().map(|feature| feature.name).collect();
 
-        assert_eq!(info.total_count, 7);
-        assert_eq!(info.features.len(), 7);
+        assert_eq!(info.total_count, 19);
+        assert_eq!(info.features.len(), 19);
+        assert!(feature_names.contains(&"default"));
         assert!(feature_names.contains(&"metrics-gpu"));
+        assert!(feature_names.contains(&"sftp"));
         assert!(feature_names.contains(&"io-scheduler-debug"));
+        assert!(feature_names.contains(&"tracing-chunk-debug"));
+        assert!(feature_names.contains(&"e2e-test-hooks"));
+        assert!(feature_names.contains(&"connect-e2e-short-credentials"));
+        assert!(feature_names.contains(&"offline-enrollment-e2e-root"));
+        assert!(feature_names.contains(&"rio-v2"));
+        assert!(feature_names.contains(&"pyroscope"));
+        assert!(feature_names.contains(&"dial9"));
+        assert!(feature_names.contains(&"hotpath"));
+        assert!(feature_names.contains(&"hotpath-alloc"));
+        assert!(feature_names.contains(&"hotpath-cpu"));
         assert!(!feature_names.contains(&"manual-test-runners"));
         assert!(!feature_names.contains(&"metrics"));
         assert!(!feature_names.contains(&"direct-io"));
@@ -1016,10 +1112,16 @@ mod tests {
 
         assert!(output.contains("| metrics-gpu |"));
         assert!(output.contains("| io-scheduler-debug |"));
+        assert!(output.contains("| tracing-chunk-debug |"));
+        assert!(output.contains("| sftp |"));
+        assert!(output.contains("| rio-v2 |"));
+        assert!(output.contains("| dial9 |"));
+        assert!(output.contains("| hotpath-cpu |"));
+        assert!(output.contains("| default | enabled by default |"));
         assert!(!output.contains("| manual-test-runners |"));
         assert!(output.contains("| ftps | enabled by default |"));
         assert!(output.contains("| webdav | enabled by default |"));
-        assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav |"));
+        assert!(output.contains("| full | metrics-gpu + ftps + swift + webdav + sftp + pyroscope |"));
         assert!(!output.contains("| direct-io |"));
     }
 

rustfs/src/error.rs

@@ -17,6 +17,23 @@ use crate::storage_api::error::{QuotaError, StorageError};
 use rustfs_kms::KmsUnavailableError;
 use s3s::{S3Error, S3ErrorCode};
 
+/// Marks a request body that exceeded a presigned upload size capability.
+///
+/// This marker must survive the body-reader and storage layers so the client
+/// receives `EntityTooLarge` instead of a generic internal error.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct UploadLimitExceeded {
+    pub limit: u64,
+}
+
+impl std::fmt::Display for UploadLimitExceeded {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(f, "upload exceeds the maximum content length of {} bytes", self.limit)
+    }
+}
+
+impl std::error::Error for UploadLimitExceeded {}
+
 #[derive(Debug)]
 pub struct ApiError {
     pub code: S3ErrorCode,
@@ -274,6 +291,17 @@ impl From<StorageError> for ApiError {
             };
         }
 
+        if let StorageError::Io(ref io_err) = err
+            && let Some(inner) = io_err.get_ref()
+            && error_chain_has_type::<UploadLimitExceeded>(inner)
+        {
+            return ApiError {
+                code: S3ErrorCode::EntityTooLarge,
+                message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                source: Some(Box::new(err)),
+            };
+        }
+
         if let StorageError::Io(ref io_err) = err
             && io_err
                 .get_ref()
@@ -344,6 +372,7 @@ impl From<StorageError> for ApiError {
             StorageError::ObjectExistsAsDirectory(_, _) => S3ErrorCode::InvalidArgument,
             StorageError::InvalidPart(_, _, _) => S3ErrorCode::InvalidPart,
             StorageError::EntityTooSmall(_, _, _) => S3ErrorCode::EntityTooSmall,
+            StorageError::EntityTooLarge(_, _) => S3ErrorCode::EntityTooLarge,
             StorageError::PreconditionFailed => S3ErrorCode::PreconditionFailed,
             StorageError::NotModified => S3ErrorCode::NotModified,
             StorageError::InvalidRangeSpec(_) => S3ErrorCode::InvalidRange,
@@ -399,6 +428,13 @@ impl From<std::io::Error> for ApiError {
                     source: Some(Box::new(err)),
                 };
             }
+            if error_chain_has_type::<UploadLimitExceeded>(inner) {
+                return ApiError {
+                    code: S3ErrorCode::EntityTooLarge,
+                    message: ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge),
+                    source: Some(Box::new(err)),
+                };
+            }
             if error_chain_has_type::<rustfs_rio::IncompleteBody>(inner) {
                 return ApiError {
                     code: S3ErrorCode::IncompleteBody,
@@ -815,6 +851,16 @@ mod tests {
         assert!(api_error.source.is_some());
     }
 
+    #[test]
+    fn upload_limit_marker_maps_to_entity_too_large_across_io_boundaries() {
+        let direct: ApiError = IoError::other(UploadLimitExceeded { limit: 5 }).into();
+        assert_eq!(direct.code, S3ErrorCode::EntityTooLarge);
+
+        let storage: ApiError = StorageError::Io(IoError::other(IoError::other(UploadLimitExceeded { limit: 5 }))).into();
+        assert_eq!(storage.code, S3ErrorCode::EntityTooLarge);
+        assert_eq!(storage.message, ApiError::error_code_to_message(&S3ErrorCode::EntityTooLarge));
+    }
+
     #[test]
     fn test_api_error_from_storage_io_copy_object_terminal_error_stays_internal() {
         let io_error = IoError::other(StorageError::FileCorrupt);

rustfs/src/server/health.rs

@@ -321,13 +321,15 @@ pub(crate) fn build_health_response_parts(
             ),
         };
 
-    let object_traffic_stalled = degraded_reasons.iter().any(|reason| {
+    let readiness_overlay_degraded = degraded_reasons.iter().any(|reason| {
         matches!(
             reason,
-            ReadinessDegradedReason::ObjectReadStalled | ReadinessDegradedReason::ObjectWriteStalled
+            ReadinessDegradedReason::ObjectReadStalled
+                | ReadinessDegradedReason::ObjectWriteStalled
+                | ReadinessDegradedReason::StartupFinalizationPending
         )
     });
-    if probe == HealthProbe::Readiness && (object_traffic_stalled || matches!(kms_ready, Some(false))) {
+    if probe == HealthProbe::Readiness && (readiness_overlay_degraded || matches!(kms_ready, Some(false))) {
         health = HealthCheckState {
             status_code: StatusCode::SERVICE_UNAVAILABLE,
             status: "degraded",

rustfs/src/server/http.rs

@@ -156,6 +156,7 @@ fn rustfs_s3_config() -> S3Config {
     let mut s3_config = S3Config::default();
     s3_config.normalize_forward_slash_path = true;
     s3_config.enable_sig_v2 = true;
+    s3_config.sig_v4_allowed_services.push("s3tables".to_string());
     s3_config
 }
 
@@ -1677,7 +1678,10 @@ fn process_connection(
                 .option_layer(if is_console { Some(RedirectLayer) } else { None })
                 .layer(BodylessStatusFixLayer)
                 .layer(HeadRequestBodyFixLayer)
-                .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx)))
+                .layer(PublicHealthEndpointLayer::new(
+                    Arc::clone(&server_ctx),
+                    Arc::clone(&readiness),
+                ))
                 .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer))
                 .layer(DoubleSlashListBucketsCompatLayer)
                 .service(service)
@@ -2270,6 +2274,9 @@ mod tests {
 
         assert!(s3_config.normalize_forward_slash_path);
         assert!(s3_config.enable_sig_v2);
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3"));
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "sts"));
+        assert!(s3_config.sig_v4_allowed_services.iter().any(|service| service == "s3tables"));
     }
 
     #[test]

rustfs/src/server/layer.rs

@@ -26,6 +26,7 @@ use crate::server::{
     build_health_response_parts, collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path,
     kms_probe_staleness_limit, kms_ready_from_probe,
 };
+use crate::shared_types::ReadinessDegradedReason;
 use crate::storage_api::server::layer::apply_cors_headers;
 use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
 use bytes::{Bytes, BytesMut};
@@ -36,6 +37,7 @@ use http_body_util::{BodyExt, Full};
 use hyper::body::Incoming;
 use pin_project_lite::pin_project;
 use quick_xml::events::Event;
+use rustfs_common::GlobalReadiness;
 use rustfs_obs::HTTP_SERVER_LOG_TARGET;
 #[cfg(feature = "swift")]
 use rustfs_protocols::swift::SwiftRouter;
@@ -1243,11 +1245,12 @@ where
 #[derive(Clone)]
 pub struct PublicHealthEndpointLayer {
     server_ctx: Arc<crate::runtime_sources::ServerContextSlot>,
+    readiness: Arc<GlobalReadiness>,
 }
 
 impl PublicHealthEndpointLayer {
-    pub fn new(server_ctx: Arc<crate::runtime_sources::ServerContextSlot>) -> Self {
-        Self { server_ctx }
+    pub fn new(server_ctx: Arc<crate::runtime_sources::ServerContextSlot>, readiness: Arc<GlobalReadiness>) -> Self {
+        Self { server_ctx, readiness }
     }
 }
 
@@ -1258,6 +1261,7 @@ impl<S> Layer<S> for PublicHealthEndpointLayer {
         PublicHealthEndpointService {
             inner,
             server_ctx: Arc::clone(&self.server_ctx),
+            readiness: Arc::clone(&self.readiness),
         }
     }
 }
@@ -1266,6 +1270,7 @@ impl<S> Layer<S> for PublicHealthEndpointLayer {
 pub struct PublicHealthEndpointService<S> {
     inner: S,
     server_ctx: Arc<crate::runtime_sources::ServerContextSlot>,
+    readiness: Arc<GlobalReadiness>,
 }
 
 fn health_endpoint_enabled() -> bool {
@@ -1334,6 +1339,7 @@ async fn build_public_health_http_response<RestBody, GrpcBody>(
     method: Method,
     path: String,
     object_traffic_health: Option<Arc<ObjectTrafficHealth>>,
+    readiness: &GlobalReadiness,
 ) -> Response<HybridBody<RestBody, GrpcBody>>
 where
     RestBody: From<Bytes>,
@@ -1358,7 +1364,15 @@ where
             .expect("failed to build health busy response");
     }
 
-    let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await;
+    let mut readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await;
+    if probe == HealthProbe::Readiness
+        && !readiness.is_ready()
+        && let Some(report) = readiness_report.as_mut()
+    {
+        report
+            .degraded_reasons
+            .push(ReadinessDegradedReason::StartupFinalizationPending);
+    }
     let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() {
         Some(health_kms_ready().await)
     } else {
@@ -1408,7 +1422,10 @@ where
                 .server_ctx
                 .installed_app_context()
                 .map(|context| context.object_traffic_health());
-            return Box::pin(async move { Ok(build_public_health_http_response(method, path, object_traffic_health).await) });
+            let readiness = Arc::clone(&self.readiness);
+            return Box::pin(async move {
+                Ok(build_public_health_http_response(method, path, object_traffic_health, readiness.as_ref()).await)
+            });
         }
 
         let mut inner = self.inner.clone();
@@ -2210,14 +2227,25 @@ mod tests {
     use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt};
 
     fn public_health_layer() -> PublicHealthEndpointLayer {
-        PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new())
+        let readiness = Arc::new(GlobalReadiness::new());
+        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+        PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new(), readiness)
     }
 
     async fn public_health_layer_with_tracker(object_traffic_health: Arc<ObjectTrafficHealth>) -> PublicHealthEndpointLayer {
+        let readiness = Arc::new(GlobalReadiness::new());
+        readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+        public_health_layer_with_tracker_and_readiness(object_traffic_health, readiness).await
+    }
+
+    async fn public_health_layer_with_tracker_and_readiness(
+        object_traffic_health: Arc<ObjectTrafficHealth>,
+        readiness: Arc<GlobalReadiness>,
+    ) -> PublicHealthEndpointLayer {
         let app_context = crate::app::gating_test_env::app_context_with_object_traffic_health(object_traffic_health).await;
         let server_ctx = crate::runtime_sources::ServerContextSlot::new();
         assert!(server_ctx.install(app_context));
-        PublicHealthEndpointLayer::new(server_ctx)
+        PublicHealthEndpointLayer::new(server_ctx, readiness)
     }
 
     #[derive(Clone, Debug)]
@@ -2987,6 +3015,82 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn public_readiness_waits_for_s3_admission_publication() {
+        async_with_vars(
+            [
+                (rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true")),
+                (rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false")),
+            ],
+            async {
+                let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO));
+                let readiness = Arc::new(GlobalReadiness::new());
+                let inner = CountingHybridService::default();
+                let calls = inner.calls();
+                let mut service = public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness))
+                    .await
+                    .layer(inner);
+
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request before admission publication"),
+                    )
+                    .await
+                    .expect("readiness response before admission publication");
+                assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("readiness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON");
+                assert_eq!(payload["ready"], false);
+                assert_eq!(payload["details"]["storage"]["ready"], true);
+                assert_eq!(payload["details"]["iam"]["ready"], true);
+                assert_eq!(payload["details"]["lock"]["ready"], true);
+                assert_eq!(payload["degradedReasons"], serde_json::json!(["startup_finalization_pending"]));
+
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::GET)
+                            .uri(HEALTH_COMPAT_LIVE_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("liveness request before admission publication"),
+                    )
+                    .await
+                    .expect("liveness response before admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                let body = BodyExt::collect(response.into_body())
+                    .await
+                    .expect("liveness body before admission publication")
+                    .to_bytes();
+                let payload: serde_json::Value = serde_json::from_slice(&body).expect("liveness JSON");
+                assert_eq!(payload["status"], "ok");
+                assert!(payload.get("ready").is_none());
+
+                readiness.mark_stage(rustfs_common::SystemStage::FullReady);
+                let response = service
+                    .call(
+                        Request::builder()
+                            .method(Method::HEAD)
+                            .uri(MINIO_HEALTH_READY_PATH)
+                            .body(Full::<Bytes>::from(Bytes::new()))
+                            .expect("readiness request after admission publication"),
+                    )
+                    .await
+                    .expect("readiness response after admission publication");
+                assert_eq!(response.status(), StatusCode::OK);
+                assert_eq!(calls.load(Ordering::SeqCst), 0);
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial]
     async fn public_readiness_aliases_use_the_installed_object_progress() {

rustfs/src/shared_types.rs

@@ -45,6 +45,7 @@ pub enum ReadinessDegradedReason {
     ObjectWriteStalled,
     ClusterHealthTimeout,
     PeerHealthUnavailable,
+    StartupFinalizationPending,
     StorageAndIamUnavailable,
     StorageAndLockUnavailable,
     IamAndLockUnavailable,
@@ -62,6 +63,7 @@ impl ReadinessDegradedReason {
             ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled",
             ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout",
             ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable",
+            ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending",
             ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable",
             ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable",
             ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable",

rustfs/src/site_replication/mod.rs

@@ -34,6 +34,9 @@ pub(crate) mod retry;
 pub(crate) mod state;
 pub(crate) mod transport;
 
+#[cfg(test)]
+mod tests;
+
 pub(crate) use self::hooks::*;
 pub(crate) use self::repair::*;
 pub(crate) use self::retry::*;

rustfs/src/storage/access.rs

@@ -16,8 +16,10 @@ use super::ObjectOptions;
 use super::ecfs::FS;
 use super::{ECStore, PolicySys, ReplicationStatusType, StorageError, get_lock_acquire_timeout, is_err_bucket_not_found};
 use crate::auth::{
-    check_key_valid_with_context, get_condition_values_with_client_info, get_condition_values_with_query_and_client_info,
-    get_session_token,
+    AuthType, RUSTFS_MAX_CONTENT_LENGTH_QUERY, RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY, VerifiedPresignedRequest,
+    VerifiedSigV4Request, check_key_valid_with_context, get_condition_values_with_client_info,
+    get_condition_values_with_query_and_client_info, get_request_auth_type_with_query, get_session_token,
+    parse_presigned_multipart_max_total_object_size, parse_presigned_put_max_content_length,
 };
 use crate::error::ApiError;
 use crate::license::license_check;
@@ -1770,9 +1772,41 @@ impl S3Access for FS {
 
         // Publish this server's context slot so downstream data-plane handlers
         // resolve the same store (backlog#1052 S6).
-        let ext = cx.extensions_mut();
-        ext.insert(self.server_ctx().clone());
-        ext.insert(req_info);
+        let auth_type = get_request_auth_type_with_query(cx.headers(), cx.uri().query());
+        let verified_presigned = matches!(auth_type, AuthType::Presigned);
+        let verified_sigv4 = matches!(auth_type, AuthType::Presigned | AuthType::Signed);
+        {
+            let ext = cx.extensions_mut();
+            ext.insert(self.server_ctx().clone());
+            ext.insert(req_info);
+            if verified_presigned {
+                ext.insert(VerifiedPresignedRequest);
+            }
+            if verified_sigv4 {
+                ext.insert(VerifiedSigV4Request);
+            }
+        }
+
+        // The size capability is intentionally scoped to the single-object
+        // PutObject operation. Validate this at the operation-aware access
+        // boundary so unsupported GET/HEAD/DELETE/bucket routes cannot silently
+        // ignore a signed capability query.
+        if parse_presigned_put_max_content_length(cx.headers(), cx.uri().query(), verified_presigned)?.is_some()
+            && cx.s3_op().name() != "PutObject"
+        {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("{RUSTFS_MAX_CONTENT_LENGTH_QUERY} is only supported for presigned PutObject"),
+            ));
+        }
+        if parse_presigned_multipart_max_total_object_size(cx.headers(), cx.uri().query(), verified_sigv4)?.is_some()
+            && cx.s3_op().name() != "CreateMultipartUpload"
+        {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                format!("{RUSTFS_MAX_TOTAL_OBJECT_SIZE_QUERY} is only supported for CreateMultipartUpload"),
+            ));
+        }
         license_check().map_err(|er| match er.kind() {
             std::io::ErrorKind::PermissionDenied => s3_error!(AccessDenied, "{er}"),
             _ => {

rustfs/src/storage/concurrency/manager.rs

@@ -12,7 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-//! Concurrency manager for coordinating concurrent GetObject requests.
+//! Concurrency manager for coordinating concurrent GetObject and PutObject requests.
 
 use super::io_schedule::{
     IoLoadLevel, IoLoadMetrics, IoPriority, IoPriorityQueue, IoPriorityQueueConfig, IoQueueStatus, IoSchedulerConfig, IoStrategy,
@@ -34,6 +34,8 @@ use std::time::Duration;
 use tokio::sync::Semaphore;
 use tracing::debug;
 
+const DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX: usize = 32;
+
 /// Global concurrency manager instance
 pub(crate) static CONCURRENCY_MANAGER: LazyLock<ConcurrencyManager> = LazyLock::new(ConcurrencyManager::new);
 
@@ -65,11 +67,8 @@ pub struct ConcurrencyManager {
     bandwidth_monitor: Arc<Mutex<BandwidthMonitor>>,
     /// Metrics collector for I/O latency tracking (P50, P95, P99)
     metrics_collector: Arc<MetricsCollector>,
-    /// Experimental fixed-count foreground PutObject admission gate.
-    put_admission_semaphore: Arc<Semaphore>,
-    put_admission_enabled: bool,
-    put_admission_limit: usize,
-    put_admission_wait_timeout: Duration,
+    /// Foreground PutObject admission policy, resolved once at startup.
+    put_admission_policy: PutAdmissionPolicy,
 }
 
 impl std::fmt::Debug for ConcurrencyManager {
@@ -127,10 +126,201 @@ pub enum PutObjectAdmission {
     /// Request is admitted and must hold the permit until the store write
     /// returns or the request fails before mutation.
     Admitted(tokio::sync::OwnedSemaphorePermit),
-    /// The fixed-count gate stayed full until the configured wait timeout.
+    /// The selected foreground PUT admission gate stayed full until the configured wait timeout.
     Rejected,
 }
 
+#[derive(Clone)]
+struct PutAdmissionGate {
+    semaphore: Arc<Semaphore>,
+    limit: usize,
+    wait_timeout: Duration,
+}
+
+impl PutAdmissionGate {
+    fn new(limit: usize, wait_timeout: Duration) -> Self {
+        Self {
+            semaphore: Arc::new(Semaphore::new(limit)),
+            limit,
+            wait_timeout,
+        }
+    }
+
+    fn active(&self) -> usize {
+        self.limit.saturating_sub(self.semaphore.available_permits())
+    }
+
+    async fn admit(&self) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
+        if self.wait_timeout.is_zero() {
+            return Ok(match self.semaphore.clone().try_acquire_owned() {
+                Ok(permit) => PutObjectAdmission::Admitted(permit),
+                Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected,
+                Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected,
+            });
+        }
+
+        match tokio::time::timeout(self.wait_timeout, self.semaphore.clone().acquire_owned()).await {
+            Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)),
+            Err(_) => Ok(PutObjectAdmission::Rejected),
+        }
+    }
+}
+
+#[derive(Clone)]
+enum PutAdmissionPolicy {
+    /// Strict admission was explicitly enabled with limit `0`.
+    Disabled,
+    /// No hard PUT gate is configured; foreground write snapshots use the
+    /// existing active request counter as a soft pressure signal.
+    LegacyCounterOnly,
+    /// Explicit all-PUT admission gate.
+    Strict(PutAdmissionGate),
+    /// Default large/unknown-size PUT admission gate.
+    Large { gate: PutAdmissionGate, min_size_bytes: usize },
+}
+
+impl PutAdmissionPolicy {
+    fn from_env(max_disk_reads: usize) -> Self {
+        let strict_enabled = rustfs_utils::get_env_bool(
+            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE,
+            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE,
+        );
+        if strict_enabled {
+            let strict_limit = rustfs_utils::get_env_usize(
+                rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT,
+                rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT,
+            );
+            let strict_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
+                rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+                rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+            ));
+            return if strict_limit == 0 {
+                Self::Disabled
+            } else {
+                Self::Strict(PutAdmissionGate::new(strict_limit, strict_wait_timeout))
+            };
+        }
+
+        let large_enabled = rustfs_utils::get_env_bool(
+            rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_ENABLE,
+        );
+        if !large_enabled {
+            return Self::LegacyCounterOnly;
+        }
+
+        let large_limit = derive_large_put_admission_limit(
+            rustfs_utils::get_env_usize(
+                rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT,
+                rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_LIMIT,
+            ),
+            max_disk_reads,
+        );
+        let min_size_bytes = rustfs_utils::get_env_usize(
+            rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+        );
+        let wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
+            rustfs_config::ENV_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
+        ));
+
+        Self::Large {
+            gate: PutAdmissionGate::new(large_limit, wait_timeout),
+            min_size_bytes,
+        }
+    }
+
+    #[cfg(test)]
+    fn strict_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self {
+        if enabled {
+            if limit == 0 {
+                Self::Disabled
+            } else {
+                Self::Strict(PutAdmissionGate::new(limit, wait_timeout))
+            }
+        } else {
+            Self::LegacyCounterOnly
+        }
+    }
+
+    #[cfg(test)]
+    fn large_for_test(enabled: bool, limit: usize, min_size_bytes: usize, wait_timeout: Duration) -> Self {
+        if enabled && limit > 0 {
+            Self::Large {
+                gate: PutAdmissionGate::new(limit, wait_timeout),
+                min_size_bytes,
+            }
+        } else {
+            Self::LegacyCounterOnly
+        }
+    }
+
+    async fn admit(&self, size: i64) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
+        match self {
+            Self::Disabled | Self::LegacyCounterOnly => Ok(PutObjectAdmission::Disabled),
+            Self::Strict(gate) => gate.admit().await,
+            Self::Large { gate, min_size_bytes } if should_gate_large_put(size, *min_size_bytes) => gate.admit().await,
+            Self::Large { .. } => Ok(PutObjectAdmission::Disabled),
+        }
+    }
+
+    fn snapshot(&self, legacy_limit: usize) -> WorkloadAdmissionSnapshot {
+        match self {
+            Self::Disabled => put_admission_snapshot(0, 0, None),
+            Self::LegacyCounterOnly => put_admission_snapshot(PutObjectGuard::concurrent_count(), legacy_limit, None),
+            Self::Strict(gate) => {
+                put_admission_snapshot(gate.active(), gate.limit, Some("foreground write admission permits exhausted"))
+            }
+            Self::Large { gate, .. } => {
+                put_admission_snapshot(gate.active(), gate.limit, Some("large foreground write admission permits exhausted"))
+            }
+        }
+    }
+}
+
+fn put_admission_snapshot(active: usize, limit: usize, hard_gate_reason: Option<&'static str>) -> WorkloadAdmissionSnapshot {
+    let state = if limit == 0 {
+        AdmissionState::Disabled
+    } else if active >= limit {
+        AdmissionState::Saturated
+    } else {
+        AdmissionState::Open
+    };
+
+    let admission =
+        WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));
+
+    match state {
+        AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
+        AdmissionState::Saturated => {
+            admission.with_reason(hard_gate_reason.unwrap_or("foreground write concurrency reached local pressure limit"))
+        }
+        _ => admission,
+    }
+}
+
+fn derive_large_put_admission_limit(configured_limit: usize, max_disk_reads: usize) -> usize {
+    if configured_limit > 0 {
+        return configured_limit;
+    }
+
+    let scheduler_base = if max_disk_reads == 0 {
+        rustfs_config::DEFAULT_OBJECT_MAX_CONCURRENT_DISK_READS
+    } else {
+        max_disk_reads
+    };
+    scheduler_base.div_ceil(2).clamp(1, DERIVED_LARGE_PUT_ADMISSION_LIMIT_MAX)
+}
+
+fn should_gate_large_put(size: i64, min_size_bytes: usize) -> bool {
+    if min_size_bytes == 0 || size < 0 {
+        return true;
+    }
+
+    usize::try_from(size).is_ok_and(|size| size >= min_size_bytes)
+}
+
 impl ConcurrencyManager {
     /// Create a new concurrency manager with default settings
     ///
@@ -177,18 +367,7 @@ impl ConcurrencyManager {
         // Initialize metrics collector for I/O latency tracking
         // Keep 1000 samples for P95/P99 calculation
         let metrics_collector = Arc::new(MetricsCollector::new(performance_metrics, 1000));
-        let put_admission_enabled = rustfs_utils::get_env_bool(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_ENABLE,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE,
-        );
-        let put_admission_limit = rustfs_utils::get_env_usize(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_LIMIT,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT,
-        );
-        let put_admission_wait_timeout = Duration::from_millis(rustfs_utils::get_env_u64(
-            rustfs_config::ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-            rustfs_config::DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS,
-        ));
+        let put_admission_policy = PutAdmissionPolicy::from_env(max_disk_reads);
 
         // Build queue config directly from scheduler config.
         let queue_config = IoPriorityQueueConfig::from_scheduler_config(&scheduler_config);
@@ -204,10 +383,7 @@ impl ConcurrencyManager {
             pattern_detector,
             bandwidth_monitor,
             metrics_collector,
-            put_admission_semaphore: Arc::new(Semaphore::new(if put_admission_enabled { put_admission_limit } else { 0 })),
-            put_admission_enabled,
-            put_admission_limit,
-            put_admission_wait_timeout,
+            put_admission_policy,
         }
     }
 
@@ -234,10 +410,19 @@ impl ConcurrencyManager {
     #[cfg(test)]
     pub(crate) fn with_put_admission_for_test(enabled: bool, limit: usize, wait_timeout: Duration) -> Self {
         let mut manager = Self::new();
-        manager.put_admission_semaphore = Arc::new(Semaphore::new(if enabled { limit } else { 0 }));
-        manager.put_admission_enabled = enabled;
-        manager.put_admission_limit = limit;
-        manager.put_admission_wait_timeout = wait_timeout;
+        manager.put_admission_policy = PutAdmissionPolicy::strict_for_test(enabled, limit, wait_timeout);
+        manager
+    }
+
+    #[cfg(test)]
+    pub(crate) fn with_large_put_admission_for_test(
+        enabled: bool,
+        limit: usize,
+        min_size_bytes: usize,
+        wait_timeout: Duration,
+    ) -> Self {
+        let mut manager = Self::new();
+        manager.put_admission_policy = PutAdmissionPolicy::large_for_test(enabled, limit, min_size_bytes, wait_timeout);
         manager
     }
 
@@ -326,30 +511,13 @@ impl ConcurrencyManager {
         }
     }
 
-    /// Admit a foreground PutObject request under the experimental fixed-count gate.
+    /// Admit a foreground PutObject request under the configured write gate.
     ///
-    /// The default-off path returns [`PutObjectAdmission::Disabled`] without
-    /// touching the semaphore, preserving legacy behavior. When enabled, the
-    /// permit must be acquired before body ingest and held until the store write
-    /// returns, so saturated foreground writes can fail with `SlowDown` before
-    /// creating visible side effects.
-    pub async fn admit_put_object(&self) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
-        if !self.put_admission_enabled || self.put_admission_limit == 0 {
-            return Ok(PutObjectAdmission::Disabled);
-        }
-
-        if self.put_admission_wait_timeout.is_zero() {
-            return Ok(match self.put_admission_semaphore.clone().try_acquire_owned() {
-                Ok(permit) => PutObjectAdmission::Admitted(permit),
-                Err(tokio::sync::TryAcquireError::NoPermits) => PutObjectAdmission::Rejected,
-                Err(tokio::sync::TryAcquireError::Closed) => PutObjectAdmission::Rejected,
-            });
-        }
-
-        match tokio::time::timeout(self.put_admission_wait_timeout, self.put_admission_semaphore.clone().acquire_owned()).await {
-            Ok(permit) => Ok(PutObjectAdmission::Admitted(permit?)),
-            Err(_) => Ok(PutObjectAdmission::Rejected),
-        }
+    /// The strict experimental gate applies to every PUT only when explicitly
+    /// enabled. Otherwise the default-on large-object gate protects sustained
+    /// erasure/RPC pressure while keeping small PUTs on the legacy path.
+    pub async fn admit_put_object(&self, size: i64) -> Result<PutObjectAdmission, tokio::sync::AcquireError> {
+        self.put_admission_policy.admit(size).await
     }
 
     // ============================================
@@ -760,35 +928,7 @@ impl ConcurrencyManager {
 
     /// Get a read-only workload admission snapshot for foreground writes.
     pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot {
-        let (active, limit, hard_gate_enabled) = if self.put_admission_enabled && self.put_admission_limit > 0 {
-            (
-                self.put_admission_limit
-                    .saturating_sub(self.put_admission_semaphore.available_permits()),
-                self.put_admission_limit,
-                true,
-            )
-        } else {
-            (PutObjectGuard::concurrent_count(), self.scheduler_config.max_concurrent_reads, false)
-        };
-        let state = if limit == 0 {
-            AdmissionState::Disabled
-        } else if active >= limit {
-            AdmissionState::Saturated
-        } else {
-            AdmissionState::Open
-        };
-
-        let admission =
-            WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit));
-
-        match state {
-            AdmissionState::Disabled => admission.with_reason("foreground write admission disabled"),
-            AdmissionState::Saturated if hard_gate_enabled => {
-                admission.with_reason("foreground write admission permits exhausted")
-            }
-            AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"),
-            _ => admission,
-        }
+        self.put_admission_policy.snapshot(self.scheduler_config.max_concurrent_reads)
     }
 
     /// Get a read-only workload admission registry snapshot for local storage concurrency.
@@ -862,7 +1002,7 @@ impl Default for ConcurrencyManager {
 mod integration_tests {
     use super::super::io_schedule::{IoLoadLevel, IoPriority};
     use super::super::request_guard::GetObjectGuard;
-    use super::{ConcurrencyManager, PutObjectAdmission};
+    use super::{ConcurrencyManager, PutObjectAdmission, derive_large_put_admission_limit};
     use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
     use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
     use rustfs_io_core::io_profile::{AccessPattern, StorageMedia};
@@ -941,7 +1081,7 @@ mod integration_tests {
     #[serial]
     async fn test_concurrency_manager_workload_admission_snapshot_tracks_put_requests() {
         crate::storage::concurrency::reset_active_put_requests();
-        let manager = ConcurrencyManager::new();
+        let manager = ConcurrencyManager::with_put_admission_for_test(false, 0, Duration::ZERO);
         let initial = manager.put_object_admission_snapshot();
 
         assert_eq!(initial.class, WorkloadClass::ForegroundWrite);
@@ -965,26 +1105,42 @@ mod integration_tests {
         let manager = ConcurrencyManager::with_put_admission_for_test(false, 1, Duration::ZERO);
 
         let admission = manager
-            .admit_put_object()
+            .admit_put_object(1024)
             .await
             .expect("disabled put admission must not close");
 
         assert!(matches!(admission, PutObjectAdmission::Disabled));
-        assert_eq!(manager.put_admission_semaphore.available_permits(), 0);
         assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Open);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_strict_put_admission_zero_limit_disables_large_gate() {
+        let manager = ConcurrencyManager::with_put_admission_for_test(true, 0, Duration::ZERO);
+
+        let admission = manager
+            .admit_put_object(32 * 1024 * 1024)
+            .await
+            .expect("strict zero-limit put admission must not close");
+
+        assert!(matches!(admission, PutObjectAdmission::Disabled));
+        assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Disabled);
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_concurrency_manager_put_admission_rejects_when_limit_full() {
         let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
 
-        let first = manager.admit_put_object().await.expect("first put admission should acquire");
+        let first = manager
+            .admit_put_object(1024)
+            .await
+            .expect("first put admission should acquire");
         assert!(matches!(first, PutObjectAdmission::Admitted(_)));
         assert_eq!(manager.put_object_admission_snapshot().state, AdmissionState::Saturated);
 
         let second = manager
-            .admit_put_object()
+            .admit_put_object(1024)
             .await
             .expect("full put admission gate should reject, not close");
         assert!(matches!(second, PutObjectAdmission::Rejected));
@@ -995,11 +1151,14 @@ mod integration_tests {
     async fn test_concurrency_manager_put_admission_reuses_released_permit() {
         let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO);
 
-        let first = manager.admit_put_object().await.expect("first put admission should acquire");
+        let first = manager
+            .admit_put_object(1024)
+            .await
+            .expect("first put admission should acquire");
         drop(first);
 
         let second = manager
-            .admit_put_object()
+            .admit_put_object(1024)
             .await
             .expect("released put admission permit should be reusable");
         assert!(matches!(second, PutObjectAdmission::Admitted(_)));
@@ -1009,10 +1168,13 @@ mod integration_tests {
     #[serial]
     async fn test_concurrency_manager_put_admission_wait_timeout_rejects() {
         let manager = ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(5));
-        let held = manager.admit_put_object().await.expect("first put admission should acquire");
+        let held = manager
+            .admit_put_object(1024)
+            .await
+            .expect("first put admission should acquire");
         let waiter_manager = manager.clone();
 
-        let waiter = tokio::spawn(async move { waiter_manager.admit_put_object().await });
+        let waiter = tokio::spawn(async move { waiter_manager.admit_put_object(1024).await });
         tokio::task::yield_now().await;
         tokio::time::advance(Duration::from_secs(5)).await;
 
@@ -1024,6 +1186,85 @@ mod integration_tests {
         drop(held);
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_large_put_admission_bypasses_small_puts() {
+        let min_size = rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES;
+        let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, min_size, Duration::ZERO);
+
+        let held = manager
+            .admit_put_object(min_size as i64)
+            .await
+            .expect("large put admission should acquire");
+        assert!(matches!(held, PutObjectAdmission::Admitted(_)));
+
+        let small = manager
+            .admit_put_object((min_size - 1) as i64)
+            .await
+            .expect("small put should bypass large admission");
+        assert!(matches!(small, PutObjectAdmission::Disabled));
+
+        let large = manager
+            .admit_put_object(min_size as i64)
+            .await
+            .expect("second large put should reject when the gate is full");
+        assert!(matches!(large, PutObjectAdmission::Rejected));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_large_put_admission_gates_unknown_size() {
+        let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 1, 32 * 1024 * 1024, Duration::ZERO);
+
+        let held = manager
+            .admit_put_object(-1)
+            .await
+            .expect("unknown-size put admission should acquire");
+        assert!(matches!(held, PutObjectAdmission::Admitted(_)));
+
+        let second = manager
+            .admit_put_object(-1)
+            .await
+            .expect("unknown-size put admission should reject when the gate is full");
+        assert!(matches!(second, PutObjectAdmission::Rejected));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_large_put_snapshot_tracks_gate() {
+        let manager = ConcurrencyManager::with_large_put_admission_for_test(true, 2, 32 * 1024 * 1024, Duration::ZERO);
+        let first = manager
+            .admit_put_object(32 * 1024 * 1024)
+            .await
+            .expect("first large put admission should acquire");
+        let initial = manager.put_object_admission_snapshot();
+
+        assert_eq!(initial.class, WorkloadClass::ForegroundWrite);
+        assert_eq!(initial.state, AdmissionState::Open);
+        assert_eq!(initial.active, Some(1));
+        assert_eq!(initial.limit, Some(2));
+
+        let second = manager
+            .admit_put_object(32 * 1024 * 1024)
+            .await
+            .expect("second large put admission should acquire");
+        let saturated = manager.put_object_admission_snapshot();
+
+        assert_eq!(saturated.state, AdmissionState::Saturated);
+        assert_eq!(saturated.active, Some(2));
+        assert_eq!(saturated.limit, Some(2));
+        drop((first, second));
+    }
+
+    #[test]
+    fn test_concurrency_manager_derives_large_put_admission_limit_from_scheduler_cap() {
+        assert_eq!(derive_large_put_admission_limit(7, 64), 7);
+        assert_eq!(derive_large_put_admission_limit(0, 64), 32);
+        assert_eq!(derive_large_put_admission_limit(0, 8), 4);
+        assert_eq!(derive_large_put_admission_limit(0, 1), 1);
+        assert_eq!(derive_large_put_admission_limit(0, 0), 32);
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() {

rustfs/src/storage/sse.rs

@@ -961,9 +961,9 @@ fn sse_kms_key_policy_enforced(principal: Option<&SseKmsPrincipal>) -> bool {
 
 /// Report the configured SSE-KMS authorization mode once, at startup.
 ///
-/// The disabled case warns rather than logs: it is the compatibility default for this
-/// release only, and operators need the lead time to grant the kms actions before the
-/// default flips.
+/// The disabled case warns rather than logs: while enforcement is off, any identity
+/// allowed to write an object can encrypt it under any key, and operators should hear
+/// about that even though disabled is the long-term default.
 pub(crate) fn log_sse_kms_key_policy_mode() {
     if sse_kms_key_policy_enforced(None) {
         tracing::info!(
@@ -984,7 +984,7 @@ pub(crate) fn log_sse_kms_key_policy_mode() {
         "SSE-KMS requests are not authorized against the KMS key they name; any identity allowed to \
          write an object may encrypt it under any key, and any identity allowed to read it may have it \
          decrypted. Grant kms:GenerateDataKey and kms:Decrypt on the keys your workloads use, then set \
-         {ENV_RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY}=true. A later release defaults this to enabled."
+         {ENV_RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY}=true."
     );
 }
 
@@ -1027,6 +1027,25 @@ async fn authorize_sse_kms_key(
         "Principal is not authorized for the KMS key resolved for this request"
     );
 
+    // One warn per process, not per request: anonymous denials are driven by
+    // unauthenticated traffic, so a per-request warn would let anyone flood the
+    // log. Per-request detail stays on the audit entry and the debug event above.
+    if principal.account.is_empty() {
+        static ANONYMOUS_DENIAL_WARNED: std::sync::Once = std::sync::Once::new();
+        ANONYMOUS_DENIAL_WARNED.call_once(|| {
+            tracing::warn!(
+                component = LOG_COMPONENT_STORAGE,
+                subsystem = LOG_SUBSYSTEM_SSE,
+                event = "sse_kms_anonymous_key_authorization_denied",
+                action = ?action,
+                "Anonymous requests are being denied by SSE-KMS per-key authorization: anonymous \
+                 callers hold no kms grants, so a public bucket serving SSE-KMS objects is \
+                 incompatible with {ENV_RUSTFS_KMS_ENFORCE_SSE_KEY_POLICY}=true. Reported once per \
+                 process; per-request denials are on audit entries and at debug level."
+            );
+        });
+    }
+
     Err(ApiError {
         code: S3ErrorCode::AccessDenied,
         message: "Access Denied".to_string(),

rustfs/src/storage_api.rs

@@ -227,6 +227,8 @@ pub(crate) mod site_replication {
         BUCKET_REPLICATION_CONFIG, BUCKET_TARGETS_FILE, BUCKET_VERSIONING_CONFIG, BucketMetadata,
     };
 
+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::ecstore_bucket::replication::merge_incoming_replication_config;
     pub(crate) use crate::storage::storage_api::ecstore_bucket::replication::{
         OperatorRuleContract, assign_site_replication_rule_priorities, is_site_replication_role,
         replication_target_arn_deployment_id, site_replication_rule_deployment_id,
@@ -238,6 +240,8 @@ pub(crate) mod site_replication {
     pub(crate) use crate::storage::storage_api::ecstore_bucket::versioning::VersioningApi;
     #[cfg(test)]
     pub(crate) use crate::storage::storage_api::ecstore_config::com::save_config;
+    #[cfg(test)]
+    pub(crate) use crate::storage::storage_api::{Endpoint, Endpoints, PoolEndpoints};
 
     pub(crate) use crate::storage::storage_api::{
         ECStore, EndpointServerPools, StorageError, delete_config_no_lock, lock_bucket_targets_metadata, read_config,
@@ -260,6 +264,8 @@ pub(crate) mod site_replication {
             LifecycleRule, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule,
             ReplicationRuleStatus, SourceSelectionCriteria, VersioningConfiguration,
         };
+        #[cfg(test)]
+        pub(crate) use s3s::dto::{ExpirationStatus, LifecycleExpiration, Timestamp, Transition, TransitionStorageClass};
         pub(crate) use s3s::{Body, S3Error, S3ErrorCode, S3Response, S3Result, s3_error};
     }
 }

scripts/check_embedded_secrets.sh

@@ -88,7 +88,7 @@ PATTERNS=(
 # the guard fire again. Entries that stop matching anything are reported as
 # stale, so the list cannot decay into a blanket exclusion.
 #
-# 1-2: rustfs/src/admin/handlers/site_replication.rs negative fixtures for
+# 1-2: rustfs/src/site_replication/tests.rs negative fixtures for
 #      `validate_peer_connection_inner`, which must reject a private key
 #      submitted where a peer CA certificate is expected. Asserting on the
 #      rejection requires the header in the input; the key bodies are the

scripts/check_logging_guardrails.sh

@@ -37,6 +37,7 @@ checked_files=(
   "rustfs/src/site_replication/retry.rs"
   "rustfs/src/site_replication/repair.rs"
   "rustfs/src/site_replication/hooks.rs"
+  "rustfs/src/site_replication/tests.rs"
   "rustfs/src/admin/handlers/group.rs"
   "rustfs/src/admin/handlers/quota.rs"
   "rustfs/src/admin/handlers/rebalance.rs"

scripts/security/check_preview_release_workflow.sh

@@ -140,6 +140,14 @@ done
 latest_guard="startsWith(github.ref, 'refs/tags/') && (needs.build-check.outputs.build_type == 'release' || needs.build-check.outputs.build_type == 'prerelease')"
 require_job_if "$build_workflow" "update-latest-version" "    if: $latest_guard"
 require_line "$build_workflow" "    needs: [ build-check, publish-release ]" "latest update must follow release publication"
+
+# Preview releases are internal validation artifacts: once the deliverable
+# release is published they are deleted, while their tags stay behind.
+require_job_if "$build_workflow" "cleanup-preview-releases" "    if: $latest_guard"
+require_line "$build_workflow" "            gh release delete \"\$preview_tag\" --yes" "preview release cleanup after publication"
+require_line "$build_workflow" "              | select(.tag_name | startswith(\$tag + \"-preview.\"))" "cleanup must match the target's own preview tags"
+require_line "$build_workflow" "              | select(.tag_name | ltrimstr(\$tag + \"-preview.\") | test(\"^[0-9]+\$\"))" "cleanup must match a numeric preview iteration"
+require_absent "$build_workflow" "--cleanup-tag" "preview tags must survive their release cleanup"
 require_line "$build_workflow" "          TARGET_COMMITISH=\$(git rev-parse --verify \"refs/tags/\${TAG}^{commit}\")" "release target commit resolution"
 require_line "$build_workflow" "          ./scripts/release/create_or_update_release.sh \\" "managed release creation"
 require_absent "$build_workflow" "git tag -l --format='%(contents)'" "annotated tag messages must not become release notes"

scripts/test/rustfs_heal_test.md

@@ -0,0 +1,104 @@
+# RustFS Heal Test
+
+Node-outage heal test driven by
+[`scripts/test/rustfs_heal_test.sh`](rustfs_heal_test.sh), based on the
+Obsidian note "RustFS Heal 测试步骤". Uses the same 3-node test environment as
+the pool expansion test (`vm000 vm001 vm002`).
+
+All status checks talk to the RustFS admin API directly (SigV4-signed,
+`jq` assertions), no `rc` required.
+
+## What it does
+
+1. Downloads the `.deb` package on all nodes (release tag or a direct URL such
+   as the nightly/R2 package).
+2. Installs it, writes the 3x4 config
+   (`http://rustfs-node{1...3}:9000/data/rustfs{1...4}/mnmd`), starts all
+   three nodes simultaneously, verifies the cluster is up.
+3. Writes data with `warp` while monitoring disk usage on the surviving nodes
+   (`df -B1G | grep /data/rustfs`):
+   - when both surviving nodes reach `STOP_NODE_AT_GB` (default 15 GiB), stop
+     the outage node (`vm002`, `OUTAGE_NODE_INDEX=2`);
+   - keep writing until both surviving nodes reach `WARP_STOP_AT_GB`
+     (default 40 GiB), then stop warp.
+4. Restarts the outage node.
+5. Starts cluster heal: `POST /rustfs/admin/v3/heal/` with body
+   `{"recursive":true}` (retried, returns a `clientToken`).
+6. Monitors the heal task via `POST /rustfs/admin/v3/heal/?clientToken=<token>`
+   until the summary is a terminal success (`finished`/`completed`),
+   `objects_failed == 0`, **and** the outage node's disk usage reaches
+   `HEAL_TARGET_GB` (default 40 GiB).
+7. Result analysis: heal stats (scanned/healed/failed), per-node disk usage,
+   pass/fail verdict.
+
+Success requires **both** the heal API completion (the server's scan/repair
+verdict) and the outage node's disk reaching the target.
+
+## Self-hosted runner prerequisites
+
+- Register the admin host (e.g. `heal`) as a runner with the
+  `smoke-testing` label.
+- Install `jq`, `openssl`, `curl` and `warp` on the runner. `rc` is **not**
+  required.
+- The runner user must be able to SSH to `vm000/vm001/vm002` without a
+  password prompt; nodes need passwordless `sudo` for the SSH user and
+  resolvable `rustfs-node*` hostnames.
+- Admin API credentials need the `admin:server-info`, `admin:heal` and
+  `admin:rebalance` actions.
+
+## Configuration
+
+Same repository secrets/variables as the pool expansion workflow:
+
+| Kind   | Name                  | Purpose                                        |
+| ------ | --------------------- | ---------------------------------------------- |
+| Secret | `RUSTFS_ACCESS_KEY`   | RustFS access key (default `rustfs@test`)      |
+| Secret | `RUSTFS_SECRET_KEY`   | RustFS secret key (default `rustfs@test`)      |
+| Var    | `RUSTFS_API_ENDPOINT` | Admin API endpoint, e.g. `http://127.0.0.1:9000` (`RUSTFS_RC_ENDPOINT` fallback) |
+| Var    | `RUSTFS_NODES`        | `vm000 vm001 vm002`                            |
+| Var    | `RUSTFS_SSH_USER`     | `azureuser`                                    |
+| Var    | `RUSTFS_NIGHTLY_PACKAGE_URL` | Default nightly deb URL (defaults to the R2 `latest` alias) |
+
+## Workflow inputs
+
+| Input            | Default | Meaning                                   |
+| ---------------- | ------- | ----------------------------------------- |
+| `package_url`    | nightly | Direct `.deb` URL; empty = latest nightly |
+| `stop_node_gb`   | `15`    | Stop outage node at N GiB on survivors    |
+| `warp_stop_gb`   | `40`    | Stop warp at N GiB on survivors           |
+| `heal_target_gb` | `40`    | Outage node must reach N GiB after heal   |
+| `cleanup_before` | `true`  | Reset nodes before the test               |
+| `cleanup_after`  | `true`  | Reset nodes after the test                |
+
+> ⚠️ `--reset` purges the `rustfs` package and deletes the data directories on
+> all nodes. Only run against a dedicated test environment.
+
+## Manual usage
+
+```bash
+./scripts/test/rustfs_heal_test.sh --all -y \
+  --package-url https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb \
+  --endpoint http://127.0.0.1:9000
+
+./scripts/test/rustfs_heal_test.sh --steps 5,6,7
+./scripts/test/rustfs_heal_test.sh --reset -y
+```
+
+## Known issues
+
+- Nightly builds gate pool/rebalance activation on a live fleet capability
+  proof (rustfs/backlog#2031); the script retries heal/rebalance starts and
+  prints a hint when the signature appears.
+- The cluster-level `GET /rustfs/admin/v3/background-heal/status` aggregator
+  returns 501 in the single-pool 3x4 topology (no notification system), so the
+  script monitors the started heal task via its `clientToken` instead.
+- The heal task may report `progress: null` while running; the script logs
+  this as evidence (rustfs/backlog#2035) rather than coercing it to zero, and
+  reads the canonical camelCase progress fields
+  (`objectsScanned`/`objectsHealed`/`objectsFailed`/`progressPercentage`) with
+  a snake_case fallback.
+- The server-side per-task heal timeout defaults to 5 minutes; the script
+  writes `RUSTFS_HEAL_TASK_TIMEOUT_SECS=21600` (6h) into the node config so a
+  multi-tens-of-GiB heal can finish. The background scanner is disabled
+  (`RUSTFS_HEAL_AUTO_HEAL_ENABLE=false`) so the explicit heal is the only
+  repair mechanism and the outage effect stays observable.

scripts/test/rustfs_pool_expand.sh

@@ -93,6 +93,9 @@ WARP_BUCKET="test-10mb"
 WARP_OBJ_SIZE="100MiB"
 WARP_CONCURRENT=32
 WARP_DURATION="5m"
+# Warp log path; empty = auto-created unique temp file (the runner user may
+# not be able to write a shared /tmp path owned by another user).
+WARP_LOG_FILE="${RUSTFS_WARP_LOG_FILE:-}"
 STORAGE_THRESHOLD=85            # stop writing when usage reaches N% (note suggests 80-85)
 POLL_INTERVAL=30                # status polling interval (seconds)
 
@@ -102,6 +105,8 @@ DECOMMISSION_TIMEOUT=86400
 SERVICE_TIMEOUT=300
 DECOMMISSION_RETRIES=3          # auto clear+retry attempts after a failed decommission
 DECOMMISSION_RETRY_DELAY=30     # delay between retries (seconds)
+REBALANCE_START_RETRIES=6       # rebalance start retries (fleet proof may take ~10-20s after a topology change)
+REBALANCE_START_RETRY_DELAY=20  # delay between rebalance start retries (seconds)
 
 # Pool to decommission (zero-based; 0 in the note)
 DECOMMISSION_POOL_ID=0
@@ -325,9 +330,45 @@ wait_service_active() {
     sleep 5
     waited=$((waited + 5))
   done
+  diagnose_node_start_failure "${node}"
   die "${node}: timed out waiting for ${RUSTFS_SERVICE} (${SERVICE_TIMEOUT}s)"
 }
 
+# Known server-side issues the test can hit. Format:
+#   "<error signature>|<tracking>|<hint>"
+KNOWN_SERVER_ISSUES=(
+  "pool activation requires a live fleet capability proof|rustfs/backlog#2031|server-side cold-start recovery is covered by this PR; if this appears, collect node journals and treat it as a regression"
+)
+
+# Print a hint when $1 matches a known server-side issue signature.
+hint_server_issue() {
+  local text="$1" entry sig tracking hint
+  for entry in "${KNOWN_SERVER_ISSUES[@]}"; do
+    sig="${entry%%|*}"
+    tracking="${entry#*|}"
+    hint="${tracking#*|}"
+    tracking="${tracking%%|*}"
+    if printf '%s' "${text}" | grep -qiF "${sig}"; then
+      printf '\033[1;33m[KNOWN SERVER ISSUE]\033[0m %s (%s): %s\n' "${sig}" "${tracking}" "${hint}" >&2
+      return 0
+    fi
+  done
+  return 1
+}
+
+# Fetch the journal tail from a node whose service failed to start and
+# annotate known server-side issues.
+diagnose_node_start_failure() {
+  local node="$1" journal
+  if ! journal="$(ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
+    "SUDO=\"\"; [ \"\$(id -u)\" -ne 0 ] && SUDO=\"sudo -n\"; \${SUDO} journalctl -u ${RUSTFS_SERVICE} --no-pager -n 60 2>/dev/null || true")"; then
+    journal="unable to collect journal (SSH command failed)"
+  fi
+  printf '%s\n' "--- ${node}: ${RUSTFS_SERVICE} journal (last 60 lines) ---" >&2
+  printf '%s\n' "${journal}" >&2
+  hint_server_issue "${journal}" || true
+}
+
 # Generate the /etc/default/rustfs content
 rustfs_config_body() {
   local volumes="$1"
@@ -406,9 +447,13 @@ service_action() {
   local action="$1" node="$2"
   log "${node}: systemctl ${action} ${RUSTFS_SERVICE}"
   if [ "${DRY_RUN}" -eq 1 ]; then return 0; fi
-  ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
-    "if [ \"\$(id -u)\" -ne 0 ]; then sudo -n systemctl ${action} ${RUSTFS_SERVICE}; else systemctl ${action} ${RUSTFS_SERVICE}; fi" \
-    || die "${node}: systemctl ${action} failed"
+  if ! ssh "${SSH_OPTS[@]}" "${SSH_USER}@${node}" \
+      "if [ \"\$(id -u)\" -ne 0 ]; then sudo -n systemctl ${action} ${RUSTFS_SERVICE}; else systemctl ${action} ${RUSTFS_SERVICE}; fi"; then
+    if [ "${action}" = "start" ]; then
+      diagnose_node_start_failure "${node}"
+    fi
+    die "${node}: systemctl ${action} failed"
+  fi
 }
 
 service_action_all() {
@@ -587,6 +632,29 @@ wait_rebalance() {
   die "timed out waiting for rebalance (${REBALANCE_TIMEOUT}s)"
 }
 
+# Start rebalance via the admin API. Nightly builds gate rebalance activation
+# on a live cross-pool fence fleet capability proof that is re-established
+# shortly after a pool joins, so retry a few times before failing.
+start_rebalance_with_retry() {
+  local attempts="${REBALANCE_START_RETRIES}" delay="${REBALANCE_START_RETRY_DELAY}" attempt=1 body code id
+  while :; do
+    body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")"
+    code="$(admin_api_code)"
+    if [ "${code}" = "200" ]; then
+      id="$(printf '%s' "${body}" | jq -r '.id // empty')"
+      log "rebalance started: id=${id}"
+      return 0
+    fi
+    warn "rebalance start attempt ${attempt}/${attempts} failed (HTTP ${code}): ${body}"
+    if [ "${attempt}" -ge "${attempts}" ]; then
+      hint_server_issue "${body}" || true
+      die "rebalance start failed after ${attempts} attempts (see last error above)"
+    fi
+    attempt=$((attempt + 1))
+    sleep "${delay}"
+  done
+}
+
 # Print a detailed decommission failure/progress report for one pool
 print_decommission_detail() {
   local body="$1" pool_id="$2" label="$3"
@@ -817,16 +885,18 @@ step4_write_data() {
       log "DRY-RUN: monitoring storage usage until ${STORAGE_THRESHOLD}%"
       return 0
     fi
-    log "starting warp writes (background)..."
+    local warp_log warp_pid
+    warp_log="${WARP_LOG_FILE:-$(mktemp "${TMPDIR:-/tmp}/rustfs-warp.XXXXXX.log")}"
+    log "starting warp writes (background), log: ${warp_log}"
     warp put --host "${API_ENDPOINT#http://}" \
       --bucket "${WARP_BUCKET}" \
       --access-key "${ACCESS_KEY}" \
       --secret-key "${SECRET_KEY}" \
       --obj.size "${WARP_OBJ_SIZE}" \
       --concurrent "${WARP_CONCURRENT}" \
-      --noprefix --duration "${WARP_DURATION}" --noclear >/tmp/rustfs-warp.log 2>&1 &
-    local warp_pid=$!
-    log "warp PID=${warp_pid}, log /tmp/rustfs-warp.log"
+      --noprefix --duration "${WARP_DURATION}" --noclear >"${warp_log}" 2>&1 &
+    warp_pid=$!
+    log "warp PID=${warp_pid}"
     trap 'kill "${warp_pid:-}" 2>/dev/null || true' EXIT
     monitor_storage "${STORAGE_THRESHOLD}" "${warp_pid}"
     kill "${warp_pid}" 2>/dev/null || true
@@ -866,12 +936,8 @@ step5_expand_pool2() {
 step6_rebalance() {
   log "step 6: start data rebalance (admin API)"
   confirm "About to start rebalance (POST ${API_ENDPOINT}/rustfs/admin/v3/rebalance/start). Continue?"
-  local body id
   if [ "${DRY_RUN}" -eq 0 ]; then
-    body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")"
-    [ "$(admin_api_code)" = "200" ] || die "rebalance start failed (HTTP $(admin_api_code)): ${body}"
-    id="$(printf '%s' "${body}" | jq -r '.id // empty')"
-    log "rebalance started: id=${id}"
+    start_rebalance_with_retry
   fi
   wait_rebalance
 }
@@ -894,12 +960,8 @@ step7_expand_pool3() {
 step8_rebalance() {
   log "step 8: start data rebalance (admin API)"
   confirm "About to start rebalance (POST ${API_ENDPOINT}/rustfs/admin/v3/rebalance/start). Continue?"
-  local body id
   if [ "${DRY_RUN}" -eq 0 ]; then
-    body="$(admin_api POST /rustfs/admin/v3/rebalance/start "")"
-    [ "$(admin_api_code)" = "200" ] || die "rebalance start failed (HTTP $(admin_api_code)): ${body}"
-    id="$(printf '%s' "${body}" | jq -r '.id // empty')"
-    log "rebalance started: id=${id}"
+    start_rebalance_with_retry
   fi
   wait_rebalance
 }
@@ -928,6 +990,7 @@ step9_decommission() {
       break
     fi
     if [ "${attempt}" -ge "${DECOMMISSION_RETRIES}" ]; then
+      warn "if the source bucket has many objects and the tested version is 1.0.0-rc.3, this is the known metacache-listing decommission bug; remove the test bucket (rc rb --force rustfs/${WARP_BUCKET}) or lower --storage-threshold, then re-run step 9"
       die "pool ${DECOMMISSION_POOL_ID} still failed after ${attempt} attempts; investigate manually (POST ${API_ENDPOINT}/rustfs/admin/v3/pools/clear?by-id=true&pool=${DECOMMISSION_POOL_ID} to reset)"
     fi
     warn "attempt ${attempt} failed; clearing metadata and retrying in ${DECOMMISSION_RETRY_DELAY}s"
@@ -1065,6 +1128,12 @@ main() {
   trap 'rm -f "${ADMIN_API_CODE_FILE}"' EXIT
   if [ -n "${LOG_FILE}" ]; then
     mkdir -p "$(dirname "${LOG_FILE}")"
+    if ! touch "${LOG_FILE}" 2>/dev/null; then
+      # A fixed /tmp path may be owned by another user (e.g. a previous root
+      # run); fall back to a unique, always-writable temp file.
+      LOG_FILE="$(mktemp "${TMPDIR:-/tmp}/rustfs-pool-test.XXXXXX.log")"
+      warn "log file not writable; using ${LOG_FILE}"
+    fi
     exec > >(tee -a "${LOG_FILE}") 2>&1
   fi
   if [ "${RESET}" -eq 1 ]; then