test(e2e): register paginator regression in Darwin inventory

fix(storage): publish delete rollback backups atomically
Stage rollback metadata outside the rollback directory and publish it only after the full write succeeds. A short write must not leave a backup that quorum rollback can rename over acknowledged version history. Add an isolated real short-write regression and register the backported ListParts SDK test in the smoke and Linux full inventories.
2026-10-30 15:55:55 +00:00 · 2026-10-03 07:07:41 +08:00 · 2026-10-03 06:43:25 +08:00 · 2026-10-03 06:32:24 +08:00 · 2026-10-03 06:32:24 +08:00 · 2026-10-02 19:59:28 +08:00
31 changed files with 1366 additions and 883 deletions
@@ -1,2 +1,2 @@
-sha256-darwin=14c8cd76b8cbf643329c371daa70a5edd4357d42259a2fd9c653e548675d4a6d
-sha256-linux=80d0fae627007c31f250fb1fb21c8a952861195d53d7f60b6e71636d5ca65b46
+sha256-darwin=6b8e35ef69456bd244d6a6b19179e408763763e1e163aaaf90055490ae12da4b
+sha256-linux=ff4c40f288ab5e91fd28d533c277538c0e3281fa353328ab52f5a8f88e0dc931
@@ -1 +1 @@
-sha256=476cb92aa9e385c9bb509ee5df3e5bd07e8c929730040b60d16791b6c0616d00
+sha256=7e3989625e2e6087c2e7db7ee02d30624b566d5a99832122032fb413ff456662
@@ -551,54 +551,6 @@ jobs:
          if-no-files-found: error
          retention-days: 30

-  # Docker Hub's overview is a separate `full_description` field that
-  # `docker push` never touches, so it drifted from README.md and kept
-  # linking to pages that no longer exist (#7995). Republish it from the
-  # commit the images were built from, only when images were pushed.
-  sync-dockerhub-description:
-    name: Sync Docker Hub Description
-    needs: [ build-check, build-docker ]
-    if: needs.build-check.outputs.should_build == 'true' && needs.build-check.outputs.should_push == 'true'
-    runs-on: sm-standard-2
-    timeout-minutes: 10
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-          ref: ${{ needs.build-check.outputs.source_ref }}
-          sparse-checkout: README.md
-          sparse-checkout-cone-mode: false
-
-      - name: Sync Docker Hub description
-        uses: peter-evans/dockerhub-description@1b9a80c056b620d92cedb9d9b5a223409c68ddfa # v5.0.0
-        with:
-          username: ${{ env.DOCKERHUB_USERNAME }}
-          password: ${{ secrets.DOCKERHUB_TOKEN }}
-          repository: ${{ env.REGISTRY_DOCKERHUB }}
-          readme-filepath: ./README.md
-          # README.md links to docs/ and CONTRIBUTING.md by relative path;
-          # rewrite those to github.com URLs so they resolve on Docker Hub.
-          enable-url-completion: true
-
-      # Docker Hub caps `full_description` at 25,000 bytes and the action
-      # truncates to fit with only a warning, which would publish a README
-      # cut off mid-sentence. README.md is already near the cap, so read the
-      # published overview back and fail if it hit the limit.
-      - name: Check published description was not truncated
-        env:
-          REPOSITORY: ${{ env.REGISTRY_DOCKERHUB }}
-        run: |
-          limit=25000
-          published=$(curl -sSf "https://hub.docker.com/v2/repositories/${REPOSITORY}/" \
-            | jq -j '.full_description' | wc -c | tr -d ' ')
-          echo "Published overview: ${published} bytes (Docker Hub limit: ${limit})"
-          # Truncation lands within one UTF-8 character of the cap.
-          if [ "$published" -ge $((limit - 4)) ]; then
-            echo "::error file=README.md::Docker Hub overview was truncated at ${limit} bytes; trim README.md before the next release."
-            exit 1
-          fi
-
  # Docker build summary
  docker-summary:
    name: Docker Build Summary
@@ -685,9 +685,9 @@ dependencies = [

 [[package]]
 name = "async-compression"
-version = "0.4.48"
+version = "0.4.50"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "fb61aea1a7def73ee7c350a184f0e70b32c182344e2e75bf70c9b621b83417fd"
+checksum = "ee19bd99b43e3691acbad4e840420a4881cea6c0b66a208125a824f8fd53f5a1"
 dependencies = [
 "compression-codecs",
 "compression-core",
@@ -771,17 +771,6 @@ dependencies = [
 "url",
 ]

-[[package]]
-name = "async-recursion"
-version = "1.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.119",
-]
-
 [[package]]
 name = "async-rs"
 version = "0.8.12"
@@ -1012,9 +1001,9 @@ dependencies = [

 [[package]]
 name = "aws-sdk-s3"
-version = "1.150.0"
+version = "1.151.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2fecaa77c862fa1e9d636d54fa06e0049a19a4f071b0d72ea1b9086887d09359"
+checksum = "c82d4b41fe41bfca3c9e1de55645d9a525868ef5ce5e14cee9499ccaf1fc7225"
 dependencies = [
 "arc-swap",
 "aws-credential-types",
@@ -1215,9 +1204,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-http-client"
-version = "1.4.2"
+version = "1.5.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7bd25384a4e437aa8d8f339afad4b69e786b936a7cb10db668a7aaf66717b1a8"
+checksum = "51c89cc3f1f281d659a67a519a1b5c6d445b5ce09fa7e5aee40c2c2707e9509d"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-protocol-test",
@@ -1297,9 +1286,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-runtime"
-version = "1.15.0"
+version = "1.16.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "591e0024cdf2a8de8711860f7e1d5cfbca26b800ce13793a9aa2f4f5918e6d95"
+checksum = "d6e302ac1d88b99652489df31abdec6ac42a2ab2ac3982ad0ac49f64dfaf28ba"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-http",
@@ -1321,9 +1310,9 @@ dependencies = [

 [[package]]
 name = "aws-smithy-runtime-api"
-version = "1.18.0"
+version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9bb0deb40a69de59f7809251cfe1ab16de135f42c49d222d1202b4d0ce19720f"
+checksum = "c0730c16f91124c6a2abb4932c77e299288b3dd9f967ea2e9ec48cc6731e87a4"
 dependencies = [
 "aws-smithy-async",
 "aws-smithy-runtime-api-macros",
@@ -1544,9 +1533,9 @@ dependencies = [

 [[package]]
 name = "bigdecimal"
-version = "0.4.10"
+version = "0.4.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d6867f1565b3aad85681f1015055b087fcfd840d6aeee6eee7f2da317603695"
+checksum = "8fa3f3d8cbf4dffcfe4991de61d012bef509a409ecbe9dd41049bfe32b4d4653"
 dependencies = [
 "autocfg",
 "libm",
@@ -2209,11 +2198,11 @@ dependencies = [

 [[package]]
 name = "compression-codecs"
-version = "0.4.43"
+version = "0.4.45"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bef16c47ba2797aa6a909cc37d39911f3a6743811fe7408ac0b0cc0276b656e9"
+checksum = "98fc98460ba0ad5317075d3632b8dfc45d0be8c4a49347c2a38272019717614a"
 dependencies = [
- "brotli 8.0.4",
+ "brotli 9.0.0",
 "bzip2",
 "compression-core",
 "flate2",
@@ -4804,9 +4793,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-gax-internal"
-version = "0.7.19"
+version = "0.7.20"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c47c73e95dc5e1de74bebe21b406394b53ff5ccfdfe0e81a0443cdef468e6744"
+checksum = "e873f7e6e39030c07b6d9697bf3b999a269175c8b433e058f56ae10d129f12f7"
 dependencies = [
 "bytes",
 "futures",
@@ -4820,9 +4809,9 @@ dependencies = [
 "http-body-util",
 "hyper",
 "lazy_static",
- "opentelemetry 0.32.0",
- "opentelemetry-semantic-conventions 0.32.1",
- "opentelemetry_sdk 0.32.1",
+ "opentelemetry",
+ "opentelemetry-semantic-conventions",
+ "opentelemetry_sdk",
 "percent-encoding",
 "pin-project",
 "prost 0.14.4",
@@ -4838,7 +4827,7 @@ dependencies = [
 "tonic-prost",
 "tower",
 "tracing",
- "tracing-opentelemetry 0.33.0",
+ "tracing-opentelemetry",
 ]

 [[package]]
@@ -4908,9 +4897,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-storage"
-version = "1.19.0"
+version = "1.20.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5f7fa1e66d6b47220dbfc5f243b634136e5b8d14d64cb13419bd59f976681da3"
+checksum = "570eb8d0784118a5f4ae1219fd76ff3b14e90b7b0f877478c02b1374e9bbcb8c"
 dependencies = [
 "async-trait",
 "base64 0.23.1",
@@ -4961,9 +4950,9 @@ dependencies = [

 [[package]]
 name = "google-cloud-wkt"
-version = "1.8.0"
+version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ad4b36fefda4485f0922de7a58a3e562e984d513442a1aae2281dce25124be2e"
+checksum = "7f3037907e542cf8b8580aa4cf8ad68568d7ac8e7274519a7d0a121867d434da"
 dependencies = [
 "base64 0.23.1",
 "bytes",
@@ -5282,9 +5271,9 @@ dependencies = [

 [[package]]
 name = "hotpath"
-version = "0.26.1"
+version = "0.28.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "036ff36fa1c4ae8dfb686606537eaa15b5d5da0c51bf94abeaf9dfc6ee54cac7"
+checksum = "d8345e171844f80e5ae4cbed8bc8276e094f6ff2293a6f8e547c200183976674"
 dependencies = [
 "arc-swap",
 "async-channel",
@@ -5310,6 +5299,7 @@ dependencies = [
 "rustc-demangle",
 "serde",
 "serde_json",
+ "time",
 "tiny_http",
 "tokio",
 ]
@@ -5322,9 +5312,9 @@ checksum = "e149b660b611650bdf3307941ef3e9b81be3c30ed1dfdbe7f670319674f023a6"

 [[package]]
 name = "hotpath-macros"
-version = "0.26.1"
+version = "0.28.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1ff68b15fdbcddfbad685f445597580d6fcbf28b49f01ff975e934351079e13b"
+checksum = "bf73e28fe4a7595c322e4265c1608cb10c215cbbce11e49cfcc258766ba686d6"
 dependencies = [
 "proc-macro2",
 "quote",
@@ -6044,9 +6034,9 @@ dependencies = [

 [[package]]
 name = "lazy_static"
-version = "1.5.0"
+version = "1.5.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "bbd2bcb4c963f2ddae06a2efc7e9f3591312473c50c6685e1f298068316e66fe"
+checksum = "20870f649af7073d53e38067b2a84312175d56ea15217e1b15bc83506ec50afb"
 dependencies = [
 "spin 0.9.9",
 ]
@@ -7318,20 +7308,6 @@ version = "0.2.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"

-[[package]]
-name = "opentelemetry"
-version = "0.32.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0142c63252a9e054e68a4c61a5778f7b14f576274d593f8ce883d191a099682"
-dependencies = [
- "futures-core",
- "futures-sink",
- "js-sys",
- "pin-project-lite",
- "thiserror 2.0.21",
- "tracing",
-]
-
 [[package]]
 name = "opentelemetry"
 version = "0.33.0"
@@ -7352,7 +7328,7 @@ version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e2b304e8d8f11326ba366a99ca5c7aaf7cd397b67268a332437175aeb42a40db"
 dependencies = [
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "tracing",
 "tracing-core",
 "tracing-log",
@@ -7368,7 +7344,7 @@ dependencies = [
 "async-trait",
 "bytes",
 "http 1.5.0",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "reqwest",
 ]

@@ -7381,10 +7357,10 @@ dependencies = [
 "flate2",
 "http 1.5.0",
 "httpdate",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "opentelemetry-http",
 "opentelemetry-proto",
- "opentelemetry_sdk 0.33.0",
+ "opentelemetry_sdk",
 "prost 0.14.4",
 "reqwest",
 "thiserror 2.0.21",
@@ -7396,17 +7372,11 @@ version = "0.33.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "25da1ac11a0aeccf38d7f77ee0348715adaf8340f65ad46c94a02c6b20e2f65d"
 dependencies = [
- "opentelemetry 0.33.0",
- "opentelemetry_sdk 0.33.0",
+ "opentelemetry",
+ "opentelemetry_sdk",
 "prost 0.14.4",
 ]

-[[package]]
-name = "opentelemetry-semantic-conventions"
-version = "0.32.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c913ac17a6c451661ee255f4625d143e51647ae78ebd969b75e41c4442f4fe47"
-
 [[package]]
 name = "opentelemetry-semantic-conventions"
 version = "0.33.0"
@@ -7420,24 +7390,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f71803f481d3f5ccbe477edb42a78f473613712b2327835911a9ea0ab069824"
 dependencies = [
 "chrono",
- "opentelemetry 0.33.0",
- "opentelemetry_sdk 0.33.0",
-]
-
-[[package]]
-name = "opentelemetry_sdk"
-version = "0.32.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9b59f80e1ac4d5ff7a2db8fb6c80badb7f0f3f858211fba08dd9aaec750894f9"
-dependencies = [
- "futures-channel",
- "futures-executor",
- "futures-util",
- "opentelemetry 0.32.0",
- "percent-encoding",
- "portable-atomic",
- "rand 0.9.5",
- "thiserror 2.0.21",
+ "opentelemetry",
+ "opentelemetry_sdk",
 ]

 [[package]]
@@ -7449,7 +7403,7 @@ dependencies = [
 "futures-channel",
 "futures-executor",
 "futures-util",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "percent-encoding",
 "portable-atomic",
 "rand 0.9.5",
@@ -7939,9 +7893,9 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"

 [[package]]
 name = "pin-utils"
-version = "0.1.0"
+version = "0.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "8b870d8c151b6f2fb93e84a13146138f05d02ed11c7e7c54f8826aaaf7c9f184"
+checksum = "13bee6c73da26345c729282832b60b0363cf3dd9f4bfd81d8551b7a1c889a113"

 [[package]]
 name = "piper"
@@ -8473,9 +8427,9 @@ dependencies = [

 [[package]]
 name = "pulldown-cmark-to-cmark"
-version = "22.0.1"
+version = "22.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "ab1ad36992cead65f02aa399a373a42730922f1525d988172634fdefdecb8a60"
+checksum = "84bbb29c624230c4bd1047bbdb2aa47e41c860e9665ce62ba9504eebe91bf867"
 dependencies = [
 "pulldown-cmark",
 ]
@@ -8710,9 +8664,9 @@ dependencies = [

 [[package]]
 name = "quinn-proto"
-version = "0.11.18"
+version = "0.11.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a9746dbde176634f4f2f1faf2404e30a31b2bc1e9cafb5329c95d8177a18c9fc"
+checksum = "0e750cca55fe4f0439a15d0bb529da9651e79993e8e72c61a899a36d462befbe"
 dependencies = [
 "aws-lc-rs",
 "bytes",
@@ -8733,9 +8687,9 @@ dependencies = [

 [[package]]
 name = "quinn-udp"
-version = "0.5.15"
+version = "0.5.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "35a133f956daabe89a61a685c2649f13d82d5aa4bd5d12d1277e1072a21c0694"
+checksum = "af66907df18639dcf4db56ca65490cabc4b27a97dbadd96f2926cca73298f016"
 dependencies = [
 "cfg_aliases",
 "libc",
@@ -9416,9 +9370,9 @@ dependencies = [

 [[package]]
 name = "russh-sftp"
-version = "3.0.0"
+version = "3.0.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "093197e526668d92bba562e2bbbe98d1af9831bf080b619c736316ca1fa35101"
+checksum = "9bb17f72aba4314656b359ffba72c68dcbd70ed9e97497898b75498fe7198042"
 dependencies = [
 "bitflags 2.13.2",
 "bytes",
@@ -9554,9 +9508,9 @@ dependencies = [
 "mime_guess",
 "moka",
 "object 0.40.0",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "opentelemetry-proto",
- "opentelemetry_sdk 0.33.0",
+ "opentelemetry_sdk",
 "p256 0.14.0",
 "parking_lot",
 "percent-encoding",
@@ -9647,7 +9601,7 @@ dependencies = [
 "tower",
 "tower-http 0.7.1",
 "tracing",
- "tracing-opentelemetry 0.34.0",
+ "tracing-opentelemetry",
 "tracing-subscriber",
 "url",
 "urlencoding",
@@ -9797,7 +9751,6 @@ version = "1.0.1"
 dependencies = [
 "arc-swap",
 "async-channel",
- "async-recursion",
 "async-trait",
 "aws-credential-types",
 "aws-sdk-s3",
@@ -9837,8 +9790,8 @@ dependencies = [
 "metrics-util",
 "moka",
 "num_cpus",
- "opentelemetry 0.33.0",
- "opentelemetry_sdk 0.33.0",
+ "opentelemetry",
+ "opentelemetry_sdk",
 "parking_lot",
 "path-absolutize",
 "pin-project-lite",
@@ -9899,7 +9852,7 @@ dependencies = [
 "tower",
 "tracing",
 "tracing-core",
- "tracing-opentelemetry 0.34.0",
+ "tracing-opentelemetry",
 "tracing-subscriber",
 "url",
 "urlencoding",
@@ -10416,12 +10369,12 @@ dependencies = [
 "metrics-util",
 "num_cpus",
 "nvml-wrapper",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "opentelemetry-appender-tracing",
 "opentelemetry-otlp",
- "opentelemetry-semantic-conventions 0.33.0",
+ "opentelemetry-semantic-conventions",
 "opentelemetry-stdout",
- "opentelemetry_sdk 0.33.0",
+ "opentelemetry_sdk",
 "percent-encoding",
 "pyroscope",
 "rcgen",
@@ -10453,7 +10406,7 @@ dependencies = [
 "tracing",
 "tracing-appender",
 "tracing-error",
- "tracing-opentelemetry 0.34.0",
+ "tracing-opentelemetry",
 "tracing-subscriber",
 "url",
 "zstd 0.14.0",
@@ -10762,7 +10715,6 @@ dependencies = [
 name = "rustfs-s3select-query"
 version = "1.0.1"
 dependencies = [
- "async-recursion",
 "async-trait",
 "bytes",
 "datafusion",
@@ -11170,9 +11122,9 @@ dependencies = [

 [[package]]
 name = "rustls-connector"
-version = "0.23.8"
+version = "0.23.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1babecfcc65b139b812e74bcc7f9ec7b4e00db659fd42d99567b7e77f0c714c6"
+checksum = "524994fd6957827c775abc941f41a146ce351bbd0a9b8cf632ee37f43a13c205"
 dependencies = [
 "futures-io",
 "futures-rustls",
@@ -12283,9 +12235,9 @@ checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"

 [[package]]
 name = "suppaftp"
-version = "12.1.0"
+version = "12.1.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "d79f066c76927a775d61b5642e0b712a17582ebe7f62b7e5abba246e619878f7"
+checksum = "b161d71217da0cc91bd09c8f98ac5ad2a5e5b2735c6dfe1c6cb08b4471612f35"
 dependencies = [
 "async-trait",
 "chrono",
@@ -12776,9 +12728,9 @@ dependencies = [

 [[package]]
 name = "tokio-rustls"
-version = "0.26.5"
+version = "0.26.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b0c85f2c3ef0b1cd58b36682f4b17aaa995f0e5db534d85692b4903abce21f67"
+checksum = "c9cc2678c2cdd569ef8215e2afd7954ada2ae20b4fdd2c5fe6139a3b02d105db"
 dependencies = [
 "rustls",
 "tokio",
@@ -13086,20 +13038,6 @@ dependencies = [
 "tracing-core",
 ]

-[[package]]
-name = "tracing-opentelemetry"
-version = "0.33.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "adbc64cba7137545b8044cb1fe9814f7aacf3c6b5f9b45be8bb5db538befdb26"
-dependencies = [
- "js-sys",
- "opentelemetry 0.32.0",
- "tracing",
- "tracing-core",
- "tracing-subscriber",
- "web-time",
-]
-
 [[package]]
 name = "tracing-opentelemetry"
 version = "0.34.0"
@@ -13107,7 +13045,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0a904802a1b902f43638b677ff2a650847e3b4404101b6c586d648e8c1e3e8fe"
 dependencies = [
 "js-sys",
- "opentelemetry 0.33.0",
+ "opentelemetry",
 "smallvec",
 "tracing",
 "tracing-core",
@@ -14042,9 +13980,9 @@ dependencies = [

 [[package]]
 name = "xxhash-rust"
-version = "0.8.18"
+version = "0.8.19"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "aee1b19627c7c60102ab80d3a9cbe18de90bfe03bfa6c3715447681f0e8c8af6"
+checksum = "550a2b930b62486a393c52d5c3b84bff264b28aa437ed64694d31e93b1757af7"

 [[package]]
 name = "yansi"
@@ -146,8 +146,7 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.1" }
 async-channel = "2.5.0"
 async_zip = { default-features = false, version = "0.0.19" }
 mysql_async = { default-features = false, version = "0.37.1" }
-async-compression = { version = "0.4.48" }
-async-recursion = "1.1.1"
+async-compression = { version = "0.4.50" }
 async-trait = "0.1.92"
 async-nats = { version = "0.50.0", default-features = false }
 axum = "0.8.9"
@@ -170,7 +169,7 @@ reqwest = "0.13.5"
 rustfs-kafka-async = { version = "1.3.1" }
 socket2 = { version = "0.6.5" }
 tokio = { version = "1.53.1" }
-tokio-rustls = { default-features = false, version = "0.26.5" }
+tokio-rustls = { default-features = false, version = "0.26.6" }
 tokio-stream = { version = "0.1.19" }
 tokio-test = "0.4.6"
 tokio-util = { version = "0.7.19" }
@@ -246,11 +245,11 @@ atomic_enum = "0.3.0"
 aws-config = { version = "1.12.0" }
 aws-credential-types = { version = "1.3.0" }
 aws-sdk-kms = { default-features = false, version = "1.122.0" }
-aws-sdk-s3 = { default-features = false, version = "1.150.0" }
+aws-sdk-s3 = { default-features = false, version = "1.151.0" }
 aws-sdk-sts = { default-features = false, version = "1.118.0" }
 aws-smithy-async = { version = "1.3.0" }
-aws-smithy-http-client = { default-features = false, version = "1.4.2" }
-aws-smithy-runtime-api = { version = "1.18.0" }
+aws-smithy-http-client = { default-features = false, version = "1.5.0" }
+aws-smithy-runtime-api = { version = "1.19.0" }
 aws-smithy-types = { version = "1.8.1" }
 base64-simd = "0.8.0"
 brotli = "9.0.0"
@@ -269,7 +268,7 @@ enumset = "1.1.14"
 faster-hex = "1.0.0"
 flate2 = "1.1.10"
 glob = "0.3.4"
-google-cloud-storage = "1.19.0"
+google-cloud-storage = "1.20.0"
 google-cloud-auth = "1.17.0"
 hashbrown = { version = "0.17.1" }
 # Base32 for RFC 6238 TOTP shared secrets (RFC 4648 unpadded, the alphabet
@@ -279,7 +278,7 @@ hex-simd = "0.8.0"
 highway = { version = "1.3.0" }
 hostname = "0.4.2"
 ipnetwork = { version = "0.21.1" }
-lazy_static = "1.5.0"
+lazy_static = "1.5.1"
 libc = "0.2.189"
 libsystemd = "0.7.2"
 local-ip-address = "0.6.13"
@@ -346,7 +345,7 @@ walkdir = "2.5.0"
 winapi-util = "0.1.11"
 windows = { version = "0.62.2" }
 windows-sys = "0.61.2"
-xxhash-rust = { version = "0.8.18" }
+xxhash-rust = { version = "0.8.19" }
 zip = "8.6.0"
 zstd = "0.14.0"

@@ -366,10 +365,10 @@ pyroscope = { version = "2.1.1" }
 # FTP and SFTP
 libunftp = { version = "0.23.1" }
 unftp-core = "0.1.0"
-suppaftp = { version = "12.1.0" }
+suppaftp = { version = "12.1.1" }
 rcgen = { version = "0.14.10", default-features = false, features = ["aws_lc_rs", "crypto", "pem"] }
 russh = { version = "0.63.3" }
-russh-sftp = "3.0.0"
+russh-sftp = "3.0.1"

 # WebDAV
 dav-server = "0.11.0"
@@ -377,7 +376,7 @@ dav-server = "0.11.0"
 # Performance Analysis and Memory Profiling
 rustfs-mimalloc = { version = "0.5.6" }
 # Preserve Unicode focus filters until rustfs/backlog#2302 is resolved.
-hotpath = { version = "0.26.1", default-features = false }
+hotpath = { version = "0.28.0", default-features = false }
 # Snapshot testing for output format regression detection
 insta = { version = "1.48" }

@@ -65,6 +65,130 @@ mod tests {
        Ok(())
    }

+    #[tokio::test]
+    async fn list_parts_sdk_paginator_terminates_and_preserves_completion() -> Result<(), Box<dyn std::error::Error + Send + Sync>>
+    {
+        init_logging();
+        let mut env = RustFSTestEnvironment::new().await?;
+        env.start_rustfs_server(Vec::new()).await?;
+        let client = env.create_s3_client();
+        let bucket = "multipart-pagination";
+        let key = "sparse-parts.bin";
+        env.create_test_bucket(bucket).await?;
+        let upload = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
+        let upload_id = upload.upload_id().ok_or("CreateMultipartUpload returned no upload ID")?;
+
+        let mut empty_pages = client
+            .list_parts()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .into_paginator()
+            .stop_on_duplicate_token(false)
+            .send();
+        let empty = tokio::time::timeout(std::time::Duration::from_secs(30), empty_pages.next())
+            .await?
+            .ok_or("empty upload must return one page")??;
+        assert!(empty.parts().is_empty());
+        assert_eq!(empty.is_truncated(), Some(false));
+        assert_eq!(empty.next_part_number_marker(), None);
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_secs(30), empty_pages.next())
+                .await?
+                .is_none()
+        );
+
+        let mut expected_body = Vec::new();
+        let mut completed_parts = Vec::new();
+        for (part_number, body) in [(1, vec![b'a'; PART_SIZE]), (3, vec![b'b'; PART_SIZE]), (10, b"tail".to_vec())] {
+            expected_body.extend_from_slice(&body);
+            let uploaded = client
+                .upload_part()
+                .bucket(bucket)
+                .key(key)
+                .upload_id(upload_id)
+                .part_number(part_number)
+                .body(ByteStream::from(body))
+                .send()
+                .await?;
+            completed_parts.push(
+                CompletedPart::builder()
+                    .part_number(part_number)
+                    .set_e_tag(uploaded.e_tag().map(str::to_owned))
+                    .build(),
+            );
+        }
+
+        for (page_size, expected_pages) in [(1, 3), (2, 2), (3, 1), (1000, 1)] {
+            let mut pages = client
+                .list_parts()
+                .bucket(bucket)
+                .key(key)
+                .upload_id(upload_id)
+                .into_paginator()
+                .page_size(page_size)
+                // The server must terminate pagination even without the SDK's duplicate-token safeguard.
+                .stop_on_duplicate_token(false)
+                .send();
+            let mut page_count = 0;
+            let mut listed_parts = Vec::new();
+            while let Some(page) = tokio::time::timeout(std::time::Duration::from_secs(30), pages.next()).await? {
+                assert!(page_count < expected_pages, "paginator must terminate after the final page");
+                let page = page?;
+                page_count += 1;
+                let truncated = page_count < expected_pages;
+                assert_eq!(page.is_truncated(), Some(truncated));
+                let last_part = page.parts().last().and_then(|part| part.part_number());
+                let expected_marker = truncated.then(|| last_part.expect("truncated page must contain parts").to_string());
+                assert_eq!(page.next_part_number_marker(), expected_marker.as_deref());
+                listed_parts.extend(
+                    page.parts()
+                        .iter()
+                        .map(|part| part.part_number().expect("part number must be present")),
+                );
+            }
+            assert_eq!(page_count, expected_pages);
+            assert_eq!(listed_parts, [1, 3, 10]);
+        }
+
+        let mut resumed_pages = client
+            .list_parts()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .part_number_marker("2")
+            .into_paginator()
+            .page_size(1)
+            .stop_on_duplicate_token(false)
+            .send();
+        for (part_number, next_marker) in [(3, Some("3")), (10, None)] {
+            let page = tokio::time::timeout(std::time::Duration::from_secs(30), resumed_pages.next())
+                .await?
+                .ok_or("parts above a missing marker must still be listed")??;
+            assert_eq!(page.parts().len(), 1);
+            assert_eq!(page.parts()[0].part_number(), Some(part_number));
+            assert_eq!(page.is_truncated(), Some(next_marker.is_some()));
+            assert_eq!(page.next_part_number_marker(), next_marker);
+        }
+        assert!(
+            tokio::time::timeout(std::time::Duration::from_secs(30), resumed_pages.next())
+                .await?
+                .is_none()
+        );
+
+        client
+            .complete_multipart_upload()
+            .bucket(bucket)
+            .key(key)
+            .upload_id(upload_id)
+            .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed_parts)).build())
+            .send()
+            .await?;
+        assert_completed_object(&client, bucket, key, "STANDARD", &expected_body).await?;
+        env.stop_server();
+        Ok(())
+    }
+
    #[tokio::test]
    async fn multipart_upload_preserves_standard_and_rrs_across_retry_and_resume()
    -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
@@ -208,7 +208,6 @@ aws-sdk-s3 = { workspace = true, default-features = false, features = ["sigv4a",
 urlencoding = { workspace = true }
 smallvec = { workspace = true, features = ["serde"] }
 shadow-rs = { workspace = true, default-features = false }
-async-recursion.workspace = true
 aws-credential-types = { workspace = true }
 aws-smithy-types = { workspace = true }
 aws-smithy-runtime-api = { workspace = true, features = ["http-1x"] }
@@ -56,7 +56,7 @@ impl DiagnosticPacing {
        // Reservation is never refunded: a failed or dropped RPC may have sent some payload.
        let previous = self
            .charged_bytes
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |charged| {
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |charged| {
                charged.checked_add(bytes).filter(|total| *total <= MAX_NETWORK_PROBE_BYTES)
            })
            .map_err(|_| NetworkPeerProbeError::LimitExceeded)?;
@@ -12976,7 +12976,7 @@ impl ECStore {
                .targets
                .iter()
                .find(|target| target.pool_index == permitted_target_pool_index)
-                .and_then(&candidate)
+                .and_then(candidate)
            {
                return Ok(pool_index);
            }
@@ -306,7 +306,7 @@ pub mod test_util {
        let Some(fault) = fault else { return Ok(None) };
        if fault
            .remaining
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |remaining| remaining.checked_sub(1))
            != Ok(1)
        {
            return Ok(None);
@@ -73,7 +73,7 @@ impl ProcessResultBudget {
        }
        inner
            .available
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |available| available.checked_sub(bytes))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |available| available.checked_sub(bytes))
            .map(|_| {
                Some(ResultBudgetReservation {
                    inner: Arc::clone(inner),
@@ -908,7 +908,7 @@ fn explicit_url_port(raw: &str) -> Result<Option<u16>> {
        return Err(invalid());
    }

-    let (scheme, remainder) = raw.split_once("://").ok_or_else(&invalid)?;
+    let (scheme, remainder) = raw.split_once("://").ok_or_else(invalid)?;
    if !scheme.eq_ignore_ascii_case("http") && !scheme.eq_ignore_ascii_case("https") {
        return Err(invalid());
    }
@@ -916,17 +916,17 @@ fn explicit_url_port(raw: &str) -> Result<Option<u16>> {
        .split('/')
        .next()
        .filter(|authority| !authority.is_empty())
-        .ok_or_else(&invalid)?;
+        .ok_or_else(invalid)?;
    if authority.contains('@') {
        return Err(invalid());
    }

    let port = if let Some(bracketed) = authority.strip_prefix('[') {
-        let (_, suffix) = bracketed.split_once(']').ok_or_else(&invalid)?;
+        let (_, suffix) = bracketed.split_once(']').ok_or_else(invalid)?;
        if suffix.is_empty() {
            return Ok(None);
        }
-        suffix.strip_prefix(':').ok_or_else(&invalid)?
+        suffix.strip_prefix(':').ok_or_else(invalid)?
    } else if let Some((_, port)) = authority.rsplit_once(':') {
        port
    } else {
@@ -1001,11 +1001,7 @@ fn parse_explicit_local_endpoint_host(raw: &str) -> Result<Host<String>> {

    let host = Host::parse(raw).map_err(|_| invalid())?;
    let host = match host {
-        Host::Domain(domain) => Host::Domain(
-            domain_without_optional_trailing_dot(&domain)
-                .ok_or_else(&invalid)?
-                .to_string(),
-        ),
+        Host::Domain(domain) => Host::Domain(domain_without_optional_trailing_dot(&domain).ok_or_else(invalid)?.to_string()),
        host => host,
    };
    if matches!(&host, Host::Domain(domain) if domain.contains('*'))
@@ -396,7 +396,7 @@ impl InstanceContext {
    pub(crate) fn begin_namespace_commit(self: &Arc<Self>) -> Arc<NamespaceCommitGuard> {
        let counted = self
            .namespace_commits
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |count| count.checked_add(1))
            .is_ok();
        if counted {
            self.advance_namespace_commit_generation();
@@ -412,7 +412,7 @@ impl InstanceContext {
    fn advance_namespace_commit_generation(&self) {
        let _ = self
            .namespace_commit_generation
-            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |generation| Some(generation.saturating_add(1)));
+            .try_update(Ordering::AcqRel, Ordering::Acquire, |generation| Some(generation.saturating_add(1)));
    }

    pub(crate) fn namespace_commit_generation(&self) -> u64 {
@@ -7182,10 +7182,10 @@ fn parts_after_marker(part_numbers: &[usize], part_number_marker: usize) -> Opti
        return Some(part_numbers);
    }

-    part_numbers
-        .iter()
-        .position(|&part_number| part_number != 0 && part_number == part_number_marker)
-        .map(|index| &part_numbers[index + 1..])
+    // reduce_quorum_part_numbers returns sorted numbers; the marker need not exist.
+    part_numbers.last().filter(|&&last| part_number_marker <= last)?;
+    let index = part_numbers.partition_point(|&part_number| part_number <= part_number_marker);
+    Some(&part_numbers[index..])
 }

 pub fn canonicalize_etag(etag: &str) -> String {
@@ -13983,6 +13983,24 @@ mod tests {
        assert!(parts_after_marker(&part_numbers, 4).is_none());
    }

+    #[test]
+    fn parts_after_marker_uses_exclusive_numeric_boundary_for_sparse_parts() {
+        let part_numbers = [1, 3, 10];
+        for (marker, expected) in [
+            (0, Some(&part_numbers[..])),
+            (1, Some(&part_numbers[1..])),
+            (2, Some(&part_numbers[1..])),
+            (3, Some(&part_numbers[2..])),
+            (9, Some(&part_numbers[2..])),
+            (10, Some(&part_numbers[3..])),
+            (11, None),
+            (usize::MAX, None),
+        ] {
+            assert_eq!(parts_after_marker(&part_numbers, marker), expected, "marker {marker}");
+        }
+        assert_eq!(parts_after_marker(&[], 1), None);
+    }
+
    #[test]
    fn delete_file_info_version_id_maps_explicit_null_version_to_stored_null() {
        assert_eq!(delete_file_info_version_id(Some(Uuid::nil())), None);
@@ -2125,7 +2125,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {

        if object_parts.len() > ret.parts.len() {
            ret.is_truncated = true;
-            ret.next_part_number_marker = ret.parts.last().map(|v| v.part_num).unwrap_or_default();
+            ret.next_part_number_marker = ret.parts.last().map(|v| v.part_num);
        }

        ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
@@ -4505,6 +4505,54 @@ mod tests {
        }
    }

+    #[tokio::test]
+    async fn list_object_parts_pagination_preserves_sparse_markers_and_terminates() {
+        let (_temp_dirs, disks, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-pagination";
+        let object = "object";
+        make_bucket_on_all(&disks, bucket).await;
+        let opts = ObjectOptions::default();
+        let upload = set_disks
+            .new_multipart_upload(bucket, object, &opts)
+            .await
+            .expect("multipart upload should be created");
+
+        let empty = set_disks
+            .list_object_parts(bucket, object, &upload.upload_id, None, 2, &opts)
+            .await
+            .expect("empty upload should be listable");
+        assert!(empty.parts.is_empty());
+        assert!(!empty.is_truncated);
+        assert_eq!(empty.next_part_number_marker, None);
+
+        for part_number in [10, 1, 3] {
+            put_test_part(&set_disks, bucket, object, &upload.upload_id, part_number, b"part", 4).await;
+        }
+
+        for (marker, max_parts, expected_parts, next_marker) in [
+            (None, 0, vec![], None),
+            (None, 1, vec![1], Some(1)),
+            (None, 2, vec![1, 3], Some(3)),
+            (None, 3, vec![1, 3, 10], None),
+            (None, MAX_PARTS_COUNT + 1, vec![1, 3, 10], None),
+            (Some(1), 2, vec![3, 10], None),
+            (Some(2), 1, vec![3], Some(3)),
+            (Some(3), 2, vec![10], None),
+            (Some(10), 2, vec![], None),
+            (Some(11), 2, vec![], None),
+        ] {
+            let page = set_disks
+                .list_object_parts(bucket, object, &upload.upload_id, marker, max_parts, &opts)
+                .await
+                .expect("uploaded parts should be listable");
+            assert_eq!(page.part_number_marker, marker.unwrap_or_default());
+            assert_eq!(page.max_parts, max_parts.min(MAX_PARTS_COUNT));
+            assert_eq!(page.parts.iter().map(|part| part.part_num).collect::<Vec<_>>(), expected_parts);
+            assert_eq!(page.is_truncated, next_marker.is_some());
+            assert_eq!(page.next_part_number_marker, next_marker);
+        }
+    }
+
    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);
@@ -4362,7 +4362,7 @@ mod tests {
            // successful attempts. Only injected faults spend this global budget.
            // A real failure may consume an attempt, so preserve the final chance.
            faults
-                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
+                .try_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
                    (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1))
                        .then_some(faults.saturating_add(1))
                })
@@ -918,10 +918,19 @@ async fn partial_write_sigkill_replay_scenario(protected: bool) {
            "replayed responsibility must be released only after verified repair"
        );
    } else {
+        // A repair can finish before the next legacy check parks its responsibility.
        assert!(
            wait_until(|| async {
                let snapshot = manager.operations_snapshot().await;
-                snapshot.queue_length == 0 && snapshot.active_tasks == 0
+                snapshot.queue_length == 0
+                    && snapshot.active_tasks == 0
+                    && mrf_queue::list_legacy_responsibilities(None, 32).await.is_ok_and(|state| {
+                        state.responsibilities.iter().any(|entry| {
+                            entry.bucket == "partial-crash"
+                                && entry.object == "crash.bin"
+                                && entry.status == "held_unverified_legacy"
+                        })
+                    })
            })
            .await,
            "legacy replay attempts must finish"
@@ -43,7 +43,6 @@ hotpath-cpu = ["hotpath", "hotpath/hotpath-cpu", "rustfs-s3select-api/hotpath-cp
 [dependencies]
 hotpath.workspace = true
 rustfs-s3select-api = { workspace = true }
-async-recursion = { workspace = true }
 async-trait.workspace = true
 bytes.workspace = true
 datafusion = { workspace = true, default-features = false, features = ["parquet", "recursive_protection", "sql"] }
@@ -14,7 +14,6 @@

 use std::{convert::Infallible, ops::ControlFlow};

-use async_recursion::async_recursion;
 use async_trait::async_trait;
 use datafusion::sql::{
    planner::{IdentNormalizer, SqlToRel},
@@ -58,7 +57,6 @@ impl<'a, S: ContextProviderExtension + Send + Sync + 'a> SqlPlanner<'a, S> {
    }

    /// Generate a logical plan from an  Extent SQL statement
-    #[async_recursion]
    pub(crate) async fn statement_to_plan(&self, statement: ExtStatement, session: &SessionCtx) -> QueryResult<Plan> {
        match statement {
            ExtStatement::SqlStatement(stmt) => self.df_sql_to_plan(*stmt, session).await,
@@ -87,13 +87,14 @@ pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_st
 pub use rustfs_scanner_metrics::last_minute;
 pub use scanner::{
    SCANNER_RECOVERY_INTENT_ACTION_USAGE_FULL_REBUILD, ScannerCycleRecoveryMarker, ScannerCycleRecoveryStatus,
-    ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogReplicaStatus,
-    ScannerPauseBacklogReplicaStatusState, ScannerPauseBacklogStatus, ScannerPauseBacklogThresholds,
-    ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord, ScannerRecoveryIntentRequest,
-    ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent, get_scanner_usage_recovery_intent, init_data_scanner,
-    init_scanner_with_recovery, register_scanner_pause_backlog_retirement, reset_scanner_cycle_recovery,
-    reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent, scanner_cycle_recovery_status,
-    scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256, scanner_topology_digest,
+    ScannerCycleScheduleStatus, ScannerPauseBacklogAlertReason, ScannerPauseBacklogPhase, ScannerPauseBacklogReplicaId,
+    ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState, ScannerPauseBacklogStatus,
+    ScannerPauseBacklogThresholds, ScannerRecoveryIntentAcceptResult, ScannerRecoveryIntentConflict, ScannerRecoveryIntentRecord,
+    ScannerRecoveryIntentRequest, ScannerUsageStateResetResult, accept_scanner_usage_recovery_intent,
+    get_scanner_usage_recovery_intent, init_data_scanner, init_scanner_with_recovery, register_scanner_pause_backlog_retirement,
+    reset_scanner_cycle_recovery, reset_scanner_usage_state_for_full_rebuild, run_scanner_usage_recovery_intent,
+    scanner_cycle_recovery_status, scanner_cycle_schedule_status, scanner_pause_backlog_status, scanner_recovery_actor_sha256,
+    scanner_topology_digest,
 };
 pub use scanner_io::{
    ScannerDirtyUsageAckError, ScannerDirtyUsageBucket, ScannerDirtyUsageClearObserver, ScannerDirtyUsageMutationObserver,
@@ -3800,7 +3800,7 @@ mod usage_store;

 use activity::*;
 use backlog::*;
-pub use backlog::{ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState};
+pub use backlog::{ScannerPauseBacklogReplicaId, ScannerPauseBacklogReplicaStatus, ScannerPauseBacklogReplicaStatusState};
 use cycle_state::*;
 use leadership::*;
 pub(crate) use usage_store::RootPublicationProof;
@@ -173,6 +173,7 @@ pub struct ScannerPauseBacklogReplicaStatus {
    pub committed_generation: Option<u64>,
    pub committed_writer_epoch: Option<u64>,
    pub committed_replica_count: Option<usize>,
+    pub committed_replicas: Option<Vec<ScannerPauseBacklogReplicaId>>,
    pub error: Option<String>,
 }

@@ -682,9 +683,9 @@ pub(super) enum ScannerPauseBacklogAttemptDecision {
 }

 #[derive(Clone, Copy, Debug, Deserialize, PartialEq, Eq, Hash, PartialOrd, Ord, Serialize)]
-struct ScannerPauseBacklogReplicaId {
-    pool_index: usize,
-    set_index: usize,
+pub struct ScannerPauseBacklogReplicaId {
+    pub pool_index: usize,
+    pub set_index: usize,
 }

 #[derive(Clone, Debug, Deserialize, PartialEq, Eq, Serialize)]
@@ -795,9 +796,6 @@ impl LoadedScannerPauseBacklog {
    fn status(&self, now: u64, error: Option<String>) -> ScannerPauseBacklogStatus {
        let persistence_unavailable = error.is_some();
        let replica_degraded = self.durable && self.stale_or_unavailable_replicas > 0;
-        let diagnostics_needed =
-            persistence_unavailable || !self.durable || replica_degraded || self.persistence_state != "healthy";
-        let replica_diagnostics = diagnostics_needed.then(|| scanner_pause_backlog_replica_diagnostics(&self.replicas));
        status_from_ledger(
            &self.ledger,
            now,
@@ -809,7 +807,7 @@ impl LoadedScannerPauseBacklog {
            replica_degraded,
            persistence_unavailable,
            error,
-            replica_diagnostics,
+            None,
        )
    }
 }
@@ -979,79 +977,55 @@ async fn read_scanner_pause_backlog_replica(store: Arc<SetDisks>) -> ScannerPaus
    ScannerPauseBacklogReplica { id, revision, state }
 }

-fn scanner_pause_backlog_replica_diagnostics(replicas: &[ScannerPauseBacklogReplica]) -> Vec<ScannerPauseBacklogReplicaStatus> {
+fn scanner_pause_backlog_replica_diagnostics(replicas: Vec<ScannerPauseBacklogReplica>) -> Vec<ScannerPauseBacklogReplicaStatus> {
    replicas
-        .iter()
+        .into_iter()
        .map(|replica| {
-            let revision = match &replica.revision {
-                Some(DataUsageCacheRevision::Etag(etag)) => Some(etag.clone()),
-                Some(DataUsageCacheRevision::Missing) | None => None,
-            };
-            let (
-                state,
-                stable_generation,
-                stable_writer_epoch,
-                committed_generation,
-                committed_writer_epoch,
-                committed_replica_count,
-                error,
-            ) = match &replica.state {
-                ScannerPauseBacklogReplicaState::Missing => {
-                    (ScannerPauseBacklogReplicaStatusState::Missing, None, None, None, None, None, None)
-                }
-                ScannerPauseBacklogReplicaState::Valid(record) => {
-                    let stable = record.stable.as_ref();
-                    let committed = record.committed.as_ref();
-                    (
-                        ScannerPauseBacklogReplicaStatusState::Valid,
-                        stable.map(|ledger| ledger.generation),
-                        stable.map(|ledger| ledger.writer_epoch),
-                        committed.map(|commit| commit.ledger.generation),
-                        committed.map(|commit| commit.ledger.writer_epoch),
-                        committed.map(|commit| commit.replicas.len()),
-                        None,
-                    )
-                }
-                ScannerPauseBacklogReplicaState::Invalid(reason) => (
-                    ScannerPauseBacklogReplicaStatusState::Invalid,
-                    None,
-                    None,
-                    None,
-                    None,
-                    None,
-                    Some(reason.clone()),
-                ),
-                ScannerPauseBacklogReplicaState::FutureSchema(version) => (
-                    ScannerPauseBacklogReplicaStatusState::FutureSchema,
-                    None,
-                    None,
-                    None,
-                    None,
-                    None,
-                    Some(format!("unsupported future schema {version}")),
-                ),
-                ScannerPauseBacklogReplicaState::Unavailable(reason) => (
-                    ScannerPauseBacklogReplicaStatusState::Unavailable,
-                    None,
-                    None,
-                    None,
-                    None,
-                    None,
-                    Some(reason.clone()),
-                ),
-            };
-            ScannerPauseBacklogReplicaStatus {
+            let mut status = ScannerPauseBacklogReplicaStatus {
                pool_index: replica.id.pool_index,
                set_index: replica.id.set_index,
-                state,
-                revision,
-                stable_generation,
-                stable_writer_epoch,
-                committed_generation,
-                committed_writer_epoch,
-                committed_replica_count,
-                error,
+                state: ScannerPauseBacklogReplicaStatusState::Missing,
+                revision: match replica.revision {
+                    Some(DataUsageCacheRevision::Etag(etag)) => Some(etag),
+                    Some(DataUsageCacheRevision::Missing) | None => None,
+                },
+                stable_generation: None,
+                stable_writer_epoch: None,
+                committed_generation: None,
+                committed_writer_epoch: None,
+                committed_replica_count: None,
+                committed_replicas: None,
+                error: None,
+            };
+            match replica.state {
+                ScannerPauseBacklogReplicaState::Missing => {}
+                ScannerPauseBacklogReplicaState::Valid(record) => {
+                    status.state = ScannerPauseBacklogReplicaStatusState::Valid;
+                    if let Some(stable) = record.stable {
+                        status.stable_generation = Some(stable.generation);
+                        status.stable_writer_epoch = Some(stable.writer_epoch);
+                    }
+                    if let Some(committed) = record.committed {
+                        status.committed_generation = Some(committed.ledger.generation);
+                        status.committed_writer_epoch = Some(committed.ledger.writer_epoch);
+                        status.committed_replica_count = Some(committed.replicas.len());
+                        status.committed_replicas = Some(committed.replicas);
+                    }
+                }
+                ScannerPauseBacklogReplicaState::Invalid(reason) => {
+                    status.state = ScannerPauseBacklogReplicaStatusState::Invalid;
+                    status.error = Some(reason);
+                }
+                ScannerPauseBacklogReplicaState::FutureSchema(version) => {
+                    status.state = ScannerPauseBacklogReplicaStatusState::FutureSchema;
+                    status.error = Some(format!("unsupported future schema {version}"));
+                }
+                ScannerPauseBacklogReplicaState::Unavailable(reason) => {
+                    status.state = ScannerPauseBacklogReplicaStatusState::Unavailable;
+                    status.error = Some(reason);
+                }
            }
+            status
        })
        .collect()
 }
@@ -2092,10 +2066,20 @@ pub async fn scanner_pause_backlog_status(storeapi: Arc<ECStore>) -> ScannerPaus
    let replicas = read_scanner_pause_backlog_replicas(storeapi).await;
    let replica_count = replicas.len();
    match select_scanner_pause_backlog_replicas(replicas) {
-        Ok(loaded) => loaded.status(now, runtime_error()),
+        Ok(loaded) => {
+            let mut status = loaded.status(now, runtime_error());
+            if status.error.is_some()
+                || !status.durable
+                || status.stale_or_unavailable_replicas > 0
+                || status.persistence_state != "healthy"
+            {
+                status.replica_diagnostics = Some(scanner_pause_backlog_replica_diagnostics(loaded.replicas));
+            }
+            status
+        }
        Err(failure) => {
-            let replica_diagnostics = scanner_pause_backlog_replica_diagnostics(&failure.replicas);
            let error = runtime_error().map_or(failure.to_string(), |runtime| format!("{failure}; {runtime}"));
+            let replica_diagnostics = scanner_pause_backlog_replica_diagnostics(failure.replicas);
            status_from_ledger(
                &ScannerPauseBacklogLedger::default(),
                now,
@@ -2123,6 +2107,62 @@ mod tests {

    const NATIVE_RETIREMENT_DRIVES_PER_SET: usize = 2;

+    #[test]
+    fn replica_diagnostics_move_snapshot_buffers_and_preserve_json() {
+        let mut stable = durable_ledger(100);
+        stable.generation = 7;
+        stable.writer_epoch = 3;
+        let mut committed = stable.clone();
+        committed.generation = 11;
+        committed.writer_epoch = 5;
+        let members = vec![replica_id(0, 1), replica_id(1, 0)];
+        let members_buffer = members.as_ptr();
+        let revision = "native-revision".to_string();
+        let revision_buffer = revision.as_ptr();
+        let reason = "temporary read error".to_string();
+        let reason_buffer = reason.as_ptr();
+        let record =
+            ScannerPauseBacklogReplicaRecord::new(Some(stable), Some(ScannerPauseBacklogCommitRecord::new(committed, members)));
+        record
+            .validate()
+            .expect("the diagnostic fixture must satisfy the persisted record contract");
+        let diagnostics = scanner_pause_backlog_replica_diagnostics(vec![
+            ScannerPauseBacklogReplica {
+                id: replica_id(0, 1),
+                revision: Some(DataUsageCacheRevision::Etag(revision)),
+                state: ScannerPauseBacklogReplicaState::Valid(Box::new(record)),
+            },
+            ScannerPauseBacklogReplica {
+                id: replica_id(1, 0),
+                revision: None,
+                state: ScannerPauseBacklogReplicaState::Unavailable(reason),
+            },
+        ]);
+        assert_eq!(diagnostics[0].revision.as_ref().expect("revision").as_ptr(), revision_buffer);
+        assert_eq!(diagnostics[0].committed_replicas.as_ref().expect("membership").as_ptr(), members_buffer);
+        assert_eq!(diagnostics[1].error.as_ref().expect("read error").as_ptr(), reason_buffer);
+        assert_eq!(
+            serde_json::to_value(diagnostics).expect("diagnostics must serialize"),
+            serde_json::json!([
+                {
+                    "pool_index": 0, "set_index": 1, "state": "valid", "revision": "native-revision",
+                    "stable_generation": 7, "stable_writer_epoch": 3,
+                    "committed_generation": 11, "committed_writer_epoch": 5,
+                    "committed_replica_count": 2,
+                    "committed_replicas": [{"pool_index": 0, "set_index": 1}, {"pool_index": 1, "set_index": 0}],
+                    "error": null
+                },
+                {
+                    "pool_index": 1, "set_index": 0, "state": "unavailable", "revision": null,
+                    "stable_generation": null, "stable_writer_epoch": null,
+                    "committed_generation": null, "committed_writer_epoch": null,
+                    "committed_replica_count": null, "committed_replicas": null,
+                    "error": "temporary read error"
+                }
+            ])
+        );
+    }
+
    fn run_native_retirement_test<C, F>(case: C)
    where
        C: FnOnce() -> F + Send + 'static,
@@ -2148,13 +2188,17 @@ mod tests {
    }

    async fn native_retirement_store() -> (tempfile::TempDir, Arc<ECStore>) {
+        native_pause_backlog_store(3, 2).await
+    }
+
+    async fn native_pause_backlog_store(pool_count: usize, sets_per_pool: usize) -> (tempfile::TempDir, Arc<ECStore>) {
        register_scanner_pause_backlog_retirement();
        let root = tempfile::tempdir().expect("native retirement fixture directory");
        let store = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit(
            root.path(),
            false,
-            3,
-            2,
+            pool_count,
+            sets_per_pool,
            NATIVE_RETIREMENT_DRIVES_PER_SET,
            false,
        )
@@ -2162,6 +2206,129 @@ mod tests {
        (root, store)
    }

+    #[test]
+    #[serial_test::serial]
+    fn native_single_pool_partial_commit_converges_across_24_sets() {
+        run_native_retirement_test(async || {
+            set_runtime_error(None);
+            let (_root, store) = native_pause_backlog_store(1, 24).await;
+            let now = unix_now();
+            let mut controller = ScannerPauseBacklogController::claim(Arc::clone(&store), now)
+                .await
+                .expect("bootstrap the single-pool ledger");
+            controller.observe(observation(now + 1, true, 7)).await;
+            controller.observe(observation(now + 2, false, 7)).await;
+            let stable = controller.loaded.ledger.clone();
+            assert!(stable.pending_full_scan);
+            let ids = scanner_pause_backlog_replica_ids(&controller.loaded.replicas);
+            assert_eq!(ids.len(), 24);
+            let old_commit = controller
+                .loaded
+                .authoritative_commit
+                .clone()
+                .expect("initial full membership proof");
+            drop(controller);
+            let candidate = claim_scanner_pause_backlog_writer(&stable, now + 3).expect("interrupted writer generation");
+            let new_commit = ScannerPauseBacklogCommitRecord::new(candidate, ids.clone());
+            for (index, id) in ids.iter().enumerate() {
+                let commit = if index < 12 { &new_commit } else { &old_commit };
+                replace_native_pause_backlog_replica(
+                    &store,
+                    *id,
+                    ScannerPauseBacklogReplicaRecord::new(Some(stable.clone()), Some(commit.clone())),
+                )
+                .await;
+            }
+
+            let metric_status = load_scanner_pause_backlog(Arc::clone(&store))
+                .await
+                .expect("the common rollback must be selectable")
+                .status(now + 3, None);
+            assert_eq!(metric_status.persistence_state, "rolled_back_partial_commit");
+            assert!(
+                metric_status.replica_diagnostics.is_none(),
+                "metrics must not allocate discarded replica diagnostics"
+            );
+            let status = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(status.persistence_state, "rolled_back_partial_commit");
+            assert_eq!(status.healthy_replicas, 0);
+            assert_eq!(status.replica_count, 24);
+            assert_eq!(status.generation, stable.generation);
+            let diagnostics = status.replica_diagnostics.expect("partial commit must expose every replica");
+            assert_eq!(diagnostics.len(), 24);
+            assert!(
+                diagnostics
+                    .iter()
+                    .all(|replica| replica.committed_replicas.as_ref() == Some(&ids))
+            );
+            let expected = claim_scanner_pause_backlog_writer(&stable, now + 4).expect("recovery writer generation");
+            let recovered = ScannerPauseBacklogController::claim(Arc::clone(&store), now + 4)
+                .await
+                .expect("the common stable rollback must converge without resetting the ledger");
+            assert_current_native_writer_ledger(&store, &expected).await;
+            assert!(recovered.loaded.ledger.pending_full_scan);
+            assert_eq!(recovered.loaded.ledger.dirty_usage_buckets, 7);
+            assert_eq!(recovered.loaded.ledger.writer_epoch, stable.writer_epoch + 1);
+            drop(recovered);
+            shutdown_native_retirement_store(store).await;
+            set_runtime_error(None);
+        });
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn native_single_pool_conflicting_stable_proofs_preserve_all_24_sets() {
+        run_native_retirement_test(async || {
+            set_runtime_error(None);
+            let (_root, store) = native_pause_backlog_store(1, 24).await;
+            let now = unix_now();
+            let controller = ScannerPauseBacklogController::claim(Arc::clone(&store), now)
+                .await
+                .expect("bootstrap the single-pool ledger");
+            let stable = controller.loaded.ledger.clone();
+            let ids = scanner_pause_backlog_replica_ids(&controller.loaded.replicas);
+            assert_eq!(ids.len(), 24);
+            drop(controller);
+            let first = claim_scanner_pause_backlog_writer(&stable, now + 1).expect("first valid fork");
+            let second = claim_scanner_pause_backlog_writer(&stable, now + 2).expect("second valid fork");
+            for (index, id) in ids.iter().enumerate() {
+                let ledger = if index < 12 { &first } else { &second };
+                let commit = ScannerPauseBacklogCommitRecord::new(ledger.clone(), ids.clone());
+                replace_native_pause_backlog_replica(
+                    &store,
+                    *id,
+                    ScannerPauseBacklogReplicaRecord::new(Some(ledger.clone()), Some(commit)),
+                )
+                .await;
+            }
+            let mut before = Vec::with_capacity(24);
+            for set in &store.pools[0].disk_set {
+                before.push(native_replica_bytes(set).await);
+            }
+            let error = match ScannerPauseBacklogController::claim(Arc::clone(&store), now + 3).await {
+                Ok(_) => panic!("conflicting rollback proofs must never be reset or arbitrarily selected"),
+                Err(error) => error,
+            };
+            assert!(
+                error.contains("neither a surviving membership commit nor a stable rollback point"),
+                "{error}"
+            );
+            let status = scanner_pause_backlog_status(Arc::clone(&store)).await;
+            assert_eq!(status.persistence_state, "unavailable");
+            assert_eq!(status.replica_count, 24);
+            assert_eq!(status.replica_diagnostics.as_ref().map(Vec::len), Some(24));
+            for (index, set) in store.pools[0].disk_set.iter().enumerate() {
+                assert_eq!(
+                    native_replica_bytes(set).await,
+                    before[index],
+                    "failed recovery must preserve bytes and CAS revisions"
+                );
+            }
+            shutdown_native_retirement_store(store).await;
+            set_runtime_error(None);
+        });
+    }
+
    #[test]
    #[serial_test::serial]
    fn native_status_reports_missing_and_conflicting_pause_backlog_replicas() {
@@ -2278,7 +2445,7 @@ mod tests {
                    && replica["error"].as_str().is_some_and(|reason| !reason.is_empty())
            }));

-            let unavailable = scanner_pause_backlog_replica_diagnostics(&[ScannerPauseBacklogReplica {
+            let unavailable = scanner_pause_backlog_replica_diagnostics(vec![ScannerPauseBacklogReplica {
                id: replica_id(9, 9),
                revision: None,
                state: ScannerPauseBacklogReplicaState::Unavailable("read failed".to_string()),
@@ -61,7 +61,7 @@ pub struct ListPartsInfo {
    pub upload_id: String,
    pub storage_class: String,
    pub part_number_marker: usize,
-    pub next_part_number_marker: usize,
+    pub next_part_number_marker: Option<usize>,
    pub max_parts: usize,
    pub is_truncated: bool,
    pub parts: Vec<PartInfo>,
@@ -33,6 +33,9 @@ expansion, rebalance, decommission, heal, drive replacement, restart
 recovery, and the `rc` CLI mapping), start with
 [Cluster and erasure-coding lifecycle operations](operations/cluster-lifecycle-operations.md).

+For concurrent bucket creation and deletion, see
+[Bucket operation admission](operations/bucket-operation-admission.md).
+
 For persisted administrator bucket tasks and bucket recreation, see
 [Bucket heal recovery](operations/bucket-heal-recovery.md).

@@ -0,0 +1,16 @@
+# Bucket creation and deletion admission
+
+`CreateBucket` and `DeleteBucket` share a process-wide execution budget of eight active transactions. Short bursts wait in a FIFO queue with space for 128 waiting requests. Each request can wait up to 30 seconds for an execution slot. These limits protect storage and namespace-lock resources while allowing parallel test workers to create isolated buckets.
+
+A request that finds the queue full or exceeds the queue wait returns HTTP 503 with the S3 error code `SlowDown` and `Retry-After: 1`. This is an overload response, not a bucket-name or filesystem error. The active and waiting limits are shared by creation and deletion and are currently fixed.
+
+Waiting requests do not start storage mutations. Disconnecting while waiting releases the queue slot. Once admitted, the transaction owns its execution slot until the mutation and post-commit hooks finish, including when its caller disconnects.
+
+For automated tests and applications:
+
+- Keep SDK retries enabled and use bounded exponential backoff with jitter for `SlowDown`. Honor `Retry-After` as a minimum delay when implementing your own retry policy.
+- Set the client request timeout above the queue wait plus the expected operation time. The 30-second limit covers admission waiting only; it does not limit transaction execution.
+- Reduce the client concurrency if bursts exceed the waiting capacity or repeatedly time out. A bucket per test worker remains a supported isolation pattern.
+- After an ambiguous network timeout, check the bucket state before assuming the operation failed: an admitted transaction may finish after the client stops waiting.
+
+The serving implementation is `BucketOperationAdmission` in `rustfs/src/app/bucket_usecase.rs`. This budget does not change object PUT/GET admission, erasure quorum, or persistence rules.
@@ -217,10 +217,18 @@ If a set has lost more drives than its parity, heal cannot rebuild; recover from
 - **Ordinary restart.** No topology change in flight, identical arguments on every node, one node at a time per [rolling-restart.md](rolling-restart.md). Startup reloads pool metadata, `rebalance.bin`, the decommission queue, and heal recovery records, and resumes workers by leader and owner rules.
 - **Rebalance.** Read `/rebalance/status` first. Handle active or stopping pools, `lastError`, cleanup warnings, and stop propagation before deciding to stop or restart. Do not delete `rebalance.bin`.
 - **Decommission.** Non-terminal entries resume; failed and canceled entries survive restarts until cleared or retried through the allowed path (§9).
- **Scanner pause backlog.** For an unavailable or degraded pause backlog, query `/rustfs/admin/v3/scanner/status` on each node and inspect `pause_backlog.replica_diagnostics`. The entries identify each writable pool/set, its `missing`, `valid`, `invalid`, `future_schema`, or `unavailable` state, object revision (the CAS token for that record), stable and committed generations, committed member count, and any read/decode error. Compare revisions and generations to locate divergent proofs, and compare pool order and endpoints across nodes before diagnosing membership conflicts. Keep the existing backlog records and attach the complete status output when escalating; do not delete or reset `.scanner-pause-backlog.json`. There is no supported force-reset for an authority conflict because it may discard pending data-movement work. A conflict remains fail-closed until an authority-safe recovery is available.
+- **Scanner pause backlog.** An unreadable or unprovable `.scanner-pause-backlog.json` gates scanner cycles, including a pending usage full rebuild. Follow [the diagnostic and recovery procedure](#111-scanner-pause-backlog-recovery) below; keep the original records intact.
 - **Root heal.** A graceful shutdown persists an unfinished cluster heal as `root-heal-<task-id>.json` in `.rustfs.sys` (`RootHealRecovery` in `crates/heal/src/heal/manager/root_recovery.rs`, 64 KiB bound) and replays it with the same task id. Terminal records use `terminal-root-heal-`; invalid, oversized, or unsupported records move to `quarantined-root-heal-intent-` or `quarantined-root-heal-terminal-` and are skipped with a log line until an operator repairs them. Do not delete quarantine markers or reuse a task id.
 - **Unclean shutdown.** When the previous run left its marker behind, the heal manager's `process_unclean_shutdown` (`crates/heal/src/heal/manager/unclean_shutdown.rs`) enqueues a full erasure-set heal for every local set; graceful shutdown clears the marker. See [durability-modes.md](durability-modes.md) for which sync tier can leave what behind.

+### 11.1 Scanner pause backlog recovery
+
+1. Record the running version and build commit on every node. Confirm that every potential scanner leader includes serialized replica publication and cancellation-safe publication ownership; a version label or successful rolling restart alone does not prove this. An older writer can still use the earlier concurrent publication path during a mixed-version rollout.
+2. Query `/rustfs/admin/v3/scanner/status` on each node, identify the leader, and retain complete, timestamped snapshots before and after its next persistence retry. For an unavailable or degraded backlog, `pause_backlog.replica_diagnostics` lists every currently writable `(pool_index, set_index)`, its `missing`, `valid`, `invalid`, `future_schema`, or `unavailable` state, its object revision (CAS token), stable and committed generations/epochs, and read/decode errors. `committed_replicas` names the membership claimed by that record, while `committed_replica_count` gives its size. Compare this list with the actual configured topology on every node; a reported set count differing from `replica_count` needs investigation before recovery.
+3. Restore access to unavailable members and resolve object-lock or storage failures without editing the ledger. Publication is serialized to avoid competition within one replica cohort; external contention can still make a write fail. The leader retries failed claims on the five-minute persistence schedule; read/write duration can consume that delay, so a slow attempt does not guarantee another five-minute wait after it returns.
+4. A surviving, complete membership commit or identical stable rollback values across all current replicas permit automatic recovery. The writer stabilizes that proven ledger using each member's current CAS revision, verifies all members by readback, and advances its writer epoch/generation. `rolled_back_partial_commit` can therefore recover without an operator reset. Its `healthy_replicas: 0` means no record currently matches the selected stable/commit combination; it does not mean every object is missing. Status reads can also straddle publication phases, so a single transient snapshot is not a completed recovery proof. Confirm repeated `healthy` status with no persistence error, a successfully advanced writer epoch/generation, and then scanner cycle progress; pending full-scan work must remain until it completes.
+5. If the writer cannot prove either a complete membership commit or a common stable rollback, preserve the original objects and revisions and escalate with the snapshots, build commits, topology, and lock/storage errors. Invalid or newer-schema records require a compatible reader or supported repair; conflicting records must not be resolved by choosing the largest generation. There is no supported force-reset for lost authority. Do not delete, copy over, or manually edit `.scanner-pause-backlog.json`; doing so may discard pending data-movement work. `usage-state/reset` requests a usage rebuild and does not repair or bypass this ledger gate.
+
 ## 12. Mutual exclusion and prohibited actions

 | Operation | With rebalance | With decommission | Notes |
@@ -118,7 +118,7 @@ Post-merge and tag-driven; not a substitute for a PR gate.
 |---|---|---|
 | Push to `main`, weekly schedule, dispatch | `build.yml` `build-rustfs` (a development build on a main push restricts the matrix to the Linux targets) | build artifacts; no release publication |
 | Valid release or preview tag | `build.yml` `build-rustfs`, `create-release`, `upload-release-assets`, `publish-release` | draft release, checksummed assets, publish |
-| Successful non-preview release-tag build (`workflow_run`) | `docker.yml` `build-docker`, `scan-docker-image`, `sync-dockerhub-description` | multi-architecture images, vulnerability report, and the Docker Hub overview republished from `README.md` |
+| Successful non-preview release-tag build (`workflow_run`) | `docker.yml` `build-docker`, `scan-docker-image` | multi-architecture images and vulnerability report |
 | Successful release-tag build (`workflow_run`) | `package.yml` `package` | DEB/RPM packages and checksums uploaded to the release |
 | Successful non-preview release-tag build (`workflow_run`) | `helm-package.yml` `build-helm-package`, `publish-helm-package` | versioned chart and repository index |
 | Final tag's release published | `build.yml` `cleanup-preview-releases` | deletes every `<target>-preview.<N>` Release for that target; the tags are kept |
@@ -144,17 +144,6 @@ dependencies = [
 "pin-project-lite",
 ]

-[[package]]
-name = "async-recursion"
-version = "1.1.1"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3b43422f69d8ff38f95f1b2bb76517c91589a924d1559a0e935d7c8ce0274c11"
-dependencies = [
- "proc-macro2",
- "quote",
- "syn 2.0.118",
-]
-
 [[package]]
 name = "async-trait"
 version = "0.1.91"
@@ -4135,7 +4124,6 @@ dependencies = [
 "aes-gcm",
 "arc-swap",
 "async-channel",
- "async-recursion",
 "async-trait",
 "aws-config",
 "aws-credential-types",
@@ -131,15 +131,67 @@ use std::{
    future::Future,
    io::Write,
    sync::{Arc, LazyLock},
+    time::Duration,
 };
-use tokio::sync::{Semaphore, TryAcquireError};
+use tokio::sync::{OwnedSemaphorePermit, Semaphore, TryAcquireError};
 use tracing::{Instrument as _, debug, error, info, instrument, warn};

 const LOG_COMPONENT_APP: &str = "app";
 const LOG_SUBSYSTEM_BUCKET: &str = "bucket";
 const BUCKET_OPERATION_CONCURRENCY: usize = 8;
-static BUCKET_OPERATION_ADMISSION: LazyLock<Arc<Semaphore>> =
-    LazyLock::new(|| Arc::new(Semaphore::new(BUCKET_OPERATION_CONCURRENCY)));
+const BUCKET_OPERATION_QUEUE_CAPACITY: usize = 128;
+const BUCKET_OPERATION_QUEUE_TIMEOUT: Duration = Duration::from_secs(30);
+static BUCKET_OPERATION_ADMISSION: LazyLock<Arc<BucketOperationAdmission>> = LazyLock::new(|| {
+    Arc::new(BucketOperationAdmission::new(
+        BUCKET_OPERATION_CONCURRENCY,
+        BUCKET_OPERATION_QUEUE_CAPACITY,
+    ))
+});
+
+struct BucketOperationAdmission {
+    active: Arc<Semaphore>,
+    waiting: Semaphore,
+}
+
+impl BucketOperationAdmission {
+    fn new(concurrency: usize, queue_capacity: usize) -> Self {
+        Self {
+            active: Arc::new(Semaphore::new(concurrency)),
+            waiting: Semaphore::new(queue_capacity),
+        }
+    }
+
+    async fn acquire(&self, operation: &'static str) -> S3Result<OwnedSemaphorePermit> {
+        match self.active.clone().try_acquire_owned() {
+            Ok(permit) => return Ok(permit),
+            Err(TryAcquireError::Closed) => return Err(bucket_admission_closed(operation)),
+            Err(TryAcquireError::NoPermits) => {}
+        }
+
+        // Wait on the request task; only admitted mutations become detached tasks.
+        // The queue permit bounds waiting requests and is released on cancellation.
+        let _waiting = self.waiting.try_acquire().map_err(|err| match err {
+            TryAcquireError::NoPermits => bucket_admission_slow_down(operation, "queue is full"),
+            TryAcquireError::Closed => bucket_admission_closed(operation),
+        })?;
+        tokio::time::timeout(BUCKET_OPERATION_QUEUE_TIMEOUT, self.active.clone().acquire_owned())
+            .await
+            .map_err(|_| bucket_admission_slow_down(operation, "queue wait timed out"))?
+            .map_err(|_| bucket_admission_closed(operation))
+    }
+}
+
+fn bucket_admission_closed(operation: &'static str) -> S3Error {
+    S3Error::with_message(S3ErrorCode::InternalError, format!("{operation} admission closed"))
+}
+
+fn bucket_admission_slow_down(operation: &'static str, reason: &'static str) -> S3Error {
+    let mut error = S3Error::with_message(S3ErrorCode::SlowDown, format!("{operation} {reason}; retry later"));
+    let mut headers = http::HeaderMap::new();
+    headers.insert(http::header::RETRY_AFTER, http::HeaderValue::from_static("1"));
+    error.set_headers(headers);
+    error
+}
 use urlencoding::encode;

 type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
@@ -1307,19 +1359,14 @@ where

 async fn await_bucket_usecase_on_fresh_task_with_admission<T, F>(
    operation: &'static str,
-    admission: Arc<Semaphore>,
+    admission: Arc<BucketOperationAdmission>,
    future: F,
 ) -> S3Result<T>
 where
    T: Send + 'static,
    F: Future<Output = S3Result<T>> + Send + 'static,
 {
-    let permit = admission.try_acquire_owned().map_err(|err| match err {
-        TryAcquireError::NoPermits => {
-            S3Error::with_message(S3ErrorCode::SlowDown, format!("{operation} concurrency limit reached; retry later"))
-        }
-        TryAcquireError::Closed => S3Error::with_message(S3ErrorCode::InternalError, format!("{operation} admission closed")),
-    })?;
+    let permit = admission.acquire(operation).await?;
    tokio::spawn(
        async move {
            let _permit = permit;
@@ -3239,7 +3286,7 @@ mod tests {

    #[tokio::test]
    async fn bucket_usecase_task_finishes_post_commit_hooks_after_parent_cancellation() {
-        let admission = Arc::new(Semaphore::new(1));
+        let admission = Arc::new(BucketOperationAdmission::new(1, 1));
        let committed = Arc::new(Notify::new());
        let committed_wait = committed.notified();
        let release_hook = Arc::new(Notify::new());
@@ -3274,45 +3321,39 @@ mod tests {
            hook_ran.load(Ordering::SeqCst),
            "the fresh task must own both the storage mutation and its post-commit hooks"
        );
-        assert_eq!(admission.available_permits(), 1);
+        assert_eq!(admission.active.available_permits(), 1);
    }

-    #[tokio::test]
-    async fn saturated_bucket_usecase_admission_does_not_start_work() {
-        let admission = Arc::new(Semaphore::new(1));
-        let held = admission
-            .clone()
-            .acquire_owned()
-            .await
-            .expect("test admission should remain open");
+    #[tokio::test(start_paused = true)]
+    async fn bucket_usecase_admission_times_out_without_starting_work() {
+        let admission = Arc::new(BucketOperationAdmission::new(1, 1));
+        let held = admission.active.clone().acquire_owned().await.expect("hold active slot");
        let started = Arc::new(AtomicBool::new(false));
        let started_for_task = started.clone();
-        let result = tokio::time::timeout(
-            Duration::from_secs(1),
-            await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission, async move {
-                started_for_task.store(true, Ordering::SeqCst);
-                Ok(())
-            }),
-        )
-        .await
-        .expect("saturated admission must fail within the bounded timeout");
+        let result = await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission.clone(), async move {
+            started_for_task.store(true, Ordering::SeqCst);
+            Ok(())
+        })
+        .await;
+        let error = result.expect_err("queued request must time out");
+        assert_eq!(error.code(), &S3ErrorCode::SlowDown);
+        assert_eq!(error.headers().expect("retry headers")[http::header::RETRY_AFTER], "1");
+        assert!(!started.load(Ordering::SeqCst), "timed-out work must not start");
+        assert_eq!(admission.waiting.available_permits(), 1);
        drop(held);
-
-        assert!(
-            !started.load(Ordering::SeqCst),
-            "saturated admission must not start a detached bucket operation"
-        );
-        assert_eq!(result.expect_err("saturated admission must fail fast").code(), &S3ErrorCode::SlowDown);
+        assert_eq!(admission.active.available_permits(), 1);
    }

    #[tokio::test]
-    async fn bucket_usecase_admission_rejects_excess_detached_tasks() {
-        let admission = Arc::new(Semaphore::new(BUCKET_OPERATION_CONCURRENCY));
+    async fn bucket_usecase_admission_queues_bursts_with_bounded_execution() {
+        let admission = Arc::new(BucketOperationAdmission::new(
+            BUCKET_OPERATION_CONCURRENCY,
+            BUCKET_OPERATION_QUEUE_CAPACITY,
+        ));
        let release = Arc::new(Semaphore::new(0));
        let started = Arc::new(AtomicUsize::new(0));
-        let mut tasks = Vec::with_capacity(BUCKET_OPERATION_CONCURRENCY);
-
-        for _ in 0..BUCKET_OPERATION_CONCURRENCY {
+        let mut tasks = Vec::with_capacity(100);
+        for _ in 0..100 {
            let admission_for_task = admission.clone();
            let release_for_task = release.clone();
            let started_for_task = started.clone();
@@ -3321,53 +3362,82 @@ mod tests {
                admission_for_task,
                async move {
                    started_for_task.fetch_add(1, Ordering::SeqCst);
-                    let _release = release_for_task
-                        .acquire()
-                        .await
-                        .expect("test release gate should remain open");
+                    release_for_task.acquire().await.expect("release gate").forget();
                    Ok(())
                },
            )));
        }
-
-        tokio::time::timeout(Duration::from_secs(1), async {
-            while started.load(Ordering::SeqCst) != BUCKET_OPERATION_CONCURRENCY {
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while admission.waiting.available_permits() != BUCKET_OPERATION_QUEUE_CAPACITY - 92
+                || started.load(Ordering::SeqCst) != BUCKET_OPERATION_CONCURRENCY
+            {
                tokio::task::yield_now().await;
            }
        })
        .await
-        .expect("all admitted operations should start");
-
-        let ninth_started = Arc::new(AtomicBool::new(false));
-        let ninth_started_for_task = ninth_started.clone();
-        let ninth_result = tokio::time::timeout(
-            Duration::from_secs(1),
-            await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission.clone(), async move {
-                ninth_started_for_task.store(true, Ordering::SeqCst);
-                Ok(())
-            }),
-        )
-        .await
-        .expect("an excess bucket transaction must fail within the bounded timeout");
-        assert!(
-            !ninth_started.load(Ordering::SeqCst),
-            "the ninth bucket transaction must not start when admission is saturated"
-        );
-        assert_eq!(
-            ninth_result.expect_err("the ninth bucket transaction must fail fast").code(),
-            &S3ErrorCode::SlowDown
-        );
-
-        release.add_permits(BUCKET_OPERATION_CONCURRENCY);
+        .expect("all excess operations should queue");
+        assert_eq!(started.load(Ordering::SeqCst), BUCKET_OPERATION_CONCURRENCY);
+        release.add_permits(100);
        for task in tasks {
-            task.await
-                .expect("bucket transaction parent should join")
-                .expect("bucket transaction should succeed");
+            task.await.expect("request task joins").expect("burst operation succeeds");
        }
+        assert_eq!(started.load(Ordering::SeqCst), 100);
+        assert_eq!(admission.active.available_permits(), BUCKET_OPERATION_CONCURRENCY);
+        assert_eq!(admission.waiting.available_permits(), BUCKET_OPERATION_QUEUE_CAPACITY);
+    }

+    #[tokio::test]
+    async fn bucket_usecase_admission_bounds_queue_and_cancels_waiters() {
+        let admission = Arc::new(BucketOperationAdmission::new(1, 1));
+        let held = admission.active.clone().acquire_owned().await.expect("hold active slot");
+        let started = Arc::new(AtomicBool::new(false));
+        let started_for_task = started.clone();
+        let queued = tokio::spawn(await_bucket_usecase_on_fresh_task_with_admission(
+            "test bucket operation",
+            admission.clone(),
+            async move {
+                started_for_task.store(true, Ordering::SeqCst);
+                Ok(())
+            },
+        ));
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while admission.waiting.available_permits() != 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("request should queue");
+        let error = admission
+            .acquire("test bucket operation")
+            .await
+            .expect_err("full queue rejects immediately");
+        assert_eq!(error.code(), &S3ErrorCode::SlowDown);
+        assert_eq!(error.headers().expect("retry headers")[http::header::RETRY_AFTER], "1");
+        let response = error.to_http_response().expect("serialize admission error");
+        assert_eq!(response.status(), http::StatusCode::SERVICE_UNAVAILABLE);
+        assert_eq!(response.headers()[http::header::RETRY_AFTER], "1");
+        use http_body_util::BodyExt as _;
+        let body = response.into_body().collect().await.expect("read error XML").to_bytes();
+        assert!(String::from_utf8_lossy(&body).contains("<Code>SlowDown</Code>"));
+        queued.abort();
+        assert!(queued.await.expect_err("queued request was cancelled").is_cancelled());
+        assert_eq!(admission.waiting.available_permits(), 1);
+        drop(held);
+        assert!(!started.load(Ordering::SeqCst), "cancelled waiter must not mutate storage");
        await_bucket_usecase_on_fresh_task_with_admission("test bucket operation", admission, async { Ok(()) })
            .await
-            .expect("admission must recover after active transactions finish");
+            .expect("admission recovers after cancellation");
+    }
+
+    #[tokio::test]
+    async fn bucket_usecase_admission_maps_closed_semaphore_to_internal_error() {
+        let admission = BucketOperationAdmission::new(1, 1);
+        admission.active.close();
+        let error = admission
+            .acquire("test bucket operation")
+            .await
+            .expect_err("closed admission must fail");
+        assert_eq!(error.code(), &S3ErrorCode::InternalError);
    }

    #[tokio::test]
@@ -71,7 +71,7 @@ pub(crate) fn build_list_parts_output(res: ListPartsInfo) -> ListPartsOutput {
        owner: Some(owner),
        initiator: Some(initiator),
        is_truncated: Some(res.is_truncated),
-        next_part_number_marker: res.next_part_number_marker.try_into().ok(),
+        next_part_number_marker: res.next_part_number_marker.and_then(|v| v.try_into().ok()),
        max_parts: res.max_parts.try_into().ok(),
        part_number_marker: res.part_number_marker.try_into().ok(),
        storage_class: Some(
@@ -228,7 +228,7 @@ mod tests {
            upload_id: "upload-a".to_string(),
            storage_class: "STANDARD".to_string(),
            part_number_marker: 1,
-            next_part_number_marker: 2,
+            next_part_number_marker: Some(2),
            max_parts: 1000,
            is_truncated: true,
            parts: vec![PartInfo {
@@ -259,6 +259,55 @@ mod tests {
        assert_eq!(output.initiator, Some(rustfs_initiator()));
    }

+    #[test]
+    fn test_list_parts_output_xml_omits_terminal_marker() {
+        for (marker, part_numbers) in [(0, vec![]), (3, vec![10]), (0, vec![1, 3])] {
+            let output = build_list_parts_output(ListPartsInfo {
+                part_number_marker: marker,
+                max_parts: 2,
+                parts: part_numbers
+                    .into_iter()
+                    .map(|part_num| PartInfo {
+                        part_num,
+                        ..Default::default()
+                    })
+                    .collect(),
+                ..Default::default()
+            });
+
+            assert_eq!(output.is_truncated, Some(false));
+            assert_eq!(output.next_part_number_marker, None);
+            let xml = String::from_utf8(crate::storage::storage_api::serialize(&output).expect("output should serialize"))
+                .expect("XML should be UTF-8");
+            assert!(xml.contains("<IsTruncated>false</IsTruncated>"));
+            assert!(xml.contains(&format!("<PartNumberMarker>{marker}</PartNumberMarker>")));
+            assert!(!xml.contains("NextPartNumberMarker"), "terminal page must omit the token: {xml}");
+        }
+    }
+
+    #[test]
+    fn test_list_parts_output_xml_preserves_truncated_marker() {
+        let output = build_list_parts_output(ListPartsInfo {
+            is_truncated: true,
+            next_part_number_marker: Some(3),
+            max_parts: 2,
+            parts: [1, 3]
+                .into_iter()
+                .map(|part_num| PartInfo {
+                    part_num,
+                    ..Default::default()
+                })
+                .collect(),
+            ..Default::default()
+        });
+
+        let xml = String::from_utf8(crate::storage::storage_api::serialize(&output).expect("output should serialize"))
+            .expect("XML should be UTF-8");
+        assert!(xml.contains("<IsTruncated>true</IsTruncated>"));
+        assert!(xml.contains("<NextPartNumberMarker>3</NextPartNumberMarker>"));
+        assert_eq!(output.parts.as_ref().expect("parts should be present").len(), 2);
+    }
+
    #[test]
    fn test_list_parts_output_reports_logical_size_for_compressed_parts() {
        let input = ListPartsInfo {
@@ -374,7 +423,7 @@ mod tests {
        let input = ListPartsInfo {
            storage_class: "STANDARD_IA".to_string(),
            part_number_marker: usize::MAX,
-            next_part_number_marker: usize::MAX,
+            next_part_number_marker: Some(usize::MAX),
            max_parts: usize::MAX,
            parts: vec![PartInfo {
                part_num: usize::MAX,
Author	SHA1	Message	Date
overtrue	6de965ae3c	test(e2e): register paginator regression in Darwin inventory	2026-10-03 07:07:41 +08:00
overtrue	7c56853067	fix(storage): publish delete rollback backups atomically Stage rollback metadata outside the rollback directory and publish it only after the full write succeeds. A short write must not leave a backup that quorum rollback can rename over acknowledged version history. Add an isolated real short-write regression and register the backported ListParts SDK test in the smoke and Linux full inventories.	2026-10-03 06:43:25 +08:00
Hauser	6f61761c7b	fix(s3): honor sparse ListParts markers and verify termination Resume part listings at the first part above the numeric marker, even when that marker is absent. Use binary search over the sorted part numbers and retain the existing exact-tail empty-slice path. Add storage, XML, and real AWS SDK paginator regressions for empty and terminal pages, sparse markers, and multipart completion integrity. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> (cherry picked from commit `673031eea1`)	2026-10-03 06:32:24 +08:00
Chapman	1bae764530	fix: emit NextPartNumberMarker only when ListParts is truncated ListPartsInfo.next_part_number_marker was a non-optional usize that defaulted to 0 and was only assigned when the response was truncated. The S3 serializer then emitted it unconditionally as Some(0), causing AWS SDK paginators to loop infinitely on part_number_marker=0 instead of terminating. Change the field to Option<usize> (None by default) and set it only inside the is_truncated branch. The S3 output layer now uses .and_then() so NextPartNumberMarker is absent when IsTruncated=false, matching AWS S3 behavior. Fixes #8208 (cherry picked from commit `44de803a38`)	2026-10-03 06:32:24 +08:00
overtrue	7150ed2066	ci: remove Docker Hub description sync	2026-10-02 19:59:28 +08:00
overtrue	c04d566b26	fix(heal): wait for held legacy responsibility in replay test	2026-10-02 03:13:15 +08:00
Hauser	4dd267940c	fix(scanner): diagnose and verify pause backlog recovery (#8293 ) * fix(scanner): diagnose and verify pause backlog recovery Expose the retained replica snapshot and claimed membership in abnormal admin status responses. Keep diagnostics off metrics updates and verify single-pool recovery and conflicting-proof preservation across 24 sets. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(scanner): move replica snapshots into diagnostics Consume the terminal admin read snapshot in a single state match and move membership, revision, and error buffers into the response. Verify buffer handoff and the unchanged JSON contract without altering ledger authority. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-02 01:09:00 +08:00
Hauser	2aad57c984	chore: merge upstream release branch Integrate the updated release/1.0.1 branch before pushing the dependency and atomic API maintenance changes. Preserve the upstream removal of async-recursion. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com>	2026-10-02 01:00:04 +08:00
Hauser	dc79582e40	improve	2026-10-02 00:54:32 +08:00
Hauser	25fb3d2516	chore: refresh dependencies and atomic update APIs Update workspace dependencies and the lockfile. Replace deprecated atomic fetch_update aliases with try_update while preserving closures and memory ordering. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com>	2026-10-02 00:46:57 +08:00
Hauser	5c39ed1495	fix(s3): queue bucket operations and restore strict Clippy checks (#8290 ) * fix(s3): queue concurrent bucket creation and deletion Keep eight active bucket transactions and bound admission waiting to 128 requests and 30 seconds. Preserve detached transaction ownership and return Retry-After with overload responses. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(ecstore): restore strict Clippy compatibility on Rust 1.99 Use try_update without changing atomic ordering or overflow behavior. Keep recursive storage futures boxed once at each frame and remove the redundant async-recursion macro, including its non-recursive SQL planner use. Remove needless closure borrows and orphaned dependency entries. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-02 00:41:05 +08:00