fix(storage): publish delete rollback backups atomically

Write rollback backups to a temporary file outside the rollback directory and publish only complete writes. Preserve acknowledged versions when backup creation fails partway through. Add an isolated child-process short-write regression that verifies all 100 versions survive rollback and a subsequent write preserves 101 versions.
fix(connect): send endpoint origin for relay session authentication (#8310 )
2026-11-01 16:50:45 +00:00 · 2026-10-03 06:47:48 +08:00 · 2026-10-03 01:06:25 +08:00 · 2026-10-02 20:04:45 +08:00 · 2026-10-02 10:11:22 +08:00 · 2026-10-02 05:43:24 +08:00
27 changed files with 1216 additions and 160 deletions
@@ -1,6 +1,6 @@
 ---
 name: rustfs-release-publish
-description: "Run the RustFS console gate, source version bump, preview validation, human confirmation, final-tag publication, and post-release installation and website updates. Use only when the user explicitly asks to release or publish a RustFS version (发版/发布)."
+description: "Run the RustFS console gate, source version bump, release-branch CI, preview validation, human confirmation, final-tag publication, and post-release installation and website updates. Use only when the user explicitly asks to release or publish a RustFS version (发版/发布)."
 ---
 # RustFS Release Publish (preview-validated pipeline)

@@ -19,7 +19,9 @@ Pipeline shape:
 check console main against its latest Release
  -> if ahead: publish console -> wait for Release asset + latest API
  -> bump Cargo.toml and Cargo.lock to <target> -> merge
-  -> tag <preview-tag> at that commit -> CI green
+  -> cut release/<target> from a selected main commit containing the bump
+  -> CI green for that exact release-branch commit
+  -> tag <preview-tag> at that commit
  -> verify preview Release assets -> run binary locally + console checks
  -> validate with latest rc client
  -> report preview acceptance results -> STOP for explicit human confirmation
@@ -29,7 +31,7 @@ check console main against its latest Release
  -> update installation references on main -> update rustfs.com announcement
 ```

-On preview validation failure: fix lands on main via normal PR (Cargo versions are already at `<target>`, no new source bump PR), then tag `<preview-tag N+1>` at the new main commit and restart from Phase 2. Installation references remain on the previous published deliverable throughout preview validation.
+On candidate or preview validation failure: fix lands on main via normal PR, then backport only the needed fix to `release/<target>` through a reviewed PR. Re-run release-branch CI at the new commit and, if a preview was already tagged, tag `<preview-tag N+1>` there and restart acceptance. Installation references remain on the previous published deliverable throughout preview validation.

 ## Required inputs

@@ -68,14 +70,14 @@ Rules:
 - Preview Release assets are versioned and intentionally visible on the Releases page for the duration of validation. Do not label them Latest or use them to update any latest distribution channel.
 - Never delete a preview Release by hand before Phase 6 finishes — Phase 4 downloads its assets and the final Release notes are generated while it still exists. Cleanup is CI's job; only step in manually (`gh release delete "<preview-tag>" --yes`, never `--cleanup-tag`) if `cleanup-preview-releases` failed.
 - Tags have no `v` prefix. Always annotated: `git tag -a <tag> -m "Release <tag>"`.
- The final tag MUST point at exactly `PREVIEW_HASH` — the commit the validated preview tag points at. Never tag current `main` HEAD (commits merged after validation are unvalidated), and never create an extra version-bump commit between preview and final. Phase 7 updates installation references on main after publication; neither tag moves to that follow-up commit.
+- The release branch is the candidate source during multi-day acceptance. Do not merge main wholesale into it, force-push it, or advance it with unrelated changes. The final tag MUST point at exactly `PREVIEW_HASH` — the commit the validated preview tag points at. Never tag current `main` HEAD or an advanced release-branch HEAD, and never create an extra version-bump commit between preview and final. Phase 7 updates installation references on main after publication; neither tag moves to that follow-up commit.
 - When a previous deliverable exists, GitHub Release notes for the preview and final tags MUST use it as their shared comparison baseline: the most recently published non-preview Release before the target. Internal `-preview.N` Releases are explicitly excluded from that selection, even when they point at the same commit as the final tag — cleanup runs after the notes are generated, so the preview Release is still present and would otherwise be picked as the baseline. If no previous deliverable exists, omit `previous_tag_name` and record that GitHub's default baseline fallback was used.
 - Generated Release notes carry a workflow-management marker so retries can repair them. Before manually curating a generated body, remove that marker; unmarked non-placeholder notes are preserved by later workflow runs.
- Phases run in order; a failure blocks dependent steps. For a preview/source acceptance failure before the final tag exists, land the fix, verify Cargo still matches the confirmed target, and restart from Phase 2 with the next preview at the fixed commit. If main has advanced to another target, resolve that source/version mismatch before another preview.
+- Phases run in order; a failure blocks dependent steps. For a candidate or preview/source acceptance failure before the final tag exists, land the fix on main, backport it to the release branch, verify Cargo there still matches the confirmed target, and restart from release-branch CI in Phase 2. Use the next preview iteration if any preview was already tagged. Main may advance to another target without changing this release candidate.
 - Once the final tag exists, preserve its commit. Retry failed publication jobs for that exact tag; do not recreate the tag or restart preview on newer main. If a source change is required after final-tag publication, report the failure and obtain a new target version. Phase 7 failures resume only the failed follow-up after rechecking artifact availability and whether a newer release has superseded it.
 - Completing preview acceptance does not authorize the final tag. After Phases 3–5 pass, report the acceptance evidence and stop until the user explicitly confirms continuation. The original release request, an earlier confirmation, silence, or an automated follow-up does not satisfy this gate.
 - Confirmation is scoped to the reported `<target>`, `<preview-tag>`, and `PREVIEW_HASH`. A failed or repeated acceptance cycle, including any new preview iteration, invalidates prior confirmation and requires a new one.
- If the release is abandoned after Phase 1 merged, main's version files claim a version that was never tagged. Either revert the bump PR or leave it to be overwritten by the next release — but tell the user explicitly and record the decision.
+- If the release is abandoned after Phase 1 merged, main's version files claim a version that was never tagged. Either revert the bump PR or leave it to be overwritten by the next release — but tell the user explicitly and record the decision. Do not delete the release branch or preview tags as part of abandonment without an explicit decision.
 - User-facing status updates in Chinese; commits, PR titles/bodies, and tag messages in English. No hard-wrapping in commit messages, PR bodies, or documentation prose — one logical line per sentence/paragraph, let soft wrap handle display.

 ## Phase 0 — Preflight
@@ -83,6 +85,7 @@ Rules:
 - `git status --short` clean; `git fetch origin main --tags`.
 - `gh auth status` works; confirm you can view `gh release list -L 3`.
 - Confirm the exact final target version with the user if not explicit.
+- Check whether `release/<target>` already exists before creating it; a resumed release reuses the existing branch and its history. For an in-flight release that already has a preview tag but no branch, anchor the new branch to that tag's commit only after verifying its target versions and prior acceptance evidence.

 ### Console release gate

@@ -90,18 +93,19 @@ Read and complete [the Console gate](references/console-gate.md) before Phase 1.

 ## Phase 1 — Source version bump to the final target (once)

- If Cargo.toml and all workspace members in Cargo.lock already read `<target>` (e.g. this is a restart after a failed preview), verify both and skip the source bump. Installation versions intentionally differ during preview; do not treat them as leftovers. If a previous preparation advanced installation references to an unavailable target, restore those references to the verified published baseline before selecting the next preview commit.
- Otherwise invoke the `rustfs-release-version-bump` skill with stage `prepare`, the final `<target>` (NOT a preview version), and full GitHub flow (commit/push/PR).
- Get the PR merged into main. Record the resulting main commit:
+- If `release/<target>` exists, verify its Cargo.toml and workspace members in Cargo.lock read `<target>` and skip the source bump, even if main has advanced. If the branch versions differ, stop and resolve the mismatch without downgrading main.
+- If no branch exists but an unfinished preview tag does, verify its commit has `<target>` in both version files and skip the bump; Phase 2 anchors the branch to that commit. Stop if those files disagree. Otherwise, if main already has `<target>` in both files, skip the bump. Installation references intentionally retain the previous published deliverable during preview; if they were advanced to an unavailable target, restore that baseline before choosing the next candidate.
+- Only when neither the release branch, an unfinished preview tag, nor current main already has `<target>`: if main is at a later version, select and verify a historical commit containing `<target>` rather than downgrading main; block if none exists. Otherwise invoke the `rustfs-release-version-bump` skill with stage `prepare`, the final `<target>` (NOT a preview version), and full GitHub flow (commit/push/PR).
+- If a bump PR was needed, get it merged into main and record its merge commit. For a new release branch when main already has `<target>`, identify the main commit that introduced those versions. Phase 2 chooses a release candidate containing that commit; an existing release branch keeps its own history and a newer main HEAD is not required.

-```bash
-git fetch origin main
-PREVIEW_HASH=$(git rev-parse origin/main)   # must contain the bump PR
-```
+Do not wait for the moving latest main HEAD to become green. The release-branch CI in Phase 2 validates the chosen commit before a preview tag is created.

-`PREVIEW_HASH` is the single source of truth for the rest of the pipeline — report it to the user and reuse it verbatim in Phases 2 and 6. Both the preview tag and the final tag will point at it.
+## Phase 2 — Pin and validate the release candidate, then publish the preview tag

-## Phase 2 — Publish the preview tag
+- Use `release/<target>`. On the first attempt, create it from a selected main commit containing the version bump and intended changes; it need not be the latest main HEAD. On a restart, retain the existing branch and backport only reviewed fixes from main. Verify the branch's Cargo versions still equal `<target>`. If adopting an existing preview, run this branch CI before continuing its acceptance; reuse the tag only when it points to the validated SHA.
+- The current `ci.yml` runs automatically on main pushes, not release-branch pushes. Dispatch it explicitly with `gh workflow run ci.yml --ref "release/<target>"`. Locate that dispatch run, verify its `head_sha` equals the release branch's remote HEAD, and require the full expected CI job set to pass, including `Quick Checks`, `Test and Lint`, and `End-to-End Tests (full merge gate)`. Treat a missing or skipped required lane as incomplete. A successful PR check or CI run for a different SHA does not qualify. If the branch advances or a run is cancelled, validate the new SHA again. Do not tag a candidate with failing or incomplete CI.
+- Prefer protecting `release/*` against force pushes and deletion and requiring reviewed backport PRs. Check live rulesets rather than assuming the main ruleset covers release branches; report a missing release-branch rule as a gap.
+- Set `PREVIEW_HASH` to the CI-validated release-branch commit and recheck the remote branch HEAD immediately before tagging. Report the branch, commit, and CI run URL; do not derive `PREVIEW_HASH` from a later `origin/main` fetch. When adopting an existing preview tag at this SHA, skip tag creation and continue Phase 3.

 ```bash
 git tag -a "<preview-tag>" -m "Release <preview-tag>" "$PREVIEW_HASH"
@@ -112,11 +116,12 @@ Pushing the tag triggers `.github/workflows/build.yml` ("Build and Release"); `d

 The preview run builds versioned artifacts and publishes them in a GitHub prerelease. Docker may publish exact preview image tags. Latest-channel, R2, and Helm publication must be skipped; installation references and the website announcement stay unchanged.

-On a restart (N+1), refresh `PREVIEW_HASH=$(git rev-parse origin/main)` first — it must contain the fix — and re-report it.
+On a restart after a backport, re-run release-branch CI before setting the new `PREVIEW_HASH`. Use the next unused preview iteration when the validated commit differs from an existing preview tag; never move an existing tag.

 ## Phase 3 — CI and preview Release verification

 - Find and watch the tag build: `gh run list --workflow build.yml --branch "<preview-tag>" --limit 1` then `gh run watch <run-id>`. Every build matrix target must succeed (linux x86_64/aarch64 × musl/gnu, macos-aarch64, windows-x86_64).
+- Confirm the preview tag resolves to the Phase 2 CI-validated `PREVIEW_HASH`. If the release branch advances during acceptance, repeat Phases 2–5 with a new preview tag.
 - Confirm the Release publication jobs (`create-release`, `upload-release-assets`, and `publish-release`) succeed while `update-latest-version` is skipped.
 - Verify `gh release view "<preview-tag>" --json isPrerelease,assets,url`: `isPrerelease` must be `true`, and the Release must contain all 6 versioned platform zips, checksums, SBOM, and provenance with no `-latest` assets. Confirm `gh api repos/{owner}/{repo}/releases/latest --jq .tag_name` does not return `<preview-tag>`.
 - Record `PREVIOUS_DELIVERABLE`, selected from published Releases by `publishedAt` after excluding the current tag and every `-preview.N` tag. Verify `gh release view "<preview-tag>" --json body --jq .body` contains `## What's Changed` and, when `PREVIOUS_DELIVERABLE` exists, `**Full Changelog**: https://github.com/rustfs/rustfs/compare/<PREVIOUS_DELIVERABLE>...<preview-tag>`. For a repository with no previous deliverable, verify a Full Changelog link exists and record the GitHub baseline fallback.
@@ -128,13 +133,13 @@ Read and complete [preview acceptance](references/preview-acceptance.md): verify

 ### Manual confirmation gate

-After every Phase 3–5 check passes, report the target, preview tag, `PREVIEW_HASH`, preview Release URL, console result, and rc matrix, then explicitly ask the user whether to publish the final tag. End the turn without creating or pushing `<target>`.
+After every Phase 3–5 check passes, verify the release branch still points to `PREVIEW_HASH`. Report the target, release branch, preview tag, `PREVIEW_HASH`, release-branch CI run, preview Release URL, console result, and rc matrix, then explicitly ask the user whether to publish the final tag. End the turn without creating or pushing `<target>`.

 Continue to Phase 6 only after a new user reply explicitly confirms the reported target, preview tag, and commit. A clear affirmative reply to that exact report, such as `确认继续`, is sufficient; if the reply is ambiguous or any reported value changed, ask again.

 ## Phase 6 — Publish the final tag on the validated commit

-No second source version bump. The final tag goes on the exact commit the preview validated:
+No second source version bump. Before creating the final tag, recheck the remote `release/<target>` HEAD equals `PREVIEW_HASH`. If it moved after confirmation, the approval no longer applies; validate and obtain confirmation for a new preview. The final tag goes on the exact commit the preview validated:

 ```bash
 git fetch origin --tags
@@ -158,7 +163,7 @@ Only after Phase 6 succeeds, complete [post-release updates](references/post-rel
 Always report:

 - Console gate result: previous/latest Console tags, whether merged changes required a release, `CONSOLE_HASH`, and Console run/Release URLs when a release was published.
- Target version, preview tag(s) used, `PREVIEW_HASH` (which both tags point at).
+- Target version, release branch, release-branch CI run and validated SHA, preview tag(s) used, `PREVIEW_HASH` (which both tags point at).
 - Manual confirmation gate status (`WAITING_FOR_CONFIRMATION` or `CONFIRMED`) and its exact target, preview tag, and `PREVIEW_HASH`.
 - Per-phase result (PASS/FAIL/BLOCKED) with key evidence: preview and final Release URLs, preview `isPrerelease`/`isLatest` state, final latest-channel state, console check results, the rc command matrix, and the preview-Release cleanup result (deleted Releases plus surviving tags).
 - Post-release installation PR and verification results; website banner target, text, link, PR, and observed deployment state. Report any remaining merge authorization or failed deployment explicitly rather than claiming the banner is live.
@@ -1,2 +1,2 @@
 sha256-darwin=3b6ff112ff16acbaa4148c4736095a1b6b71cc11af0dd3513ff28c6bdc334ca8
-sha256-linux=4e8bfc01870565a0a608a1634e6919fc593ce120a970aa04d974f7edf2675c21
+sha256-linux=80d0fae627007c31f250fb1fb21c8a952861195d53d7f60b6e71636d5ca65b46
@@ -542,10 +542,11 @@ fail-fast = false
 [profile.e2e-smoke.junit]
 path = "junit.xml"

-# These storage-heavy fixtures reserve capacity. Pagination tests bound listing
-# requests separately, after their seed objects have been written.
+# Bound the whole pagination fixture, including startup, seeding, and cleanup;
+# its LIST deadline begins only after the seed objects have been written.
 [[profile.e2e-smoke.overrides]]
 filter = 'package(e2e_test) & test(/^list_objects_v2_pagination_test::tests::(test_list_objects_v2_delimiter_small_page_traverses_all|test_list_objects_v2_max_keys_above_limit_returns_token|test_list_objects_v2_maxkeys_above_limit_with_delimiter)$/)'
+slow-timeout = { period = "60s", terminate-after = 2, grace-period = "10s" }
 threads-required = "num-test-threads"

 [[profile.e2e-smoke.overrides]]
@@ -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
@@ -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);
@@ -280,18 +280,19 @@ async fn write_delete_rollback_file(
 ) -> Result<()> {
    let backup_dir = object_dir.join(rollback_dir.to_string());
    let path = backup_dir.join(name);
-    if namespace_owner.is_none() {
-        fs::create_dir_all(&backup_dir).await.map_err(to_file_error)?;
-        fs::write(path, data).await.map_err(to_file_error)?;
-        return Ok(());
-    }
    let lease = os::acquire_namespace_mutation_lease_with_owner(&path, namespace_owner).await;
    let data = data.to_vec();
+    let object_dir = object_dir.to_path_buf();
    os::run_blocking_namespace_operation(lease, move || {
        std::fs::create_dir_all(&backup_dir)?;
        #[cfg(test)]
        run_owned_file_write_before_open(&path);
-        std::fs::write(path, data)
+        // Rollback must never consume a partially written backup. Stage outside
+        // the rollback directory so interrupted writes cannot become staged data.
+        let mut staged = tempfile::Builder::new().prefix(".delete-rollback-").tempfile_in(object_dir)?;
+        std::io::Write::write_all(&mut staged, &data)?;
+        staged.persist(path).map_err(|err| err.error)?;
+        Ok(())
    })
    .await
    .map_err(to_file_error)?;
@@ -6798,7 +6799,6 @@ impl LocalDisk {
    }

    #[tracing::instrument(name = "delete_file", level = "trace", skip_all)]
-    #[async_recursion::async_recursion]
    async fn delete_file_with_namespace_owner(
        &self,
        base_path: &PathBuf,
@@ -7952,7 +7952,6 @@ impl LocalDisk {
        Err(DiskError::FileCorrupt)
    }

-    #[async_recursion::async_recursion]
    #[allow(clippy::too_many_arguments)]
    async fn scan_dir<W>(
        &self,
@@ -16843,6 +16842,99 @@ mod test {
        );
    }

+    #[cfg(unix)]
+    #[tokio::test]
+    async fn delete_rollback_partial_backup_write_preserves_version_history() {
+        const CHILD: &str = "RUSTFS_TEST_DELETE_BACKUP_PARTIAL_WRITE_CHILD";
+        let bucket = "partial-backup-bucket";
+        let object = "versioned-object";
+        if let Some(root) = std::env::var_os(CHILD) {
+            let root = PathBuf::from(root);
+            let endpoint = Endpoint::try_from(root.to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+            let disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
+            let xl_path = root.join(bucket).join(object).join(STORAGE_FORMAT_FILE);
+            let original = fs::read(&xl_path).await.expect("committed history");
+            let version_id = Uuid::parse_str(&std::env::var("RUSTFS_TEST_DELETE_BACKUP_VERSION").expect("version ID"))
+                .expect("valid version ID");
+            let first = test_file_info(object, version_id, None, None);
+            let rollback_dir = Uuid::new_v4();
+            let result = disk
+                .delete_version(
+                    bucket,
+                    object,
+                    first.clone(),
+                    false,
+                    DeleteOptions {
+                        old_data_dir: Some(rollback_dir),
+                        ..Default::default()
+                    },
+                )
+                .await;
+            assert!(result.is_err(), "the backup must fail under the file-size limit");
+            disk.undo_write_with_namespace_owner(
+                bucket,
+                object,
+                first,
+                DeleteOptions {
+                    undo_write: true,
+                    undo_delete: true,
+                    old_data_dir: Some(rollback_dir),
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            .expect("failed delete rollback");
+            assert_eq!(fs::read(&xl_path).await.expect("metadata after rollback"), original);
+            return;
+        }
+
+        let dir = tempfile::tempdir().expect("temporary disk");
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("disk endpoint");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk");
+        ensure_test_volume(&disk, bucket).await;
+        ensure_test_volume(&disk, RUSTFS_META_TMP_BUCKET).await;
+        let object_dir = dir.path().join(bucket).join(object);
+        fs::create_dir_all(&object_dir).await.expect("object directory");
+        let xl_path = object_dir.join(STORAGE_FORMAT_FILE);
+        let version_id = Uuid::new_v4();
+        let mut meta = FileMeta::new();
+        meta.add_version(test_file_info(object, version_id, None, None))
+            .expect("first version");
+        for _ in 1..100 {
+            meta.add_version(test_file_info(object, Uuid::new_v4(), None, None))
+                .expect("history version");
+        }
+        let original = meta.marshal_msg().expect("version history");
+        assert!(original.len() > 4096, "backup must exceed the injected write limit");
+        fs::write(&xl_path, &original).await.expect("committed metadata");
+
+        // Isolate the process-wide limit and ignored signal in a child. A real
+        // EFBIG short write exercises the failed-backup path without privileges.
+        let output = std::process::Command::new("sh")
+            .args(["-c", "trap '' XFSZ; ulimit -f 4; exec \"$@\"", "partial-backup-test"])
+            .arg(std::env::current_exe().expect("test executable"))
+            .args([
+                "--exact",
+                "disk::local::test::delete_rollback_partial_backup_write_preserves_version_history",
+                "--nocapture",
+            ])
+            .env(CHILD, dir.path())
+            .env("RUSTFS_TEST_DELETE_BACKUP_VERSION", version_id.to_string())
+            .output()
+            .expect("isolated partial-write test should run");
+        let stdout = String::from_utf8_lossy(&output.stdout);
+        let stderr = String::from_utf8_lossy(&output.stderr);
+        assert!(output.status.success(), "partial-write child failed: {stdout} {stderr}");
+        assert!(stdout.contains("1 passed"), "child must execute the selected regression: {stdout}");
+        assert_eq!(fs::read(&xl_path).await.expect("metadata after rollback"), original);
+        disk.write_metadata("", bucket, object, test_file_info(object, Uuid::new_v4(), None, None))
+            .await
+            .expect("write after failed delete");
+        let after = fs::read(&xl_path).await.expect("metadata after subsequent write");
+        assert_eq!(FileMeta::load(&after).expect("readable history").versions.len(), 101);
+    }
+
    // The undo_write restore consumes `<rollback>/xl.meta.bkp` by rename; a
    // synthetic rollback dir is then empty and must be reclaimed so the object
    // dir can empty out (BucketNotEmpty leak). A real data dir still holds its
@@ -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 {
@@ -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))
                })
@@ -985,6 +985,19 @@ impl PreparedGetObjectReader {
        reader.body_source = crate::object_api::GetObjectBodySource::HookMissed;
        Ok(ECStore::attach_read_lock_guard(reader, self.read_lock_guard))
    }
+
+    /// Open the prepared source with bounded copy prefetch and request-owned
+    /// cancellation, including cancellation during reader construction.
+    pub async fn into_reader_for_copy(self) -> Result<(GetObjectReader, tokio_util::sync::DropGuard)> {
+        let cancellation = tokio_util::sync::CancellationToken::new();
+        let guard = cancellation.clone().drop_guard();
+        let reader = crate::set_disk::with_get_object_read_cancellation(
+            cancellation,
+            crate::set_disk::with_get_object_read_policy(crate::set_disk::GetObjectReadPolicy::CopySource, self.into_reader()),
+        )
+        .await?;
+        Ok((reader, guard))
+    }
 }

 struct LockGuardedReader {
@@ -1742,22 +1742,24 @@ impl AsyncWrite for HttpWriter {
            match this.sender.poll_reserve(cx) {
                Poll::Ready(Ok(())) => {
                    this.sender
-                        .send_item(Some(Bytes::copy_from_slice(buf)))
+                        .send_item(Some(Bytes::copy_from_slice(&buf[..HTTP_WRITER_BUFFER_SIZE])))
                        .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?;
                    this.start_request();
-                    return Poll::Ready(Ok(buf.len()));
+                    return Poll::Ready(Ok(HTTP_WRITER_BUFFER_SIZE));
                }
                Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))),
                Poll::Pending => return Poll::Pending,
            }
        }

-        this.pending_chunk.extend_from_slice(buf);
-        if !buf.is_empty() {
+        // A slot bounds memory only when every queued and pending chunk is bounded.
+        let written = buf.len().min(HTTP_WRITER_BUFFER_SIZE - this.pending_chunk.len());
+        this.pending_chunk.extend_from_slice(&buf[..written]);
+        if written > 0 {
            this.start_request();
        }

-        Poll::Ready(Ok(buf.len()))
+        Poll::Ready(Ok(written))
    }

    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), io::Error>> {
@@ -1770,6 +1772,9 @@ impl AsyncWrite for HttpWriter {
    }

    fn poll_write_vectored(mut self: Pin<&mut Self>, cx: &mut Context<'_>, bufs: &[IoSlice<'_>]) -> Poll<io::Result<usize>> {
+        if bufs.len() == 1 {
+            return self.poll_write(cx, &bufs[0]);
+        }
        let this = self.as_mut().get_mut();
        if let Err(err) = this.take_background_error() {
            return Poll::Ready(Err(err));
@@ -1783,33 +1788,21 @@ impl AsyncWrite for HttpWriter {
            }
        }

-        let total_len = bufs.iter().map(|buf| buf.len()).sum::<usize>();
-        if total_len == 0 {
-            return Poll::Ready(Ok(0));
-        }
-
-        if bufs.len() == 1 && this.pending_chunk.is_empty() && total_len >= HTTP_WRITER_BUFFER_SIZE {
-            match this.sender.poll_reserve(cx) {
-                Poll::Ready(Ok(())) => {
-                    this.sender
-                        .send_item(Some(Bytes::copy_from_slice(bufs[0].as_ref())))
-                        .map_err(|e| send_error_to_io(e, "HttpWriter send error"))?;
-                    this.start_request();
-                    return Poll::Ready(Ok(total_len));
-                }
-                Poll::Ready(Err(err)) => return Poll::Ready(Err(poll_send_error_to_io(err, "HttpWriter send error"))),
-                Poll::Pending => return Poll::Pending,
+        let available = HTTP_WRITER_BUFFER_SIZE - this.pending_chunk.len();
+        let mut written = 0;
+        for buf in bufs {
+            let len = buf.len().min(available - written);
+            this.pending_chunk.extend_from_slice(&buf[..len]);
+            written += len;
+            if written == available {
+                break;
            }
        }
-
-        for buf in bufs {
-            this.pending_chunk.extend_from_slice(buf);
-        }
-        if total_len > 0 {
+        if written > 0 {
            this.start_request();
        }

-        Poll::Ready(Ok(total_len))
+        Poll::Ready(Ok(written))
    }

    fn is_write_vectored(&self) -> bool {
@@ -2916,6 +2909,131 @@ mod tests {
        handle.abort();
    }

+    fn paused_http_writer() -> (HttpWriter, tokio::sync::mpsc::Receiver<Option<Bytes>>) {
+        let (sender, receiver) = tokio::sync::mpsc::channel(HTTP_WRITER_CHANNEL_CAPACITY);
+        let (_, err_rx) = tokio::sync::oneshot::channel();
+        let writer = HttpWriter {
+            url: "http://127.0.0.1/paused".to_string(),
+            method: Method::PUT,
+            headers: HeaderMap::new(),
+            err_rx,
+            start_tx: None,
+            sender: PollSender::new(sender),
+            handle: tokio::spawn(std::future::pending()),
+            pending_chunk: BytesMut::with_capacity(HTTP_WRITER_BUFFER_SIZE),
+            finish: false,
+            track_internode_metrics: false,
+            internode_operation: None,
+        };
+        (writer, receiver)
+    }
+
+    #[tokio::test]
+    async fn http_writer_bounds_large_writes_under_backpressure() {
+        let payload = vec![0xa5; 3 * HTTP_WRITER_BUFFER_SIZE + 17];
+        for vectored in [false, true] {
+            let (mut writer, mut receiver) = paused_http_writer();
+            for _ in 0..HTTP_WRITER_CHANNEL_CAPACITY {
+                let written = std::future::poll_fn(|cx| {
+                    if vectored {
+                        Pin::new(&mut writer).poll_write_vectored(cx, &[IoSlice::new(&payload)])
+                    } else {
+                        Pin::new(&mut writer).poll_write(cx, &payload)
+                    }
+                })
+                .await
+                .expect("a free queue slot should accept one bounded chunk");
+                assert_eq!(written, HTTP_WRITER_BUFFER_SIZE, "large writes must be partial");
+            }
+            {
+                let write = std::future::poll_fn(|cx| {
+                    if vectored {
+                        Pin::new(&mut writer).poll_write_vectored(cx, &[IoSlice::new(&payload)])
+                    } else {
+                        Pin::new(&mut writer).poll_write(cx, &payload)
+                    }
+                });
+                tokio::pin!(write);
+                assert!(futures::poll!(&mut write).is_pending(), "a stalled receiver must exert backpressure");
+                let chunk = receiver.recv().await.expect("queued body chunk").expect("body, not EOF");
+                assert_eq!(chunk.as_ref(), &payload[..HTTP_WRITER_BUFFER_SIZE]);
+                let written = tokio::time::timeout(Duration::from_secs(1), write)
+                    .await
+                    .expect("draining a queue slot should wake the blocked writer")
+                    .expect("resumed write should succeed");
+                assert_eq!(written, HTTP_WRITER_BUFFER_SIZE);
+            }
+            assert!(writer.pending_chunk.len() <= HTTP_WRITER_BUFFER_SIZE);
+            for _ in 0..HTTP_WRITER_CHANNEL_CAPACITY {
+                let chunk = receiver.try_recv().expect("queued body chunk").expect("body, not EOF");
+                assert_eq!(chunk.as_ref(), &payload[..HTTP_WRITER_BUFFER_SIZE]);
+            }
+            assert!(receiver.try_recv().is_err());
+        }
+    }
+
+    #[tokio::test]
+    async fn http_writer_bounds_mixed_writes_and_preserves_bytes() {
+        let header = [0x11; 32];
+        let payload = vec![0xa5; 3 * HTTP_WRITER_BUFFER_SIZE + 17];
+        for vectored in [false, true] {
+            let (mut writer, mut receiver) = paused_http_writer();
+            writer.write_all(&header).await.expect("small prefix should be buffered");
+            let mut remaining = payload.as_slice();
+            while !remaining.is_empty() {
+                let written = if vectored {
+                    writer
+                        .write_vectored(&[IoSlice::new(&[]), IoSlice::new(remaining), IoSlice::new(&[])])
+                        .await
+                } else {
+                    writer.write(remaining).await
+                }
+                .expect("payload write should succeed");
+                assert!(written > 0 && written <= HTTP_WRITER_BUFFER_SIZE);
+                assert!(writer.pending_chunk.len() <= HTTP_WRITER_BUFFER_SIZE, "pending bytes must stay bounded");
+                remaining = &remaining[written..];
+            }
+            writer.flush().await.expect("remaining prefix should flush");
+            let mut actual = Vec::new();
+            while let Ok(chunk) = receiver.try_recv() {
+                let chunk = chunk.expect("body, not EOF");
+                assert!(chunk.len() <= HTTP_WRITER_BUFFER_SIZE, "queued bytes must stay bounded");
+                actual.extend_from_slice(&chunk);
+            }
+            assert_eq!(&actual[..header.len()], &header);
+            assert_eq!(&actual[header.len()..], payload.as_slice());
+        }
+    }
+
+    #[tokio::test]
+    async fn http_writer_large_vectored_body_preserves_hash_and_payload() {
+        let state = TestState::default();
+        let (url, handle) = start_test_server(state.clone()).await.expect("test server should bind");
+        let header = [0x11; 32];
+        let payload: Vec<_> = (0..3 * HTTP_WRITER_BUFFER_SIZE + 17).map(|i| (i % 251) as u8).collect();
+        let mut writer = HttpWriter::new(url, Method::PUT, HeaderMap::new())
+            .await
+            .expect("HTTP writer should open");
+        let mut slices = [
+            IoSlice::new(&[]),
+            IoSlice::new(&header),
+            IoSlice::new(&payload),
+            IoSlice::new(&[]),
+        ];
+        let mut remaining = slices.as_mut_slice();
+        while remaining.iter().any(|slice| !slice.is_empty()) {
+            let written = writer.write_vectored(remaining).await.expect("body write should succeed");
+            assert!(written > 0 && written <= HTTP_WRITER_BUFFER_SIZE);
+            IoSlice::advance_slices(&mut remaining, written);
+        }
+        writer.shutdown().await.expect("request should finish");
+        let bodies = state.put_bodies.lock().await;
+        assert_eq!(bodies.len(), 1);
+        assert_eq!(&bodies[0][..header.len()], &header);
+        assert_eq!(&bodies[0][header.len()..], payload.as_slice());
+        handle.abort();
+    }
+
    #[tokio::test]
    async fn http_writer_does_not_send_empty_preflight_put() {
        let state = TestState::default();
@@ -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,
@@ -840,6 +840,7 @@ async fn scan_and_persist_local_bucket(
                }
                match persist_scanner_checkpoint(
                    set.clone(),
+                    store.clone(),
                    ScannerCheckpointPersistContext {
                        ctx: &scan_ctx,
                        expected_publication_epoch,
@@ -69,10 +69,13 @@ async fn wait_for_checkpoint_foreground_quiet(ctx: &CancellationToken) -> bool {
 /// Persist one bounded checkpoint and refresh its CAS revisions.
 ///
 /// Local and remote workers share the same publication/leader fencing and
-/// revision-refresh contract; only their lock/cancellation handling remains
+/// revision-refresh contract. Cycle/leader state belongs to the global store,
+/// while cache revisions and publication admission belong to the set store.
+/// Only their lock/cancellation handling remains
 /// at the caller because those guards have different concrete types.
-pub(crate) async fn persist_scanner_checkpoint<S>(
+pub(crate) async fn persist_scanner_checkpoint<S, F>(
    store: Arc<S>,
+    fence_store: Arc<F>,
    context: ScannerCheckpointPersistContext<'_>,
    cache_name: &str,
    checkpoint: &DataUsageCache,
@@ -80,6 +83,7 @@ pub(crate) async fn persist_scanner_checkpoint<S>(
 ) -> ScannerCheckpointPersistResult
 where
    S: ScannerObjectIO + ScannerConfigObjectDelete,
+    F: ScannerObjectIO,
 {
    let foreground_quiet = wait_for_checkpoint_foreground_quiet(context.ctx).await;
    if !foreground_quiet && context.ctx.is_cancelled() {
@@ -97,9 +101,13 @@ where
        );
    }

-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
-        .await
-        .is_err()
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(
+        context.cycle,
+        context.leader_epoch,
+        fence_store.clone(),
+    )
+    .await
+    .is_err()
    {
        return ScannerCheckpointPersistResult::FenceChanged;
    }
@@ -117,7 +125,7 @@ where
        return ScannerCheckpointPersistResult::Failed(error);
    }

-    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, store.clone())
+    if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(context.cycle, context.leader_epoch, fence_store)
        .await
        .is_err()
        || scanner_publication_admission_for_epoch(store.clone(), context.expected_publication_epoch)
@@ -1189,8 +1189,14 @@ impl ScannerIOCache for SetDisks {
                                    await_scanner_disk_shutdown(scan.as_mut()).await;
                                    break Err(Error::other("scanner leader fence changed before checkpoint save"));
                                }
+                                let Some(fence_store) = crate::resolve_scanner_object_store_handle() else {
+                                    scan_ctx.cancel();
+                                    await_scanner_disk_shutdown(scan.as_mut()).await;
+                                    break Err(Error::other("scanner global cycle fence store is unavailable before checkpoint save"));
+                                };
                                match persist_scanner_checkpoint(
                                    store_clone_clone.clone(),
+                                    fence_store,
                                    ScannerCheckpointPersistContext {
                                        ctx: &ctx_clone,
                                        expected_publication_epoch: expected_publication_epoch_clone,
@@ -27,9 +27,10 @@ use crate::storage_api::owner::{
 };
 use crate::storage_api::scan::{BucketOperations as _, DeleteBucketOptions, MakeBucketOptions, ObjectIO as _};
 use crate::{
-    DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints, ScannerObjectOptions,
-    ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests, init_ecstore_config_for_scanner_tests,
-    init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
+    DATA_USAGE_BLOOM_NAME_PATH, DiskOption, ECStore, Endpoint, EndpointServerPools, Endpoints, InstanceContext, PoolEndpoints,
+    ScannerObjectOptions, ScannerPutObjReader, UNKNOWN_TIER, init_bucket_metadata_sys_for_scanner_tests,
+    init_ecstore_config_for_scanner_tests, init_local_disks_with_instance_ctx, new_disk, path2_bucket_object_with_base_path,
+    save_config,
 };
 use rustfs_concurrency::{
    AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadAdmissionSnapshotProvider,
@@ -566,6 +567,303 @@ async fn setup_two_pool_scanner_store() -> (tempfile::TempDir, Arc<ECStore>) {
    (temp_dir, store)
 }

+#[tokio::test]
+#[serial]
+async fn checkpoint_uses_global_cycle_fence_with_set_scoped_cache_revisions() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    let cycle = 41341;
+    let leader_epoch = 17;
+    save_config(
+        store.clone(),
+        DATA_USAGE_BLOOM_NAME_PATH.as_str(),
+        crate::scanner::encode_scanner_cycle_fence_for_test(cycle, leader_epoch),
+    )
+    .await
+    .expect("persist cycle fence through global scanner store");
+    crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, store.clone())
+        .await
+        .expect("global scanner store must read the fence");
+    let ctx = CancellationToken::new();
+    let cache_name = "checkpoint-bucket/.usage-cache.bin";
+    let mut checkpoint = DataUsageCache::default();
+    checkpoint.info.name = "checkpoint-bucket".to_string();
+    checkpoint.info.next_cycle = cycle;
+    checkpoint.info.leader_epoch = leader_epoch;
+    let mut mismatched_sets = 0;
+    for (pool_index, pool) in store.pools.iter().enumerate() {
+        for (set_index, set) in pool.disk_set.iter().enumerate() {
+            let publication_epoch = scanner_publication_epoch(set.clone())
+                .await
+                .expect("set publication admission");
+            DataUsageCache::default()
+                .save(set.clone(), cache_name)
+                .await
+                .expect("seed set-scoped cache");
+            let mut revisions = DataUsageCache::read_revisions(set.clone(), cache_name)
+                .await
+                .expect("initial set-scoped revisions");
+            let initial_revisions = revisions.clone();
+            let make_context = || ScannerCheckpointPersistContext {
+                ctx: &ctx,
+                expected_publication_epoch: publication_epoch,
+                cycle,
+                leader_epoch,
+            };
+            if crate::remote_scanner::validate_remote_scanner_request_fence_with_store(cycle, leader_epoch, set.clone())
+                .await
+                .is_err()
+            {
+                mismatched_sets += 1;
+                assert!(
+                    matches!(
+                        persist_scanner_checkpoint(
+                            set.clone(),
+                            set.clone(),
+                            make_context(),
+                            cache_name,
+                            &checkpoint,
+                            &mut revisions,
+                        )
+                        .await,
+                        ScannerCheckpointPersistResult::FenceChanged
+                    ),
+                    "set-scoped cycle lookup must reproduce the original failure for pool={pool_index}, set={set_index}"
+                );
+            }
+            assert_eq!(revisions, initial_revisions, "rejected checkpoint must not update revisions");
+            for skip_healing in [false, true] {
+                checkpoint.info.skip_healing = skip_healing;
+                let before_save = revisions.clone();
+                assert!(
+                    matches!(
+                        persist_scanner_checkpoint(
+                            set.clone(),
+                            store.clone(),
+                            make_context(),
+                            cache_name,
+                            &checkpoint,
+                            &mut revisions,
+                        )
+                        .await,
+                        ScannerCheckpointPersistResult::Saved
+                    ),
+                    "checkpoint helper must use the global cycle fence"
+                );
+                assert_ne!(revisions, before_save, "new checkpoint content must advance cache revisions");
+            }
+            let saved_revisions = revisions.clone();
+            for rejected_context in [
+                ScannerCheckpointPersistContext {
+                    cycle: cycle - 1,
+                    ..make_context()
+                },
+                ScannerCheckpointPersistContext {
+                    leader_epoch: leader_epoch - 1,
+                    ..make_context()
+                },
+                ScannerCheckpointPersistContext {
+                    expected_publication_epoch: publication_epoch + 1,
+                    ..make_context()
+                },
+            ] {
+                assert!(
+                    matches!(
+                        persist_scanner_checkpoint(
+                            set.clone(),
+                            store.clone(),
+                            rejected_context,
+                            cache_name,
+                            &checkpoint,
+                            &mut revisions,
+                        )
+                        .await,
+                        ScannerCheckpointPersistResult::FenceChanged
+                    ),
+                    "stale cycle, leader or publication epoch must reject checkpoint persistence"
+                );
+                assert_eq!(revisions, saved_revisions, "rejected writer must not refresh revisions");
+                assert_eq!(
+                    DataUsageCache::read_revisions(set.clone(), cache_name)
+                        .await
+                        .expect("read rejected checkpoint"),
+                    saved_revisions,
+                    "rejected writer must not mutate persisted caches"
+                );
+            }
+
+            let mut newer = checkpoint.clone();
+            newer.info.next_cycle = cycle + 1;
+            newer
+                .save(set.clone(), cache_name)
+                .await
+                .expect("simulate a newer cache writer");
+            let newer_revisions = DataUsageCache::read_revisions(set.clone(), cache_name)
+                .await
+                .expect("capture newer cache revisions");
+            assert_ne!(newer_revisions, saved_revisions);
+            assert!(
+                matches!(
+                    persist_scanner_checkpoint(
+                        set.clone(),
+                        store.clone(),
+                        make_context(),
+                        cache_name,
+                        &checkpoint,
+                        &mut revisions,
+                    )
+                    .await,
+                    ScannerCheckpointPersistResult::Failed(StorageError::PreconditionFailed)
+                ),
+                "matching global fence must not authorize stale cache CAS revisions"
+            );
+            assert_eq!(revisions, saved_revisions, "CAS failure must not refresh the stale writer");
+            assert_eq!(
+                DataUsageCache::read_revisions(set.clone(), cache_name)
+                    .await
+                    .expect("read cache after CAS failure"),
+                newer_revisions,
+                "CAS failure must preserve the newer persisted caches"
+            );
+        }
+    }
+    assert!(mismatched_sets > 0, "multi-pool fixture must expose a set-scoped cycle fence mismatch");
+}
+
+#[derive(Debug)]
+struct CheckpointFenceStore {
+    before: Vec<u8>,
+    after: Vec<u8>,
+    fail_on_read: Option<usize>,
+    reads: std::sync::atomic::AtomicUsize,
+}
+
+#[async_trait::async_trait]
+impl crate::storage_api::owner::ObjectIO for CheckpointFenceStore {
+    type Error = crate::EcstoreError;
+    type RangeSpec = crate::storage_api::owner::HTTPRangeSpec;
+    type HeaderMap = http::HeaderMap;
+    type ObjectOptions = crate::ScannerObjectOptions;
+    type ObjectInfo = crate::ScannerObjectInfo;
+    type GetObjectReader = crate::ScannerGetObjectReader;
+    type PutObjectReader = crate::ScannerPutObjReader;
+
+    async fn get_object_reader(
+        &self,
+        bucket: &str,
+        object: &str,
+        _range: Option<Self::RangeSpec>,
+        _headers: Self::HeaderMap,
+        _opts: &Self::ObjectOptions,
+    ) -> crate::EcstoreResult<Self::GetObjectReader> {
+        assert_eq!(bucket, RUSTFS_META_BUCKET);
+        assert_eq!(object, DATA_USAGE_BLOOM_NAME_PATH.as_str(), "global store must only read the cycle fence");
+        let read = self.reads.fetch_add(1, Ordering::SeqCst);
+        if self.fail_on_read == Some(read) {
+            return Err(crate::EcstoreError::other("injected global fence read failure"));
+        }
+        let state = if read == 0 { &self.before } else { &self.after };
+        Ok(crate::ScannerGetObjectReader {
+            stream: Box::new(std::io::Cursor::new(state.clone())),
+            object_info: crate::ScannerObjectInfo::default(),
+            buffered_body: None,
+            body_source: Default::default(),
+        })
+    }
+
+    async fn put_object(
+        &self,
+        _bucket: &str,
+        _object: &str,
+        _data: &mut Self::PutObjectReader,
+        _opts: &Self::ObjectOptions,
+    ) -> crate::EcstoreResult<Self::ObjectInfo> {
+        panic!("checkpoint persistence must never write through the global fence store");
+    }
+}
+
+#[tokio::test]
+#[serial]
+async fn checkpoint_rejects_unreadable_or_changed_global_fence() {
+    let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+    let set = &store.pools[1].disk_set[0];
+    let publication_epoch = scanner_publication_epoch(set.clone()).await.expect("idle set admission");
+    let cycle = 41341;
+    let leader_epoch = 17;
+    let valid = crate::scanner::encode_scanner_cycle_fence_for_test(cycle, leader_epoch);
+    let ctx = CancellationToken::new();
+    let mut checkpoint = DataUsageCache::default();
+    checkpoint.info.name = "checkpoint-bucket".to_string();
+    checkpoint.info.next_cycle = cycle;
+    checkpoint.info.leader_epoch = leader_epoch;
+
+    for (case, before, after, fail_on_read, expected_reads) in [
+        (
+            "cycle_changed",
+            valid.clone(),
+            crate::scanner::encode_scanner_cycle_fence_for_test(cycle + 1, leader_epoch),
+            None,
+            2,
+        ),
+        (
+            "leader_changed",
+            valid.clone(),
+            crate::scanner::encode_scanner_cycle_fence_for_test(cycle, leader_epoch + 1),
+            None,
+            2,
+        ),
+        ("post_save_corrupt", valid.clone(), vec![0; 7], None, 2),
+        ("post_save_read_failure", valid.clone(), valid.clone(), Some(1), 2),
+        ("pre_save_corrupt", vec![0; 7], valid.clone(), None, 1),
+        ("pre_save_read_failure", valid.clone(), valid.clone(), Some(0), 1),
+    ] {
+        let cache_name = format!("checkpoint-bucket/{case}/.usage-cache.bin");
+        DataUsageCache::default()
+            .save(set.clone(), &cache_name)
+            .await
+            .expect("seed checkpoint cache");
+        let mut revisions = DataUsageCache::read_revisions(set.clone(), &cache_name)
+            .await
+            .expect("read seeded revisions");
+        let initial = revisions.clone();
+        let fence_store = Arc::new(CheckpointFenceStore {
+            before,
+            after,
+            fail_on_read,
+            reads: std::sync::atomic::AtomicUsize::new(0),
+        });
+        assert!(
+            matches!(
+                persist_scanner_checkpoint(
+                    set.clone(),
+                    fence_store.clone(),
+                    ScannerCheckpointPersistContext {
+                        ctx: &ctx,
+                        expected_publication_epoch: publication_epoch,
+                        cycle,
+                        leader_epoch
+                    },
+                    &cache_name,
+                    &checkpoint,
+                    &mut revisions,
+                )
+                .await,
+                ScannerCheckpointPersistResult::FenceChanged
+            ),
+            "unreadable or changed global fence must reject checkpoint success: {case}"
+        );
+        assert_eq!(fence_store.reads.load(Ordering::SeqCst), expected_reads, "global fence checks: {case}");
+        assert_eq!(revisions, initial, "rejected checkpoint must not refresh caller revisions: {case}");
+        let persisted = DataUsageCache::read_revisions(set.clone(), &cache_name)
+            .await
+            .expect("read persisted checkpoint");
+        if expected_reads == 1 {
+            assert_eq!(persisted, initial, "pre-save rejection must not mutate persisted caches: {case}");
+        } else {
+            assert_ne!(persisted, initial, "post-save fence check must run after a real cache write: {case}");
+        }
+    }
+}
+
 async fn wait_for_namespace_commit_tails(store: &ECStore) {
    tokio::time::timeout(Duration::from_secs(30), async {
        while store.scanner_data_usage_publication_blocked().await {
@@ -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 |
@@ -1769,18 +1769,25 @@ impl DefaultMultipartUsecase {
            .into());
        }

-        let (src_reader, _source_cancellation) = store
-            .get_object_reader_for_copy(&src_bucket, &src_key, rs.clone(), h, &get_opts)
+        let prepared_source = store
+            .prepare_get_object_reader(&src_bucket, &src_key, rs.clone(), h, &get_opts)
            .await
            .map_err(map_get_object_reader_error)?;

-        let src_info = src_reader.object_info;
+        let src_info = prepared_source.object_info();
+        if src_info.delete_marker {
+            let error = if src_opts.version_id.is_none() {
+                StorageError::ObjectNotFound(src_bucket.clone(), src_key.clone())
+            } else {
+                StorageError::MethodNotAllowed
+            };
+            return Err(map_get_object_reader_error(error).into());
+        }

        // Same shape as CopyObject: the part copy reads the source plaintext, and the source
        // read resolves its material inside the object layer, which carries no request identity.
        authorize_sse_kms_object_read(copy_principal.as_ref(), &src_info.user_defined).await?;

-        let src_stream = src_reader.stream;
        let resolved_src_version_id = src_info.version_id.map(|version_id| {
            if version_id == Uuid::nil() {
                "null".to_string()
@@ -1811,12 +1818,13 @@ impl DefaultMultipartUsecase {
            return Err(s3_error!(PreconditionFailed));
        }

-        let source_logical_size = match src_info.get_actual_size() {
-            Ok(size) if size >= 0 => size,
-            Ok(_) | Err(_) if destination_size_limit.is_some() => {
+        let source_actual_size = src_info.get_actual_size().ok().filter(|size| *size >= 0);
+        let source_logical_size = match source_actual_size {
+            Some(size) => size,
+            None if destination_size_limit.is_some() => {
                return Err(S3Error::new(S3ErrorCode::UnexpectedContent));
            }
-            Ok(_) | Err(_) => src_info.size,
+            None => src_info.size,
        };

        let (_start_offset, length) = if let Some(ref range_spec) = rs {
@@ -1832,6 +1840,26 @@ impl DefaultMultipartUsecase {
            (0, source_logical_size)
        };

+        // Resolve the range under the prepared source's read lock, then try the
+        // shared budget without waiting while holding namespace/lifecycle locks.
+        // Keep both the budget and source cancellation alive through the write.
+        let admission_length = source_actual_size.map_or(-1, |_| length);
+        let _copy_admission = match self.concurrency_manager().try_admit_multipart_part_copy(admission_length) {
+            ForegroundWriteAdmission::Disabled => None,
+            ForegroundWriteAdmission::Admitted(permit) => Some(permit),
+            ForegroundWriteAdmission::Rejected => {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::SlowDown,
+                    "foreground write concurrency limit reached, please reduce your request rate",
+                ));
+            }
+        };
+        let (src_reader, _source_cancellation) = prepared_source
+            .into_reader_for_copy()
+            .await
+            .map_err(map_get_object_reader_error)?;
+        let src_stream = src_reader.stream;
+
        let is_disk_compressed =
            rustfs_utils::http::contains_key_str(&mp_info.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION);

@@ -2035,6 +2063,7 @@ mod tests {
    use super::*;

    mod body_read_tests;
+    mod copy_admission_tests;
    use http::{Extensions, HeaderMap, Method, Uri, header::HeaderValue};
    use rustfs_filemeta::ObjectPartInfo;
    use rustfs_utils::http::{
@@ -0,0 +1,316 @@
+// Copyright 2026 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use super::*;
+use crate::app::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
+use crate::app::storage_api::test::contract::object::{ObjectIO, ObjectOperations};
+
+#[test]
+#[serial_test::serial]
+fn upload_part_copy_shares_foreground_admission_with_upload_part() {
+    crate::app::gating_test_env::run_large_stack_test("copy-part-admission", || async {
+        struct RestoreGetMetrics(bool);
+        impl Drop for RestoreGetMetrics {
+            fn drop(&mut self) {
+                rustfs_io_metrics::set_get_stage_metrics_enabled(self.0);
+            }
+        }
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+        let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+        let bucket = format!("copy-part-admission-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create bucket");
+        let bytes = vec![0x5a; 2 * 1024 * 1024];
+        store
+            .put_object(&bucket, "source", &mut PutObjReader::from_vec(bytes.clone()), &ObjectOptions::default())
+            .await
+            .expect("write copy source");
+        let upload = store
+            .new_multipart_upload(&bucket, "object", &ObjectOptions::default())
+            .await
+            .expect("create session");
+        let manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(true, 1, 1, Duration::ZERO));
+        let held = manager.admit_multipart_part(1024).await.expect("hold UploadPart permit");
+        assert!(matches!(held, ForegroundWriteAdmission::Admitted(_)));
+        let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::clone(&manager));
+        let input = UploadPartCopyInput::builder()
+            .bucket(bucket.clone())
+            .key("object".to_owned())
+            .copy_source(CopySource::Bucket {
+                bucket: bucket.clone().into(),
+                key: "source".into(),
+                version_id: None,
+            })
+            .part_number(1)
+            .upload_id(upload.upload_id.clone())
+            .build()
+            .expect("copy request");
+        let recorder = metrics_util::debugging::DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+        let _recorder = metrics::set_default_local_recorder(&recorder);
+        let _restore = RestoreGetMetrics(rustfs_io_metrics::get_stage_metrics_enabled());
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        let source_size_bucket = rustfs_io_metrics::get_object_size_bucket(i64::try_from(bytes.len()).expect("source size"));
+        let source_reader_observed = || {
+            snapshotter.snapshot().into_vec().iter().any(|(key, _, _, _)| {
+                key.key().name() == "rustfs_io_get_object_reader_path_by_size_total"
+                    && key
+                        .key()
+                        .labels()
+                        .any(|label| label.key() == "size_bucket" && label.value() == source_size_bucket)
+            })
+        };
+        let error = Box::pin(usecase.execute_upload_part_copy(build_request(input.clone(), Method::PUT)))
+            .await
+            .expect_err("copy must not bypass the occupied UploadPart pool");
+        assert_eq!(error.code(), &S3ErrorCode::SlowDown);
+        assert!(!source_reader_observed(), "saturated copy must not construct a source body reader");
+        let parts = store
+            .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+            .await
+            .expect("list parts");
+        assert!(parts.parts.is_empty(), "rejected copy must not create a destination part");
+        drop(held);
+        let copied = Box::pin(usecase.execute_upload_part_copy(build_request(input, Method::PUT)))
+            .await
+            .expect("retry copy after releasing UploadPart permit");
+        assert!(source_reader_observed());
+        assert_eq!(
+            manager.put_object_admission_snapshot().active,
+            Some(0),
+            "success releases the copy permit"
+        );
+        let etag = copied
+            .output
+            .copy_part_result
+            .expect("part result")
+            .e_tag
+            .expect("copy ETag")
+            .value()
+            .trim_matches('"')
+            .to_owned();
+        Arc::clone(&store)
+            .complete_multipart_upload(
+                &bucket,
+                "object",
+                &upload.upload_id,
+                vec![CompletePart {
+                    part_num: 1,
+                    etag: Some(etag),
+                    ..Default::default()
+                }],
+                &ObjectOptions::default(),
+            )
+            .await
+            .expect("commit copied object");
+        let mut reader = store
+            .get_object_reader(&bucket, "object", None, HeaderMap::new(), &ObjectOptions::default())
+            .await
+            .expect("read committed copy");
+        let mut actual = Vec::new();
+        reader.stream.read_to_end(&mut actual).await.expect("read all copied bytes");
+        assert_eq!(actual, bytes);
+        drop(reader);
+        let marker = store
+            .delete_object(
+                &bucket,
+                "source",
+                ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("create source delete marker");
+        assert!(marker.delete_marker);
+        let marker_version = marker.version_id.expect("marker version").to_string();
+        let deleted_upload = store
+            .new_multipart_upload(&bucket, "deleted-copy", &ObjectOptions::default())
+            .await
+            .expect("deleted-source destination session");
+        let _held = manager
+            .admit_multipart_part(1024)
+            .await
+            .expect("saturate gate for deleted source");
+        for (version_id, expected) in [
+            (None, S3ErrorCode::NoSuchKey),
+            (Some(marker_version), S3ErrorCode::MethodNotAllowed),
+        ] {
+            let input = UploadPartCopyInput::builder()
+                .bucket(bucket.clone())
+                .key("deleted-copy".to_owned())
+                .copy_source(CopySource::Bucket {
+                    bucket: bucket.clone().into(),
+                    key: "source".into(),
+                    version_id: version_id.map(Into::into),
+                })
+                .copy_source_if_match(Some("\"wrong-etag\"".parse().expect("source condition")))
+                .part_number(1)
+                .upload_id(deleted_upload.upload_id.clone())
+                .build()
+                .expect("deleted-source copy request");
+            let error = Box::pin(usecase.execute_upload_part_copy(build_request(input, Method::PUT)))
+                .await
+                .expect_err("deleted source must fail before conditions or admission");
+            assert_eq!(error.code(), &expected);
+        }
+    });
+}
+
+#[test]
+#[serial_test::serial]
+fn upload_part_copy_range_error_and_cancellation_release_foreground_permit() {
+    crate::app::gating_test_env::run_large_stack_test("copy-part-range-admission", || async {
+        use rustfs_utils::http::SUFFIX_MAX_TOTAL_OBJECT_SIZE;
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        let ambient = crate::app::gating_test_env::shared_gating_ambient().await;
+        let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+        let bucket = format!("copy-part-range-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create range bucket");
+        let bytes = (0..2 * 1024 * 1024)
+            .map(|i| u8::try_from(i % 251).expect("byte pattern"))
+            .collect::<Vec<_>>();
+        store
+            .put_object(&bucket, "source", &mut PutObjReader::from_vec(bytes.clone()), &ObjectOptions::default())
+            .await
+            .expect("write non-inline source");
+        let manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(true, 1, 1, Duration::ZERO));
+        let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::clone(&manager));
+        let cancelled_upload = store
+            .new_multipart_upload(&bucket, "cancelled", &ObjectOptions::default())
+            .await
+            .expect("cancelled destination session");
+        let cancelled_input = UploadPartCopyInput::builder()
+            .bucket(bucket.clone())
+            .key("cancelled".to_owned())
+            .copy_source(CopySource::Bucket {
+                bucket: bucket.clone().into(),
+                key: "source".into(),
+                version_id: None,
+            })
+            .part_number(1)
+            .upload_id(cancelled_upload.upload_id.clone())
+            .build()
+            .expect("cancelled copy request");
+        let mut cancelled_copy = Box::pin(usecase.execute_upload_part_copy(build_request(cancelled_input, Method::PUT)));
+        tokio::time::timeout(Duration::from_secs(10), async {
+            tokio::select! {
+                biased;
+                () = async {
+                    while manager.put_object_admission_snapshot().active != Some(1) {
+                        tokio::task::yield_now().await;
+                    }
+                } => {}
+                result = &mut cancelled_copy => panic!("copy completed before cancellation: {result:?}"),
+            }
+        })
+        .await
+        .expect("copy acquires its permit");
+        drop(cancelled_copy);
+        assert_eq!(manager.put_object_admission_snapshot().active, Some(0));
+        tokio::time::timeout(
+            Duration::from_secs(10),
+            store.put_object(&bucket, "source", &mut PutObjReader::from_vec(bytes.clone()), &ObjectOptions::default()),
+        )
+        .await
+        .expect("cancelled source reader releases its namespace lock")
+        .expect("overwrite source after cancellation");
+        store
+            .abort_multipart_upload(&bucket, "cancelled", &cancelled_upload.upload_id, &ObjectOptions::default())
+            .await
+            .expect("abort cancelled session");
+        for capped in [true, false] {
+            let mut opts = ObjectOptions::default();
+            if capped {
+                insert_str(&mut opts.user_defined, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "1024".to_owned());
+            }
+            let upload = store
+                .new_multipart_upload(&bucket, "object", &opts)
+                .await
+                .expect("range destination session");
+            let input = UploadPartCopyInput::builder()
+                .bucket(bucket.clone())
+                .key("object".to_owned())
+                .copy_source(CopySource::Bucket {
+                    bucket: bucket.clone().into(),
+                    key: "source".into(),
+                    version_id: None,
+                })
+                .copy_source_range(Some("bytes=12345-1060920".to_owned()))
+                .part_number(1)
+                .upload_id(upload.upload_id.clone())
+                .build()
+                .expect("range request");
+            let response = Box::pin(usecase.execute_upload_part_copy(build_request(input, Method::PUT))).await;
+            assert_eq!(
+                manager.put_object_admission_snapshot().active,
+                Some(0),
+                "copy releases its permit on every outcome"
+            );
+            if capped {
+                assert_eq!(
+                    response.expect_err("range exceeds destination size limit").code(),
+                    &S3ErrorCode::EntityTooLarge
+                );
+                let parts = store
+                    .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+                    .await
+                    .expect("failed copy parts");
+                assert!(parts.parts.is_empty());
+                store
+                    .abort_multipart_upload(&bucket, "object", &upload.upload_id, &ObjectOptions::default())
+                    .await
+                    .expect("abort capped session");
+            } else {
+                let copied = response.expect("range copy after failed request released its source and permit");
+                let etag = copied
+                    .output
+                    .copy_part_result
+                    .expect("range result")
+                    .e_tag
+                    .expect("range ETag")
+                    .value()
+                    .trim_matches('"')
+                    .to_owned();
+                Arc::clone(&store)
+                    .complete_multipart_upload(
+                        &bucket,
+                        "object",
+                        &upload.upload_id,
+                        vec![CompletePart {
+                            part_num: 1,
+                            etag: Some(etag),
+                            ..Default::default()
+                        }],
+                        &ObjectOptions::default(),
+                    )
+                    .await
+                    .expect("commit range copy");
+                let mut reader = store
+                    .get_object_reader(&bucket, "object", None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("open range copy");
+                let mut actual = Vec::new();
+                reader.stream.read_to_end(&mut actual).await.expect("read full range");
+                assert_eq!(actual, bytes[12345..=1060920]);
+            }
+        }
+    });
+}
@@ -33,7 +33,6 @@ use bytes::Bytes;
 use futures::StreamExt as _;
 use p256::ecdsa::{Signature, SigningKey, signature::Signer as _};
 use p256::pkcs8::DecodePrivateKey as _;
-use percent_encoding::{NON_ALPHANUMERIC, utf8_percent_encode};
 use reqwest::{Client, Method, Response, StatusCode, Url};
 use serde::{Deserialize, Serialize};
 use sha2::{Digest as _, Sha256};
@@ -1317,7 +1316,7 @@ async fn drain_response(

 fn object_url(endpoint: &Url, bucket: &str, key: &str, version_id: Option<&str>) -> Result<Url, SiteReplicationProbeError> {
    let mut url = endpoint
-        .join(&format!("{bucket}/{}", encode_path(key)))
+        .join(&format!("{bucket}/{key}"))
        .map_err(|_| SiteReplicationProbeError::ProtocolFailure)?;
    if let Some(version_id) = version_id {
        url.query_pairs_mut().append_pair("versionId", version_id);
@@ -1333,14 +1332,6 @@ fn list_versions_url(endpoint: &Url, bucket: &str, key: &str) -> Result<Url, Sit
    Ok(url)
 }

-fn encode_path(value: &str) -> String {
-    value
-        .split('/')
-        .map(|segment| utf8_percent_encode(segment, NON_ALPHANUMERIC).to_string())
-        .collect::<Vec<_>>()
-        .join("/")
-}
-
 fn deployment_endpoint(value: &str) -> Result<Url, SiteReplicationPerformanceError> {
    let mut url = Url::parse(value).map_err(|_| SiteReplicationPerformanceError::InvalidEndpoint)?;
    let local_http = url.scheme() == "http"
@@ -1655,6 +1646,12 @@ mod tests {
    #[test]
    fn cleanup_queries_are_version_specific_and_task_scoped() {
        let endpoint = Url::parse("https://source.example/").expect("endpoint");
+        let scoped = object_url(&endpoint, "scratch-bucket", "rustfs-connect/site-replication/019c-1234", None)
+            .expect("scoped object URL");
+        assert_eq!(
+            scoped.as_str(),
+            "https://source.example/scratch-bucket/rustfs-connect/site-replication/019c-1234"
+        );
        let object = object_url(&endpoint, "scratch-bucket", "path/a b", Some("version+1")).expect("object URL");
        assert_eq!(object.as_str(), "https://source.example/scratch-bucket/path/a%20b?versionId=version%2B1");
        let list = list_versions_url(&endpoint, "scratch-bucket", "path/a b").expect("list URL");
@@ -397,6 +397,7 @@ impl RelayHttpClient {
            let response = self
                .client
                .post(self.receive_url.clone())
+                .header(header::ORIGIN, self.receive_url.origin().ascii_serialization())
                .header(header::COOKIE, self.cookie.clone())
                .header("X-XSRF-TOKEN", self.csrf_token.clone())
                .header(header::ACCEPT, "application/json")
@@ -233,11 +233,7 @@ impl ForegroundWriteAdmissionGate {

    async fn admit(&self, permits: u32) -> Result<ForegroundWriteAdmission, tokio::sync::AcquireError> {
        if self.wait_timeout.is_zero() {
-            return Ok(match self.semaphore.clone().try_acquire_many_owned(permits) {
-                Ok(permit) => ForegroundWriteAdmission::Admitted(permit),
-                Err(tokio::sync::TryAcquireError::NoPermits) => ForegroundWriteAdmission::Rejected,
-                Err(tokio::sync::TryAcquireError::Closed) => ForegroundWriteAdmission::Rejected,
-            });
+            return Ok(self.try_admit(permits));
        }

        match tokio::time::timeout(self.wait_timeout, self.semaphore.clone().acquire_many_owned(permits)).await {
@@ -245,6 +241,13 @@ impl ForegroundWriteAdmissionGate {
            Err(_) => Ok(ForegroundWriteAdmission::Rejected),
        }
    }
+
+    fn try_admit(&self, permits: u32) -> ForegroundWriteAdmission {
+        match self.semaphore.clone().try_acquire_many_owned(permits) {
+            Ok(permit) => ForegroundWriteAdmission::Admitted(permit),
+            Err(_) => ForegroundWriteAdmission::Rejected,
+        }
+    }
 }

 #[derive(Clone)]
@@ -787,6 +790,27 @@ impl ConcurrencyManager {
            .await
    }

+    /// Admit a copy part through the UploadPart pool using its resolved logical length.
+    /// Copy holds source metadata and bucket locks here, so it must never queue
+    /// behind a writer that may need those locks. Saturation returns SlowDown.
+    pub fn try_admit_multipart_part_copy(&self, size: i64) -> ForegroundWriteAdmission {
+        match &self.foreground_write_admission_policy {
+            ForegroundWriteAdmissionPolicy::Disabled | ForegroundWriteAdmissionPolicy::LegacyCounterOnly => {
+                ForegroundWriteAdmission::Disabled
+            }
+            ForegroundWriteAdmissionPolicy::Strict(gate) => gate.try_admit(1),
+            ForegroundWriteAdmissionPolicy::Large {
+                gate,
+                large_request_permits,
+                multipart_part_min_size_bytes,
+                ..
+            } if should_gate_foreground_write(size, *multipart_part_min_size_bytes) => {
+                gate.try_admit(multipart_admission_permits(size, *large_request_permits))
+            }
+            _ => ForegroundWriteAdmission::Disabled,
+        }
+    }
+
    // ============================================
    // Adaptive I/O Strategy Methods
    // ============================================
@@ -1271,9 +1295,9 @@ mod integration_tests {
    use super::super::io_schedule::{IoLoadLevel, IoPriority};
    use super::super::request_guard::GetObjectGuard;
    use super::{
-        ConcurrencyManager, ForegroundWriteAdmission, ForegroundWriteAdmissionPolicy, SNOWBALL_ARCHIVE_DECODER_LIMIT,
-        SNOWBALL_MEMBER_COMMIT_LIMIT, SNOWBALL_STAGING_BYTES_LIMIT, derive_large_put_admission_limit,
-        derive_multipart_admission_max_pending,
+        ConcurrencyManager, ForegroundWriteAdmission, ForegroundWriteAdmissionGate, ForegroundWriteAdmissionPolicy,
+        SNOWBALL_ARCHIVE_DECODER_LIMIT, SNOWBALL_MEMBER_COMMIT_LIMIT, SNOWBALL_STAGING_BYTES_LIMIT,
+        derive_large_put_admission_limit, derive_multipart_admission_max_pending,
    };
    use crate::storage::storage_api::concurrency_consumer::PutObjectGuard;
    use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
@@ -1715,6 +1739,75 @@ mod integration_tests {
        assert_eq!(manager.put_object_admission_snapshot().queued, Some(0));
    }

+    #[tokio::test]
+    #[serial]
+    async fn test_concurrency_manager_multipart_copy_never_queues_with_source_locks() {
+        for (manager, expected_queued) in [
+            (
+                ConcurrencyManager::with_multipart_admission_queue_for_test(1, Duration::from_secs(30), 4),
+                Some(0),
+            ),
+            (ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::from_secs(30)), None),
+        ] {
+            let held = manager
+                .admit_multipart_part(32 * 1024 * 1024)
+                .await
+                .expect("hold write budget");
+            assert!(matches!(
+                manager.try_admit_multipart_part_copy(32 * 1024 * 1024),
+                ForegroundWriteAdmission::Rejected
+            ));
+            assert_eq!(manager.put_object_admission_snapshot().queued, expected_queued);
+            drop(held);
+            let copy = manager.try_admit_multipart_part_copy(32 * 1024 * 1024);
+            assert!(matches!(copy, ForegroundWriteAdmission::Admitted(_)));
+            assert!(matches!(manager.try_admit_multipart_part_copy(1), ForegroundWriteAdmission::Rejected));
+            drop(copy);
+            assert!(matches!(manager.try_admit_multipart_part_copy(1), ForegroundWriteAdmission::Admitted(_)));
+        }
+    }
+
+    #[test]
+    #[serial]
+    fn test_concurrency_manager_multipart_copy_uses_logical_size_threshold_and_weight() {
+        let mut manager = ConcurrencyManager::new();
+        manager.foreground_write_admission_policy = ForegroundWriteAdmissionPolicy::Large {
+            gate: ForegroundWriteAdmissionGate::new(4, Duration::from_secs(30)),
+            large_request_permits: 4,
+            put_object_min_size_bytes: 32 * 1024 * 1024,
+            multipart_part_min_size_bytes: 8 * 1024 * 1024,
+            multipart_wait_timeout: Duration::from_secs(30),
+            multipart_max_pending: 4,
+        };
+        assert!(matches!(
+            manager.try_admit_multipart_part_copy(8 * 1024 * 1024 - 1),
+            ForegroundWriteAdmission::Disabled
+        ));
+        let small = manager.try_admit_multipart_part_copy(8 * 1024 * 1024);
+        assert_eq!(manager.put_object_admission_snapshot().active, Some(1));
+        assert!(matches!(
+            manager.try_admit_multipart_part_copy(32 * 1024 * 1024),
+            ForegroundWriteAdmission::Rejected
+        ));
+        drop(small);
+        let full = manager.try_admit_multipart_part_copy(32 * 1024 * 1024);
+        assert!(matches!(full, ForegroundWriteAdmission::Admitted(_)));
+        assert_eq!(manager.put_object_admission_snapshot().active, Some(4));
+        drop(full);
+        assert_eq!(manager.put_object_admission_snapshot().active, Some(0));
+        let unknown = manager.try_admit_multipart_part_copy(-1);
+        assert!(matches!(unknown, ForegroundWriteAdmission::Admitted(_)));
+        assert_eq!(manager.put_object_admission_snapshot().active, Some(4));
+        drop(unknown);
+        for enabled in [false, true] {
+            let disabled = ConcurrencyManager::with_put_admission_for_test(enabled, 0, Duration::ZERO);
+            assert!(matches!(
+                disabled.try_admit_multipart_part_copy(32 * 1024 * 1024),
+                ForegroundWriteAdmission::Disabled
+            ));
+        }
+    }
+
    #[tokio::test(start_paused = true)]
    #[serial]
    async fn test_concurrency_manager_multipart_part_rejects_after_queue_wait_timeout() {
@@ -13,9 +13,18 @@
 // limitations under the License.

 use std::collections::HashMap;
+use std::convert::Infallible;
+use std::sync::{Arc, Mutex};
+use std::time::Duration;

 use base64_simd::{STANDARD as BASE64_STANDARD, URL_SAFE_NO_PAD};
 use ed25519_dalek::{Signer as _, SigningKey};
+use http_body_util::{BodyExt, Full};
+use hyper::body::Bytes;
+use hyper::service::service_fn;
+use hyper::{Request, Response, StatusCode};
+use hyper_util::rt::TokioIo;
+use rcgen::generate_simple_self_signed;
 use rustfs::connect::relay;
 use rustfs::connect::relay::{
    RelayDirection, RelayEnvelope, RelayError, RelayMaterialKind, RelayParty, RelayReceiptOutcome, RelayReceiptPayload,
@@ -23,6 +32,8 @@ use rustfs::connect::relay::{
 };
 use serde::Serialize;
 use sha2::{Digest as _, Sha256};
+use tokio::net::TcpListener;
+use tokio_rustls::TlsAcceptor;

 const TRANSFER_UID: &str = "0198f3a1-a200-7b20-8b22-112233445566";

@@ -137,6 +148,112 @@ fn trusted(key: &SigningKey) -> TrustedReceiptSigner {
    TrustedReceiptSigner::new(hex_lower(&Sha256::digest(public_key)), public_key).unwrap()
 }

+#[tokio::test]
+async fn http_relay_preserves_session_origin_and_credentials_across_retry() {
+    let _ = rustls::crypto::aws_lc_rs::default_provider().install_default();
+    let certificate = generate_simple_self_signed(vec!["localhost".to_owned()]).unwrap();
+    let config = rustls::ServerConfig::builder()
+        .with_no_client_auth()
+        .with_single_cert(
+            vec![certificate.cert.der().clone()],
+            rustls::pki_types::PrivatePkcs8KeyDer::from(certificate.signing_key.serialize_der()).into(),
+        )
+        .unwrap();
+    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let origin = format!("https://localhost:{}", listener.local_addr().unwrap().port());
+    let endpoint = format!("{origin}/console/api/");
+    let organization = "0198f3a1-4c00-7a10-8b21-0c1d2e3f4a50";
+    let signing_key = SigningKey::from_bytes(&[13; 32]);
+    let trust = trusted(&signing_key);
+    let seen = Arc::new(Mutex::new(Vec::new()));
+    let captured = seen.clone();
+    let expected_origin = origin.clone();
+    let server = tokio::spawn(async move {
+        let acceptor = TlsAcceptor::from(Arc::new(config));
+        for attempt in 0..2 {
+            let (stream, _) = listener.accept().await.unwrap();
+            let stream = acceptor.accept(stream).await.unwrap();
+            let captured = captured.clone();
+            let expected_origin = expected_origin.clone();
+            let signing_key = signing_key.clone();
+            let service = service_fn(move |request: Request<hyper::body::Incoming>| {
+                let captured = captured.clone();
+                let expected_origin = expected_origin.clone();
+                let signing_key = signing_key.clone();
+                async move {
+                    let (parts, body) = request.into_parts();
+                    let bytes = body.collect().await.unwrap().to_bytes();
+                    let body: serde_json::Value = serde_json::from_slice(&bytes).unwrap();
+                    let authenticated = parts.headers.get("origin").and_then(|value| value.to_str().ok())
+                        == Some(expected_origin.as_str())
+                        && parts.headers.get("cookie").unwrap() == "session=relay-test-only"
+                        && parts.headers.get("x-xsrf-token").unwrap() == "relay-csrf-test-only";
+                    captured.lock().unwrap().push((parts, body.clone()));
+                    let (status, response) = if !authenticated {
+                        (StatusCode::UNAUTHORIZED, Vec::new())
+                    } else if attempt == 0 {
+                        (StatusCode::SERVICE_UNAVAILABLE, Vec::new())
+                    } else {
+                        let envelope = serde_json::from_value(body["envelope"].clone()).unwrap();
+                        let destination = Destination::new(signing_key, false);
+                        (StatusCode::OK, destination.receipt(&envelope, RelayReceiptOutcome::Applied))
+                    };
+                    Ok::<_, Infallible>(
+                        Response::builder()
+                            .status(status)
+                            .header("connection", "close")
+                            .body(Full::new(Bytes::from(response)))
+                            .unwrap(),
+                    )
+                }
+            });
+            hyper::server::conn::http1::Builder::new()
+                .serve_connection(TokioIo::new(stream), service)
+                .await
+                .unwrap();
+        }
+    });
+    let client = relay::RelayHttpClient::new(
+        &endpoint,
+        certificate.cert.pem().as_bytes(),
+        organization,
+        "approved-transfer".to_owned(),
+        "session=relay-test-only",
+        "relay-csrf-test-only",
+        Duration::from_secs(5),
+        None,
+    )
+    .unwrap();
+    let prepared = relay::prepare_approved_artifact(
+        TRANSFER_UID,
+        RelayMaterialKind::DiagnosticBundleManifest,
+        b"original signed manifest",
+        party("DEVICE", "organizations/o/clusters/c/clusterDevices/d", true),
+        party("CONNECT", "organizations/o", false),
+        |_| true,
+    )
+    .unwrap();
+    let result = client.deliver(prepared, &trust).await;
+    server.abort();
+    let delivery = result.expect("session-authenticated relay delivery");
+    assert_eq!(delivery.attempts, 2);
+    assert_eq!(delivery.receipt.outcome, RelayReceiptOutcome::Applied);
+    let seen = seen.lock().unwrap();
+    assert_eq!(seen.len(), 2);
+    for (parts, body) in seen.iter() {
+        assert_eq!(parts.method, "POST");
+        assert_eq!(
+            parts.uri.path(),
+            format!("/console/api/organizations/{organization}/relayTransfers:receive")
+        );
+        assert_eq!(parts.headers["origin"], origin);
+        assert_eq!(parts.headers["cookie"], "session=relay-test-only");
+        assert_eq!(parts.headers["x-xsrf-token"], "relay-csrf-test-only");
+        assert_eq!(body["approvalReference"], "approved-transfer");
+    }
+    assert_eq!(seen[0].1, seen[1].1);
+}
+
 #[test]
 fn interrupted_transfer_retries_exact_bytes_and_applies_once() {
    let signing_key = SigningKey::from_bytes(&[7; 32]);
@@ -113,12 +113,27 @@ def fault_tolerance_counts(text):
    return counts


+def auto_testing_dir():
+    """Locate the auto-testing checkout. Lanes that run this script from a
+    subdirectory checkout (security keeps its rustfs clone in rustfs-repo/)
+    still check auto-testing out at the workspace root, so ROOT alone is not
+    always the right base."""
+    candidates = [ROOT / "auto-testing"]
+    workspace = os.environ.get("GITHUB_WORKSPACE")
+    if workspace:
+        candidates.append(Path(workspace) / "auto-testing")
+    for candidate in candidates:
+        if (candidate / ".git").exists():
+            return candidate
+    return candidates[0]
+
+
 def record(chain, suite, report, output):
    require(suite in SUITES, "unknown suite")
    result = {"schema": 1, "suite": suite, "chain": chain, "valid": False, "counts": {}, "report_sha256": None}
    error = None
    try:
-        private_head = subprocess.check_output(["git", "-C", "auto-testing", "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
+        private_head = subprocess.check_output(["git", "-C", str(auto_testing_dir()), "rev-parse", "HEAD"], cwd=ROOT, text=True).strip()
        require(private_head == chain["testing_sha"], "suite used a different private script revision")
        require(report.is_file() and 0 < report.stat().st_size <= MAX_REPORT, "missing, empty or oversized report")
        data = report.read_bytes()
Author	SHA1	Message	Date
overtrue	b17a6f162c	fix(storage): publish delete rollback backups atomically Write rollback backups to a temporary file outside the rollback directory and publish only complete writes. Preserve acknowledged versions when backup creation fails partway through. Add an isolated child-process short-write regression that verifies all 100 versions survive rollback and a subsequent write preserves 101 versions.	2026-10-03 06:47:48 +08:00
Chris	4aa3cde1d9	fix(connect): send endpoint origin for relay session authentication (#8310 )	2026-10-03 01:06:25 +08:00
Chris	87c1c851c2	ci: remove Docker Hub description sync (#8303 )	2026-10-02 20:04:45 +08:00
Hauser	370b517b4f	fix: resolve Rust 1.99 Clippy warnings (#8301 ) Replace deprecated atomic fetch_update calls with try_update, remove redundant closure borrows, and drop unnecessary async-recursion macros. Keep atomic orderings and explicitly boxed recursive calls unchanged. Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>	2026-10-02 10:11:22 +08:00
Chris	791af1888f	fix(connect): preserve site replication S3 key path (#8300 )	2026-10-02 05:43:24 +08:00
Chris	50e6c907f9	fix(s3): bound multipart copy admission and HTTP write buffers (#8288 )	2026-10-01 21:51:27 +08:00
hector	1efa86b8a3	ci: locate the auto-testing checkout for lanes that run evidence from a subdirectory (#8279 ) ## Related Issues Follow-up to #8229. ## Summary of Changes Locate the private auto-testing checkout from the lane root or the workspace root so the nested security checkout can record functional-chain evidence. ## Verification The exact PR head `b66129ab9f` passed one mechanical correctness and simplicity review, nine real-Git layout and provenance checks, and sixteen existing evidence/envelope tests. The baseline sibling layout failed with git exit 128; the corrected layout succeeded while revision mismatches and missing checkouts stayed rejected. Current PR checks are completed with successful or skipped conclusions, including the aggregate. ## Impact Both lane and private-script revision checks remain intact. No time limits, assertions, production behavior, or evidence validation requirements change. The synthetic layout checks do not execute the actual scheduled security suite; integrated main CI and release acceptance remain separate gates. ## Additional Notes Approved review 5374559393 is bound to the exact head above. Reverting the single-file change restores the previous checkout lookup.	2026-10-01 11:24:37 +08:00
Chris	a33dd896e2	fix(ci): restore E2E membership and pagination timeouts (#8281 )	2026-10-01 09:00:02 +08:00
AL	6796382cee	fix(scanner): validate checkpoints against global cycle fence (#8278 ) ## Related Issues Related to rustfs/backlog#2701. ## Summary of Changes Route scanner checkpoint cycle and leader validation through the global store while retaining the owning set for cache persistence, CAS revisions and publication admission. ## Verification Two independent final-diff source reviews found no issues across correctness, concurrency and durability, test coverage, compatibility, performance and simplicity on head `8b8fe51d092090b053f552ae283960e2e306be33`. Root approval `5373624714` is bound to that exact head. Regression tests cover real two-pool routing, stale fences, post-save rejection and CAS conflicts; their reported local execution belongs to the PR author, not this merge operation. Current required CI remains pending, and this authorized admin squash does not establish CI or runtime acceptance. ## Impact Restores checkpoint progress when global cycle and leader state differ from a set-scoped view. No format, retry, timeout, assertion or scanner-policy changes are introduced by this diff. The three prior main scanner failures remain unproved repaired. ## Additional Notes Full validation must run on the resulting exact main revision. Reverting this patch restores the earlier set-scoped fence lookup and its checkpoint rejection behavior.	2026-10-01 08:52:25 +08:00
overtrue	99a2af8687	docs(release): validate candidates on release branch	2026-10-01 08:37:30 +08:00