fix(app): keep relative paths valid in moved multipart code

Code that moves from multipart_usecase.rs into an operation file is one
module deeper, so its super:: paths gain a super:: (the codemod now does
this) and the regenerated split carries the result.

.config/make/tests.mak

@@ -62,10 +62,12 @@ script-tests: ## Run shell script tests
 	./scripts/check_embedded_secrets.sh --self-test
 	./scripts/check_s3s_footprint.sh --self-test
 	./scripts/test_check_test_count_ratchet.sh
+	./scripts/test_guard_source_set.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/codemods/s3_error_to_s3_types.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_download_release_asset.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_reference_binary.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_btf_prototypes.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py
 	$(RUSTFS_PYTHON_BIN) -m unittest discover -s scripts/release -p "test_*.py"
@@ -92,6 +94,8 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_junit_diff.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_runner_tools.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_api_coverage.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_reference_parity.py
 	bash -n ./scripts/validate_object_data_cache_cold_stampede.sh
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_object_data_cache_follower_samples.py --self-test
 	./scripts/validate_object_data_cache_cold_stampede.sh --self-test

.config/nextest.toml

@@ -137,7 +137,7 @@ filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair
 setup = 'ecstore-base-stack'
 
 [[profile.default.scripts]]
-filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
+filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::complete::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'
 
 # These handler/store fixtures and the relocated-pool GET compose deep storage
@@ -455,7 +455,7 @@ filter = '(package(rustfs) & test(/^(admin::handlers::replication::target_repair
 setup = 'ecstore-base-stack'
 
 [[profile.ci.scripts]]
-filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
+filter = 'package(rustfs-ecstore) | package(rustfs-s3select-api) | package(rustfs-scanner) | (package(rustfs) & test(/^(app::multipart_usecase::complete::tests::concurrent_completions_share_durable_bucket_quota_reservations|app::object::delete::tests::compressed_delete_requests_update_observed_usage_without_releasing_quota_floor|app::object::internal_put::tests::internal_multipart_roundtrip_completes_and_abort_leaves_nothing|app::object::restore::tests::execute_restore_object_maps_failures_to_typed_s3_errors|storage::access::tests::(delete_object_access_captures_authorized_bucket_incarnation|copy_operations_reject_recreated_source_bucket_after_authorization|request_slot_keeps_bucket_policy_bound_to_its_store))$/))'
 setup = 'ecstore-base-stack'
 
 [[profile.ci.scripts]]

.config/reference-binary.toml

@@ -0,0 +1,20 @@
+# Pinned RustFS release used as the "reference" target of parity lanes
+# (.github/workflows/reference-parity.yml). This is the single place that names
+# the reference release; .github/actions/reference-binary resolves, downloads and
+# verifies it, and fails on a missing pin, a missing asset or a digest mismatch.
+#
+# Bumping the reference is a reviewed event: every per-test difference between
+# the candidate and the reference is measured against this exact build.
+#
+# Each sha256 is the value published in the release's SHA256SUMS asset, which
+# matches the digest GitHub reports for the same asset. The action checks both.
+
+schema = 1
+repository = "rustfs/rustfs"
+version = "1.0.5"
+tag = "1.0.5"
+
+[platforms.linux-x86_64-gnu]
+asset = "rustfs-linux-x86_64-gnu-v1.0.5.zip"
+sha256 = "5addae2f74756bf80104d0975e6347cda843bc2daffdfcd1f0cb80118b95cbd0"
+executable = "rustfs"

.config/reference-parity-targets.toml

@@ -0,0 +1,24 @@
+# Targets of the s3-tests reference parity lane (.github/workflows/reference-parity.yml).
+#
+# Exactly one target uses `binary = "reference"` (the release pinned in
+# .config/reference-binary.toml). Every other target uses `binary = "candidate"`
+# (the binary built from the branch under test) and is compared per node id
+# against the reference. All targets run the same harness, test list and server
+# configuration; `env` adds only the target's own RUSTFS_* settings.
+#
+# Adding a target is one more [[targets]] table, for example:
+#
+#   [[targets]]
+#   name = "candidate-gateway"
+#   binary = "candidate"
+#   env = { RUSTFS_S3_STACK = "gateway" }
+
+schema = 1
+
+[[targets]]
+name = "reference"
+binary = "reference"
+
+[[targets]]
+name = "candidate-legacy"
+binary = "candidate"

.config/reference-parity-tolerated.txt

@@ -0,0 +1,12 @@
+# Tolerated per-test differences of the s3-tests reference parity lane.
+#
+# One entry per line, four fields separated by " | ":
+#
+#   <target> | <pytest node id> | <tracking issue URL> | <reason>
+#
+# <target> is a candidate name from .config/reference-parity-targets.toml, the
+# node id is exactly as the lane prints it (s3tests/functional/test_s3.py::...),
+# and the issue URL is https://github.com/rustfs/<repo>/issues/<number>.
+#
+# Same semantics as scripts/junit_diff.py --allow: an entry whose difference no
+# longer occurs is stale and fails the lane, so remove it when it is fixed.

.github/actions/reference-binary/action.yml

@@ -0,0 +1,96 @@
+# 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.
+
+name: "Reference RustFS binary"
+description: "Download the pinned reference RustFS release, verify its sha256 and expose the executable path. A missing pin, missing asset or digest mismatch fails the step."
+
+inputs:
+  pin-file:
+    description: "Pin file naming the reference release (repository-relative)"
+    required: false
+    default: ".config/reference-binary.toml"
+  platform:
+    description: "Platform table in the pin file"
+    required: false
+    default: "linux-x86_64-gnu"
+  destination:
+    description: "Directory that receives the executable; defaults to $RUNNER_TEMP/reference-binary"
+    required: false
+    default: ""
+
+outputs:
+  path:
+    description: "Absolute path of the verified executable"
+    value: ${{ steps.unpack.outputs.path }}
+  version:
+    description: "Pinned reference version"
+    value: ${{ steps.pin.outputs.version }}
+  sha256:
+    description: "Pinned and verified archive sha256"
+    value: ${{ steps.pin.outputs.sha256 }}
+
+runs:
+  using: "composite"
+  steps:
+    - name: Resolve reference pin
+      id: pin
+      shell: bash
+      env:
+        PIN_TOOL: ${{ github.action_path }}/reference_binary.py
+        PIN_FILE: ${{ inputs.pin-file }}
+        PIN_PLATFORM: ${{ inputs.platform }}
+      run: |
+        set -euo pipefail
+        resolved="$(python3 "$PIN_TOOL" resolve --pin "$PIN_FILE" --platform "$PIN_PLATFORM")"
+        printf '%s\n' "$resolved"
+        printf '%s\n' "$resolved" >> "$GITHUB_OUTPUT"
+
+    - name: Download and verify reference archive
+      id: download
+      shell: bash
+      env:
+        GH_TOKEN: ${{ github.token }}
+        ASSET_DOWNLOADER: ${{ github.action_path }}/../setup/download_release_asset.py
+        REF_REPOSITORY: ${{ steps.pin.outputs.repository }}
+        REF_TAG: ${{ steps.pin.outputs.tag }}
+        REF_ASSET: ${{ steps.pin.outputs.asset }}
+        REF_SHA256: ${{ steps.pin.outputs.sha256 }}
+      run: |
+        set -euo pipefail
+        archive="$RUNNER_TEMP/reference-archive/$REF_ASSET"
+        # Fails unless the release lists exactly one asset of this name whose
+        # published digest equals the pin and whose downloaded bytes match it.
+        python3 "$ASSET_DOWNLOADER" --repository "$REF_REPOSITORY" --tag "$REF_TAG" \
+          --asset "$REF_ASSET" --output "$archive" --sha256 "$REF_SHA256"
+        echo "archive=$archive" >> "$GITHUB_OUTPUT"
+
+    - name: Unpack reference executable
+      id: unpack
+      shell: bash
+      env:
+        PIN_TOOL: ${{ github.action_path }}/reference_binary.py
+        REF_ARCHIVE: ${{ steps.download.outputs.archive }}
+        REF_SHA256: ${{ steps.pin.outputs.sha256 }}
+        REF_EXECUTABLE: ${{ steps.pin.outputs.executable }}
+        REF_DESTINATION: ${{ inputs.destination }}
+      run: |
+        set -euo pipefail
+        destination="${REF_DESTINATION:-$RUNNER_TEMP/reference-binary}"
+        unpacked="$(python3 "$PIN_TOOL" unpack --archive "$REF_ARCHIVE" --sha256 "$REF_SHA256" \
+          --executable "$REF_EXECUTABLE" --destination "$destination")"
+        path="${unpacked#path=}"
+        test -x "$path"
+        echo "Reference executable: $path"
+        "$path" --version
+        echo "path=$path" >> "$GITHUB_OUTPUT"

.github/actions/reference-binary/reference_binary.py

@@ -0,0 +1,144 @@
+#!/usr/bin/env python3
+# 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.
+"""Resolve and unpack the pinned reference RustFS release binary.
+
+`resolve` reads the pin file and prints the selected platform's identity as
+`key=value` lines (GitHub step-output format). `unpack` re-verifies the
+downloaded archive digest against the pin and extracts exactly one executable.
+Every missing or malformed input is a failure; nothing is skipped.
+"""
+
+import argparse
+import hashlib
+from pathlib import Path
+import re
+import sys
+import tomllib
+import zipfile
+
+REPOSITORY = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*/[A-Za-z0-9][A-Za-z0-9_.-]*")
+TAG = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+-]*")
+PLATFORM = re.compile(r"[a-z0-9][a-z0-9_-]*")
+ASSET = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.+-]*\.zip")
+SHA256 = re.compile(r"[0-9a-f]{64}")
+EXECUTABLE = re.compile(r"[A-Za-z0-9][A-Za-z0-9_.-]*")
+MAX_EXECUTABLE_BYTES = 1024 * 1024 * 1024
+
+
+class PinError(Exception):
+    """A pin, archive or argument that cannot be trusted."""
+
+
+def field(table, key, pattern, where):
+    value = table.get(key)
+    if not isinstance(value, str) or not pattern.fullmatch(value):
+        raise PinError(f"{where}: missing or invalid '{key}'")
+    return value
+
+
+def resolve(pin_path, platform):
+    if not PLATFORM.fullmatch(platform or ""):
+        raise PinError("invalid platform name")
+    try:
+        data = tomllib.loads(Path(pin_path).read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise PinError(f"{pin_path}: reference pin file is missing") from exc
+    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
+        raise PinError(f"{pin_path}: cannot read reference pin: {exc}") from exc
+    if data.get("schema") != 1 or type(data.get("schema")) is not int:
+        raise PinError(f"{pin_path}: unsupported pin schema")
+    identity = {
+        "repository": field(data, "repository", REPOSITORY, pin_path),
+        "version": field(data, "version", TAG, pin_path),
+        "tag": field(data, "tag", TAG, pin_path),
+    }
+    platforms = data.get("platforms")
+    if not isinstance(platforms, dict) or not platforms:
+        raise PinError(f"{pin_path}: no pinned platforms")
+    entry = platforms.get(platform)
+    if not isinstance(entry, dict):
+        raise PinError(f"{pin_path}: no pin for platform '{platform}'")
+    where = f"{pin_path} [platforms.{platform}]"
+    identity["asset"] = field(entry, "asset", ASSET, where)
+    identity["sha256"] = field(entry, "sha256", SHA256, where)
+    identity["executable"] = field(entry, "executable", EXECUTABLE, where)
+    if identity["version"] not in identity["asset"]:
+        raise PinError(f"{where}: asset name does not carry version {identity['version']}")
+    unknown = set(entry) - {"asset", "sha256", "executable"}
+    if unknown:
+        raise PinError(f"{where}: unknown keys {sorted(unknown)}")
+    return identity
+
+
+def file_sha256(path):
+    with open(path, "rb") as stream:
+        return hashlib.file_digest(stream, "sha256").hexdigest()
+
+
+def unpack(archive, sha256, executable, destination):
+    if not SHA256.fullmatch(sha256 or "") or not EXECUTABLE.fullmatch(executable or ""):
+        raise PinError("invalid expected digest or executable name")
+    try:
+        actual = file_sha256(archive)
+    except OSError as exc:
+        raise PinError(f"{archive}: reference archive is missing or unreadable") from exc
+    if actual != sha256:
+        raise PinError(f"{archive}: sha256 {actual} does not match the pinned digest")
+    try:
+        with zipfile.ZipFile(archive) as bundle:
+            matches = [info for info in bundle.infolist() if info.filename == executable]
+            if len(matches) != 1:
+                raise PinError(f"{archive}: must contain exactly one top-level '{executable}'")
+            info = matches[0]
+            if info.is_dir() or not 0 < info.file_size <= MAX_EXECUTABLE_BYTES:
+                raise PinError(f"{archive}: '{executable}' is not a plausible executable")
+            destination = Path(destination)
+            destination.mkdir(parents=True, exist_ok=True)
+            target = destination / executable
+            with bundle.open(info) as source, open(target, "wb") as sink:
+                while chunk := source.read(1024 * 1024):
+                    sink.write(chunk)
+    except zipfile.BadZipFile as exc:
+        raise PinError(f"{archive}: not a valid zip archive") from exc
+    target.chmod(0o755)
+    return target
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="mode", required=True)
+    res = sub.add_parser("resolve")
+    res.add_argument("--pin", required=True)
+    res.add_argument("--platform", required=True)
+    unp = sub.add_parser("unpack")
+    unp.add_argument("--archive", required=True)
+    unp.add_argument("--sha256", required=True)
+    unp.add_argument("--executable", required=True)
+    unp.add_argument("--destination", required=True)
+    args = parser.parse_args(argv)
+    try:
+        if args.mode == "resolve":
+            for key, value in resolve(args.pin, args.platform).items():
+                print(f"{key}={value}")
+        else:
+            print(f"path={unpack(args.archive, args.sha256, args.executable, args.destination)}")
+    except PinError as exc:
+        print(f"reference binary: {exc}", file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

.github/scheduled-validations.json

@@ -869,6 +869,33 @@
       "failure_owner": "watchdog",
       "origin": "baseline"
     },
+    {
+      "workflow": ".github/workflows/reference-parity.yml",
+      "name": "Reference Parity",
+      "roles": [
+        "pr",
+        "periodic",
+        "manual"
+      ],
+      "events": [
+        "pull_request",
+        "schedule",
+        "workflow_dispatch"
+      ],
+      "cadence": "nightly",
+      "timezone": "UTC",
+      "schedule": [
+        "37 1 * * *"
+      ],
+      "owner": "@overtrue",
+      "policy": "active",
+      "monitor": "workflow",
+      "max_age_hours": 36,
+      "result_owner": "reference-parity.yml",
+      "disposition": "s3-tests per-node-id parity of branch candidates against the pinned reference release",
+      "failure_owner": "watchdog",
+      "origin": "added"
+    },
     {
       "workflow": ".github/workflows/functional-chain-health.yml",
       "name": "Functional chain health",

.github/workflows/reference-parity.yml

@@ -0,0 +1,308 @@
+# 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.
+
+# s3-tests reference parity lane (rustfs/backlog#2773).
+#
+# Runs the ceph/s3-tests implemented list (scripts/s3-tests/implemented_tests.txt)
+# against every target in .config/reference-parity-targets.toml and fails on
+# any per-node-id difference between a candidate and the reference:
+#
+#   - reference: the release pinned in .config/reference-binary.toml, downloaded
+#     and sha256-verified by .github/actions/reference-binary;
+#   - candidate targets: the debug binary built from this checkout exactly as
+#     ci.yml's build-rustfs-debug-binary job builds it for S3 Implemented Tests.
+#
+# Every target runs the same harness (scripts/s3-tests/run.sh from this
+# checkout), test list and server configuration on its own hosted runner. A
+# target's failures get one rerun against a fresh server; a case that passes it
+# is flaky, not a difference. scripts/s3-tests/reference_parity.py judges the
+# result and honours .config/reference-parity-tolerated.txt, whose entries must
+# each match a real difference.
+#
+# Report-only: not part of the Test and Lint aggregate. Scheduled failures are
+# owned by the scheduled-validation watchdog (.github/scheduled-validations.json).
+
+name: Reference Parity
+
+on:
+  pull_request:
+    paths:
+      - ".github/workflows/reference-parity.yml"
+      - ".github/actions/reference-binary/**"
+      - ".config/reference-binary.toml"
+      - ".config/reference-parity-targets.toml"
+      - ".config/reference-parity-tolerated.txt"
+      - "scripts/s3-tests/reference_parity.py"
+      - "scripts/junit_diff.py"
+  schedule:
+    - cron: "37 1 * * *"
+  workflow_dispatch:
+
+# Matches ci.yml: these variables are hashed into the rust-cache key, so the
+# candidate build reads the same ci-dev cache as build-rustfs-debug-binary.
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BACKTRACE: 1
+
+concurrency:
+  group: ${{ github.workflow }}-${{ github.ref }}
+  cancel-in-progress: true
+
+permissions:
+  contents: read
+
+defaults:
+  run:
+    shell: bash
+
+jobs:
+  plan:
+    name: Plan parity targets
+    runs-on: ubuntu-latest
+    timeout-minutes: 5
+    outputs:
+      targets: ${{ steps.targets.outputs.targets }}
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Validate target list
+        id: targets
+        run: |
+          set -euo pipefail
+          targets="$(python3 scripts/s3-tests/reference_parity.py targets --config .config/reference-parity-targets.toml)"
+          echo "targets=${targets}" >> "$GITHUB_OUTPUT"
+          echo "Parity targets: ${targets}"
+
+  build-candidate:
+    name: Build candidate binary
+    # Critical-path CPU and cache ownership: the same debug build as ci.yml's
+    # build-rustfs-debug-binary, reading the ci-dev cache written on sm-standard-4.
+    runs-on: sm-standard-4
+    timeout-minutes: 60
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: stable
+          cache-shared-key: ci-dev
+          cache-save-if: 'false'
+          install-build-packaging-tools: 'false'
+
+      - name: Build debug binary
+        run: python3 scripts/e2e_binary.py build --bins --features e2e-test-hooks
+
+      - name: Upload candidate binary
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-candidate-binary
+          overwrite: true
+          path: |
+            target/debug/rustfs
+            target/debug/rustfs.e2e.json
+          if-no-files-found: error
+          retention-days: 1
+
+  s3tests:
+    name: s3-tests (${{ matrix.target.name }})
+    needs: [ plan, build-candidate ]
+    runs-on: ubuntu-latest
+    timeout-minutes: 75
+    strategy:
+      fail-fast: false
+      matrix:
+        target: ${{ fromJSON(needs.plan.outputs.targets) }}
+    env:
+      TARGET_NAME: ${{ matrix.target.name }}
+      TARGET_KIND: ${{ matrix.target.binary }}
+      # Above this many first-run failures the target is systemically broken;
+      # a rerun would only repeat it, so the first run is judged as is.
+      RERUN_CAP: "50"
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Fetch pinned reference binary
+        id: reference
+        if: matrix.target.binary == 'reference'
+        uses: ./.github/actions/reference-binary
+
+      - name: Download candidate binary
+        if: matrix.target.binary == 'candidate'
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          name: reference-parity-candidate-binary
+          path: target/debug
+
+      - name: Select target binary
+        env:
+          REFERENCE_PATH: ${{ steps.reference.outputs.path }}
+        run: |
+          set -euo pipefail
+          case "$TARGET_KIND" in
+            reference) binary="$REFERENCE_PATH" ;;
+            candidate) chmod +x target/debug/rustfs; binary="$PWD/target/debug/rustfs" ;;
+            *) echo "unknown target binary kind: $TARGET_KIND" >&2; exit 1 ;;
+          esac
+          test -x "$binary"
+          echo "TARGET_BINARY=$binary" >> "$GITHUB_ENV"
+          mkdir -p "parity/$TARGET_NAME"
+          {
+            echo "target=$TARGET_NAME"
+            echo "binary=$TARGET_KIND"
+            echo "sha256=$(sha256sum "$binary" | cut -d' ' -f1)"
+            echo "version=$("$binary" --version | tr '\n' ' ')"
+          } | tee "parity/$TARGET_NAME/target.txt"
+
+      - name: Apply target environment
+        env:
+          TARGET_JSON: ${{ toJSON(matrix.target) }}
+        run: |
+          set -euo pipefail
+          python3 scripts/s3-tests/reference_parity.py export-env --target-json "$TARGET_JSON" \
+            | tee -a "$GITHUB_ENV" "parity/$TARGET_NAME/target-env.txt"
+
+      - name: Run implemented s3-tests
+        run: |
+          set -uo pipefail
+          report="parity/$TARGET_NAME"
+          harness="artifacts/s3tests-single"
+
+          # Same invocation as ci.yml's S3 Implemented Tests, with a fresh data
+          # root and port per run. $1 optionally replaces the test list.
+          run_suite() {
+            local list="$1" label="$2" root port
+            root="${RUNNER_TEMP}/rustfs-s3tests-${label}"
+            rm -rf "$harness" "$root"
+            mkdir -p "$root"
+            port="$(python3 -c 'import socket; s=socket.socket(); s.bind(("127.0.0.1", 0)); print(s.getsockname()[1]); s.close()')"
+            env DEPLOY_MODE=binary \
+              RUSTFS_BINARY="$TARGET_BINARY" \
+              TEST_MODE=single \
+              MAXFAIL=0 \
+              XDIST=4 \
+              S3_PORT="$port" \
+              DATA_ROOT="$root" \
+              S3TESTS_CONF="$harness/s3tests.conf" \
+              ${list:+IMPLEMENTED_TESTS_FILE="$list"} \
+              ./scripts/s3-tests/run.sh
+          }
+
+          # A run is usable only if it produced a JUnit report; test failures
+          # are data for the comparison, a missing report is a lane failure.
+          keep() {
+            local prefix="$1"
+            for file in pytest.log compat-report.md; do
+              if [[ -f "$harness/$file" ]]; then cp "$harness/$file" "$report/${prefix}${file}"; fi
+            done
+            if [[ -f "$harness/rustfs-single/rustfs.log" ]]; then cp "$harness/rustfs-single/rustfs.log" "$report/${prefix}rustfs.log"; fi
+            test -s "$harness/junit.xml" || { echo "::error::$TARGET_NAME produced no JUnit report (${prefix:-first} run)"; return 1; }
+            cp "$harness/junit.xml" "$report/${prefix}junit.xml"
+          }
+
+          run_suite "" first
+          echo "first_run_exit=$?" >> "$report/target.txt"
+          keep "" || exit 1
+          cp "$harness/selected-nodeids.txt" "$report/selected-nodeids.txt" || exit 1
+
+          failures="$(python3 scripts/s3-tests/reference_parity.py failed --report "$report")" || exit 1
+          count="$(printf '%s' "$failures" | grep -c . || true)"
+          echo "first_run_failures=$count" >> "$report/target.txt"
+          if (( count == 0 )); then
+            exit 0
+          fi
+          if (( count > RERUN_CAP )); then
+            echo "::warning::$TARGET_NAME has $count failures (cap $RERUN_CAP); judging the first run without a rerun"
+            echo "rerun=skipped-over-cap" >> "$report/target.txt"
+            exit 0
+          fi
+          printf '%s\n' "$failures" > "$RUNNER_TEMP/rerun-tests.txt"
+          run_suite "$RUNNER_TEMP/rerun-tests.txt" rerun
+          echo "rerun_exit=$?" >> "$report/target.txt"
+          keep rerun- || exit 1
+
+      - name: Upload target report
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-report-${{ matrix.target.name }}
+          overwrite: true
+          path: parity/${{ matrix.target.name }}/
+          if-no-files-found: error
+          retention-days: 14
+
+  compare:
+    name: Compare targets
+    needs: [ plan, s3tests ]
+    if: always() && needs.plan.result == 'success'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - name: Checkout repository
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+
+      - name: Download target reports
+        uses: actions/download-artifact@3e5f45b2cfb9172054b4087a40e8e0b5a5461e7c # v8.0.1
+        with:
+          pattern: reference-parity-report-*
+          path: parity-download
+
+      - name: Compare per node id
+        env:
+          TARGETS: ${{ needs.plan.outputs.targets }}
+          S3TESTS_RESULT: ${{ needs.s3tests.result }}
+        run: |
+          set -uo pipefail
+          mkdir -p parity
+          for dir in parity-download/reference-parity-report-*; do
+            if [[ -d "$dir" ]]; then mv "$dir" "parity/${dir#parity-download/reference-parity-report-}"; fi
+          done
+          python3 scripts/s3-tests/reference_parity.py compare \
+            --targets-json "$TARGETS" \
+            --reports parity \
+            --tolerated .config/reference-parity-tolerated.txt \
+            --markdown "$GITHUB_STEP_SUMMARY" \
+            --json parity/diff.json > parity/diff.md
+          status=$?
+          cat parity/diff.md
+          if [[ "$S3TESTS_RESULT" != "success" ]]; then
+            echo "::error::s3-tests target jobs finished with '$S3TESTS_RESULT'"
+            status=1
+          fi
+          exit "$status"
+
+      - name: Upload parity diff
+        if: always()
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: reference-parity-diff
+          overwrite: true
+          path: |
+            parity/diff.md
+            parity/diff.json
+          if-no-files-found: warn
+          retention-days: 14

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

@@ -41,19 +41,53 @@ struct DiscoverySession {
     touched: Instant,
 }
 
-/// Plain (non-transaction) children of an object-scoped discovery root cost one
-/// budget entry per this many names: enough to bound a hostile prefix without
-/// making a real folder prefix undeletable.
+/// Names an object-scoped scan skips without a stat (plain children of the
+/// key's directory, and entries of a UUID child that cannot be delete evidence)
+/// cost one budget entry per this many names: enough to bound a hostile prefix
+/// without making a real folder prefix undeletable.
 const PLAIN_CHILDREN_PER_BUDGET_ENTRY: u32 = 64;
 
+/// Entries one step of the resumable (background) discovery may visit. The
+/// foreground bounded protocol has its own, larger per-call allowance.
+const DISCOVERY_STEP_ENTRIES: u32 = 4096;
+
+/// Wall-clock allowance of one bounded discovery call: a third of the drive
+/// timeout that wraps the call (including its wait for a lane worker), capped at
+/// ten seconds, so an incomplete walk returns before the call can time out.
+fn bounded_discovery_time_limit() -> std::time::Duration {
+    (crate::disk::disk_store::get_max_timeout_duration() / 3).min(std::time::Duration::from_secs(10))
+}
+
+/// Whether an entry inside a UUID child of an object-scoped root can name one
+/// of the key's delete transactions: a journal file of a rollback directory or
+/// a data-directory marker. Every other name is inert to discovery.
+fn may_be_delete_evidence(name: &str) -> bool {
+    [PREPARED, STORAGE_FORMAT_FILE_BACKUP, DELETE_QUORUM_COMMITTED_FILE].contains(&name)
+        || [
+            DELETE_DATA_DIR_MARKER_PREFIX,
+            RESERVED_DELETE_DATA_DIR_MARKER_PREFIX,
+            RECOVERY_DELETE_DATA_DIR_MARKER_PREFIX,
+        ]
+        .iter()
+        .any(|prefix| delete_transaction_marker_id(name, prefix).is_some())
+}
+
 impl DiscoverySession {
     /// Consume one bounded step, returning ownership only when traversal remains.
     fn advance(self, limit: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>)> {
-        self.advance_bounded(limit, super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES)
+        self.advance_bounded(limit, DISCOVERY_STEP_ENTRIES, &|| false)
             .map(|(transactions, next, _)| (transactions, next))
     }
 
-    fn advance_bounded(mut self, limit: u32, max_entries: u32) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
+    /// Walk until `limit` transactions, `max_entries` charged entries, the end
+    /// of the scope, or `should_stop` (checked before every entry); any early
+    /// stop returns the remaining traversal as continuation state.
+    fn advance_bounded(
+        mut self,
+        limit: u32,
+        max_entries: u32,
+        should_stop: &dyn Fn() -> bool,
+    ) -> std::io::Result<(Vec<DeleteTransactionId>, Option<Self>, u32)> {
         if self.frames.is_empty() {
             match std::fs::read_dir(&self.root) {
                 Ok(entries) => self.frames.push((String::new(), entries)),
@@ -66,10 +100,11 @@ impl DiscoverySession {
         let mut transactions = BTreeSet::new();
         let mut seen_candidates = HashSet::new();
         let mut visited = 0u32;
-        let mut plain_children = 0u32;
+        let mut skipped_by_name = 0u32;
         while !self.frames.is_empty()
             && transactions.len() < usize::try_from(limit).map_err(|_| std::io::Error::from(ErrorKind::InvalidInput))?
             && visited < max_entries
+            && !should_stop()
         {
             let (parent, entries) = self
                 .frames
@@ -90,13 +125,22 @@ impl DiscoverySession {
                 return Err(std::io::Error::from(ErrorKind::InvalidData));
             }
             // An object-scoped scan only ever descends into transaction (UUID)
-            // directories at the root. Plain child keys of a prefix directory are
-            // skipped by name without a stat; they are charged at one entry per
-            // readdir batch so the work stays bounded while deleting a missing key
-            // that names a prefix with thousands of children can still complete.
-            if self.object.is_some() && parent.is_empty() && !canonical_uuid(&name).is_some_and(|id| !id.is_nil()) {
-                plain_children += 1;
-                if plain_children.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
+            // directories at the root, and inside them only journal files and
+            // data-directory markers can be evidence. Every other name (plain
+            // child keys of a prefix directory, and the `xl.meta`, data
+            // directories or children of a UUID-named child key) is skipped by
+            // name without a stat and charged at one entry per readdir batch, so
+            // the work stays bounded while deleting a missing key that names a
+            // folder prefix with many children, UUID-named or not, can complete.
+            if self.object.is_some()
+                && if parent.is_empty() {
+                    !canonical_uuid(&name).is_some_and(|id| !id.is_nil())
+                } else {
+                    !may_be_delete_evidence(&name)
+                }
+            {
+                skipped_by_name += 1;
+                if skipped_by_name.is_multiple_of(PLAIN_CHILDREN_PER_BUDGET_ENTRY) {
                     visited += 1;
                 }
                 continue;
@@ -1391,8 +1435,16 @@ impl LocalDisk {
         self.validate_delete_discovery_owner(&session)?;
         // The blocking owner drops every directory handle even if the RPC waiter disappears.
         // A partial inventory is never evidence of absence and never enters DISCOVERY_SESSIONS.
-        os::run_blocking_namespace_operation(lease, move || {
-            let (transactions, next, visited) = session.advance_bounded(256, max_entries)?;
+        // A large folder prefix may take a while to walk, so the probe rides the
+        // durable-mutation lane like bounded inspection, never the one-worker
+        // foreground lane that every other mutation on the drive queues behind.
+        // It also stops, incomplete, when its waiter goes away or well before the
+        // drive timeout, so a slow walk answers SlowDown instead of marking the
+        // drive failed or holding a lane worker after the request is gone.
+        let deadline = Instant::now() + bounded_discovery_time_limit();
+        os::run_cancellable_blocking_namespace_operation_with_sync(lease, true, move |is_cancelled| {
+            let should_stop = || is_cancelled() || Instant::now() >= deadline;
+            let (transactions, next, visited) = session.advance_bounded(256, max_entries, &should_stop)?;
             Ok(super::super::BoundedDeleteRecoveryDiscovery {
                 transactions: if next.is_none() { transactions } else { Vec::new() },
                 visited,
@@ -6561,6 +6613,93 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_charges_uuid_named_children_once_and_keeps_evidence() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        std::fs::create_dir_all(&path).expect("physical prefix");
+        // UUID-named child keys of the folder: each holds its own `xl.meta` and a
+        // data directory, neither of which can be evidence for the folder's key.
+        const CHILDREN: u32 = 2000;
+        for _ in 0..CHILDREN {
+            let child = path.join(Uuid::new_v4().to_string());
+            std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+            std::fs::write(child.join(STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+        }
+        // A UUID-named sub-prefix with children of its own.
+        const SUB_PREFIX_CHILDREN: u32 = 640;
+        let sub_prefix = path.join(Uuid::new_v4().to_string());
+        for index in 0..SUB_PREFIX_CHILDREN {
+            std::fs::create_dir_all(sub_prefix.join(format!("child-{index}"))).expect("sub-prefix child");
+        }
+        // Evidence of two interrupted transactions of the key itself: a journaled
+        // rollback directory, and a data-directory marker whose rollback is gone.
+        let journaled = Uuid::new_v4();
+        std::fs::create_dir_all(path.join(journaled.to_string())).expect("rollback directory");
+        std::fs::write(path.join(journaled.to_string()).join(PREPARED), b"{}").expect("journal");
+        let marker_only = Uuid::new_v4();
+        let data_dir = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&data_dir).expect("residual data directory");
+        std::fs::write(data_dir.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{marker_only}")), [])
+            .expect("reservation marker");
+
+        let page = disk
+            .discover_delete_transactions_bounded(
+                "recovery",
+                "prefix",
+                super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES,
+            )
+            .await
+            .expect("bounded probe");
+        assert!(page.complete, "a folder of UUID-named keys is a complete inventory");
+        // One entry per descent (children, sub-prefix, rollback and data
+        // directory) and per evidence file; inert names at 1/64.
+        let inert = 2 * CHILDREN + SUB_PREFIX_CHILDREN;
+        assert_eq!(page.visited, CHILDREN + 3 + 2 + inert / PLAIN_CHILDREN_PER_BUDGET_ENTRY);
+        let found = page.transactions.iter().map(|id| id.transaction).collect::<BTreeSet<_>>();
+        assert!(found.contains(&journaled), "a journaled rollback directory is evidence");
+        assert!(found.contains(&marker_only), "a marker-only residue is evidence");
+        assert_eq!(found.len(), 2, "no child key is mistaken for a transaction");
+    }
+
+    #[tokio::test]
+    async fn delete_recovery_bounded_discovery_stops_on_request_and_skips_only_inert_irregular_entries() {
+        let (_temp, disk, _, _) = fixture().await;
+        let path = disk.delete_discovery_path("recovery", Some("prefix")).expect("prefix path");
+        let child = path.join(Uuid::new_v4().to_string());
+        std::fs::create_dir_all(&child).expect("UUID-named child key");
+        // A stop request (cancellation or the wall-clock limit) ends the walk
+        // before any entry, leaving continuation state: never evidence of absence.
+        let session = disk
+            .new_delete_discovery_session("recovery", Some("prefix"), &path)
+            .expect("private session");
+        let (transactions, next, visited) = session
+            .advance_bounded(256, super::super::super::DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES, &|| true)
+            .expect("stopped walk");
+        assert!(transactions.is_empty());
+        assert_eq!(visited, 0);
+        assert!(next.is_some(), "a stopped walk is incomplete");
+        drop(next);
+        #[cfg(unix)]
+        {
+            // An irregular entry whose name cannot be evidence is skipped unread.
+            std::os::unix::fs::symlink(&path, child.join("inert-link")).expect("inert symlink");
+            let page = disk
+                .discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                .await
+                .expect("inert irregular entries are not evidence");
+            assert!(page.complete && page.transactions.is_empty());
+            // One with an evidence name still fails closed.
+            std::os::unix::fs::symlink(&path, child.join(PREPARED)).expect("evidence-named symlink");
+            assert!(
+                disk.discover_delete_transactions_bounded("recovery", "prefix", 4096)
+                    .await
+                    .is_err(),
+                "an evidence-named symlink is corruption, not a skipped name"
+            );
+        }
+    }
+
     #[tokio::test]
     async fn delete_recovery_bounded_discovery_refuses_partial_inventory_without_public_cursor() {
         let (_temp, disk, _, _) = fixture().await;
@@ -6663,8 +6802,11 @@ mod tests {
         let filler_id = Uuid::new_v4().to_string();
         let filler = path.join(&filler_id);
         fs::create_dir_all(&filler).await.expect("non-evidence directory");
-        for index in 0..5000 {
-            fs::write(filler.join(format!("part-{index:04}")), [])
+        // Inert names are skipped at 1/64, so fill the step with entries that
+        // pass the evidence-name filter but are directories: each costs a full
+        // budget entry and names no transaction.
+        for _ in 0..5000 {
+            fs::create_dir(filler.join(format!("{DELETE_DATA_DIR_MARKER_PREFIX}{}", Uuid::new_v4())))
                 .await
                 .expect("non-evidence entry");
         }

crates/ecstore/src/disk/mod.rs

@@ -2585,7 +2585,13 @@ pub struct BoundedDeleteRecoveryDiscovery {
     pub complete: bool,
 }
 
-pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 4096;
+/// Entries one bounded discovery call may charge on one drive. Object-scoped
+/// discovery charges each transaction-shaped (UUID) child of the key's
+/// directory once and every name that cannot be delete evidence at
+/// 1/`PLAIN_CHILDREN_PER_BUDGET_ENTRY`, so this admits a missing-key delete of a
+/// folder prefix with this many UUID-named children per drive. The request
+/// budget (`DeleteRecoveryWorkBudget`) still bounds the total across keys.
+pub const DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES: u32 = 262_144;
 pub const DELETE_RECOVERY_MAX_INSPECTION_ENTRIES: u32 = 65536;
 
 #[derive(Debug, Clone, Serialize, Deserialize)]

crates/ecstore/src/object_api/types.rs

@@ -975,9 +975,11 @@ pub struct DeleteRecoveryWorkBudget {
     remaining: std::sync::atomic::AtomicU32,
 }
 
-/// Entries one disk may visit per recovery step (discovery plus inspection).
-/// Every disk of every set a request fans out to is admitted up to this much
-/// and charged on completion, so a budget must cover that many disks.
+/// Per-drive unit a request budget is sized by (a typical discovery plus one
+/// inspection). Admission charges what a call visited, not this unit, so it only
+/// scales the request total with the drives a request can reach; it is kept
+/// independent of the per-call caps, which may let one key's probe use more of
+/// the same total on a large folder.
 const DELETE_RECOVERY_RESERVATION_PER_DISK: u32 = 4096 + 65_536;
 
 impl Default for DeleteRecoveryWorkBudget {

crates/ecstore/src/set_disk/mod.rs

@@ -11040,7 +11040,12 @@ mod tests {
         let live = Uuid::new_v4();
         let orphan = Uuid::new_v4();
         let object_dir = dir.path().join("bucket/obj");
-        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &[live]).await;
+        // Part files are inert to discovery and cost 1/64 of an entry, so the
+        // first step is exhausted by what still costs a full entry each: the
+        // data directories of many versions, all referenced by the metadata.
+        let versions = (0..4200).map(|_| Uuid::new_v4()).collect::<Vec<_>>();
+        let referenced = std::iter::once(live).chain(versions.iter().copied()).collect::<Vec<_>>();
+        write_object_meta_with_data_dirs(&object_dir, "bucket", "obj", &referenced).await;
         let live_dir = object_dir.join(live.to_string());
         let orphan_dir = object_dir.join(orphan.to_string());
         fs::create_dir_all(&live_dir).await.expect("create live multipart directory");
@@ -11049,10 +11054,14 @@ mod tests {
             .await
             .expect("write orphan payload");
         let parts_dir = live_dir.clone();
+        let versions_root = object_dir.clone();
         tokio::task::spawn_blocking(move || {
             for part in 1..=4200 {
                 std::fs::write(parts_dir.join(format!("part.{part}")), b"live").expect("write live multipart entry");
             }
+            for version in versions {
+                std::fs::create_dir(versions_root.join(version.to_string())).expect("create referenced version directory");
+            }
         })
         .await
         .expect("create multipart fixture");
@@ -11062,7 +11071,10 @@ mod tests {
             .await
             .expect("read first discovery page");
         assert!(first.transactions.is_empty());
-        assert!(first.next.is_some(), "ordinary multipart entries must exhaust the first scan budget");
+        assert!(
+            first.next.is_some(),
+            "the data directories of many versions must exhaust the first scan step"
+        );
         let last = disk
             .discover_delete_transactions("bucket", Some("obj"), first.next, 1)
             .await

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

@@ -1709,6 +1709,37 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn delete_recovery_missing_key_naming_a_folder_of_uuid_named_keys_completes() {
+        let bucket = "recover-uuid-named-folder";
+        let (dirs, store, _) = fixture(bucket).await;
+        // More UUID-named child keys per drive than the former 4096-entry probe
+        // could cover once each child's `xl.meta` and data directory were charged
+        // (three entries per child).
+        const CHILDREN: usize = 1500;
+        for dir in &dirs {
+            let prefix = dir.path().join(bucket).join("uploads");
+            for _ in 0..CHILDREN {
+                let child = prefix.join(Uuid::new_v4().to_string());
+                std::fs::create_dir_all(child.join(Uuid::new_v4().to_string())).expect("child key data dir");
+                std::fs::write(child.join(crate::disk::STORAGE_FORMAT_FILE), b"meta").expect("child key metadata");
+            }
+        }
+        let error = store
+            .delete_object(bucket, "uploads", ObjectOptions::default())
+            .await
+            .expect_err("the folder's own key does not exist");
+        assert!(
+            crate::error::is_err_object_not_found(&error),
+            "a complete inventory keeps the missing-key no-op semantics: {error:?}"
+        );
+        assert!(dirs.iter().all(|dir| {
+            std::fs::read_dir(dir.path().join(bucket).join("uploads"))
+                .map(|entries| entries.count() == CHILDREN)
+                .unwrap_or(false)
+        }));
+    }
+
     #[tokio::test]
     async fn delete_recovery_batch_budget_survives_pool_and_set_fanout() {
         use crate::layout::endpoint::Endpoint;

docs/architecture/s3-compatibility-matrix.md

@@ -130,7 +130,7 @@ These behaviors are client-visible and differ from AWS S3 or MinIO. They bound h
 | Bucket tag key or value outside the AWS tag character class (`[\p{L}\p{N}\p{Z}+\-=._:/@]`) | `400 InvalidTag` on `PutBucketTagging` | AWS enforces the same class; MinIO is more permissive | AWS parity. Legacy stored tags still load. |
 | `CopyObject` with the default `x-amz-tagging-directive: COPY` from a source that carries tags | `403 AccessDenied` unless the caller also holds `s3:GetObjectTagging` (or `s3:GetObjectVersionTagging`) on the source; the destination's `s3:RequestObjectTag/*` conditions are evaluated against the copied tag set | AWS and MinIO copy the tags with `s3:GetObject` alone | A copy would otherwise disclose source tags to a principal that may not read them, and source tags could satisfy destination conditions the caller cannot assert (rustfs/backlog#3034). Use `REPLACE` with an explicit `x-amz-tagging` to copy without the tagging grant. |
 | `ListObjectVersions` / `ListObjectsV2` with the include-deleted extension | Requires `s3:ListBucketVersions` | Vendor extension | Delete markers and non-current versions are version-listing data. |
-| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits the request budget (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 4096 entries per drive, one entry per 64 plain children); `503 SlowDown` beyond that, deterministically, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions under the prefix and is bounded so a hostile prefix cannot pin the request. Prefixes with more than 262,144 direct children per drive, or roughly 1,400 to 4,096 UUID-named children, exceed the bound (rustfs/backlog#3038 residual). |
+| `DeleteObject` of a missing key that names a prefix | `204 NoContent` when the recovery probe of the prefix fits its per-drive allowance (`DELETE_RECOVERY_MAX_DISCOVERY_ENTRIES`, 262,144 entries: one per UUID-named direct child, one per 64 other names), its per-call time limit (a third of the drive timeout, at most 10 seconds) and the request budget (`DeleteRecoveryWorkBudget`, shared by every key and drive of the request); `503 SlowDown` beyond that, and for the duration of a rolling upgrade while a drive in the set still runs a release without the bounded recovery RPC | `204 NoContent` | The probe looks for interrupted delete transactions of the key and is bounded so a hostile prefix cannot pin the request. A folder completes with up to 262,144 UUID-named direct children or about 16.7 million other direct children per drive when the drive can walk them within the time limit (a warm cache walks roughly 25,000 UUID-named children per second; a cold HDD far fewer); inside a UUID-named child only journal files and data-directory markers are examined, so a child key's own `xl.meta`, data directories or children cost 1/64 each. |
 | `PutObject` / `UploadPart` body delivering less than 64 KiB per 60 seconds of server wait | `400 RequestTimeout` (the minimum-progress floor, logged as `upload_body_progress_stalled`); independent of `RUSTFS_HTTP_REQUEST_BODY_READ_TIMEOUT`, including when that timeout is 0 | No minimum upload rate | A trickle upload otherwise pins upload admission and its buffers indefinitely; roughly 1.1 KiB/s is below any practical client. |
 | First write after `SetBucketQuota`, or after the quota changes, on a bucket without an authoritative usage ledger | `503 SlowDown` (retryable) until a background scan of the bucket publishes the exact ledger; a failed scan retries after 30 seconds | AWS has no bucket quota; MinIO admits with unknown usage | Quota admission uses only the authoritative ledger (scanner reporting is informational, see `reporting_retirement` tests), and the request must not hold its bucket metadata guard across a whole-bucket listing. The window is one bucket listing long. |
 

docs/testing/ci-gates.md

@@ -101,6 +101,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run. The v
 | `coverage.yml` (weekly) | `coverage` | report-only trend; lcov and JSON artifact | yes (disabled) | `make coverage` |
 | `runner-hygiene.yml` (monthly) | `check-ephemerality` | runner ephemerality | yes | dispatch |
 | `e2e-upgrade.yml` (weekly, and on version tags) | `upgrade` (serial cases sharing one harness build) | upgrade and rollback gate; server logs | yes | `python3 scripts/e2e_binary.py build`, then the job's `python3 scripts/e2e_binary.py run -- cargo test --locked -p e2e_test` commands with `RUSTFS_UPGRADE_SOURCE_BINARY` pointing at the pinned previous release (`UPGRADE_SOURCE_VERSION`) |
+| `reference-parity.yml` (nightly, and on PRs that change the lane's own files) | `plan`, `build-candidate`, `s3tests` (one job per target in `.config/reference-parity-targets.toml`), `compare` | report-only parity gate: fails on any per-node-id s3-tests difference between a branch candidate and the release pinned in `.config/reference-binary.toml` that `.config/reference-parity-tolerated.txt` does not list, and on a stale tolerated entry; per-target JUnit, logs and `diff.md`/`diff.json` | yes | `DEPLOY_MODE=binary RUSTFS_BINARY=<binary> MAXFAIL=0 XDIST=4 ./scripts/s3-tests/run.sh` once per target, then `python3 scripts/s3-tests/reference_parity.py compare` as in the `compare` job |
 | `oidc-keycloak.yml` (weekly) | `oidc-keycloak-live` | live OIDC gate | yes | see the PR row |
 | `targets-integration.yml` (nightly) | `targets-live` | live target gate; container logs | yes (disabled) | see the PR row |
 | `gateway-integration-sync.yml` (weekdays) | `sync` | merges `main` into `gateway/integration` as a `--no-ff` merge commit; a conflicting merge is aborted and opens a `sync/<source>-<date>` pull request listing the conflicting files, which keeps the run green; a rejected push or missing pull request fails it | yes | dispatch, optionally with `source_ref` |

rustfs/src/app/bucket_usecase.rs

@@ -3912,16 +3912,36 @@ mod tests {
         req
     }
 
-    fn usecase_method_source<'a>(source: &'a str, method: &str) -> &'a str {
-        let public_start_marker = format!("pub async fn {method}");
-        let private_start_marker = format!("async fn {method}");
-        let (start, marker_len) = if let Some(start) = source.find(&public_start_marker) {
-            (start, public_start_marker.len())
+    /// Bucket usecase source: this file plus `bucket_usecase/`, where a split moves its methods.
+    fn bucket_usecase_sources() -> crate::guard_sources::SourceSet {
+        crate::guard_sources::SourceSet::read("src/app/bucket_usecase.rs", "src/app/bucket_usecase")
+    }
+
+    /// Text after each exact `{marker}(` or `{marker}<` definition, in set order.
+    fn method_definitions<'a>(source: &'a crate::guard_sources::SourceSet, marker: &str) -> Vec<&'a str> {
+        source
+            .files()
+            .iter()
+            .flat_map(|file| {
+                file.text
+                    .match_indices(marker)
+                    .map(move |(start, _)| &file.text[start + marker.len()..])
+                    .filter(|rest| rest.starts_with(['(', '<']))
+            })
+            .collect()
+    }
+
+    /// The body of the one definition of `method` across the bucket usecase set, preferring the
+    /// public form; it ends at the next method of the same indentation or at the end of its file.
+    fn usecase_method_source<'a>(source: &'a crate::guard_sources::SourceSet, method: &str) -> &'a str {
+        let public = method_definitions(source, &format!("pub async fn {method}"));
+        let definitions = if public.is_empty() {
+            method_definitions(source, &format!("async fn {method}"))
         } else {
-            let start = source.find(&private_start_marker).expect("method should exist");
-            (start, private_start_marker.len())
+            public
         };
-        let rest = &source[start + marker_len..];
+        assert_eq!(definitions.len(), 1, "{method} must be defined exactly once across {}", source.describe());
+        let rest = definitions[0];
         let end = [rest.find("\n    pub async fn "), rest.find("\n    async fn ")]
             .into_iter()
             .flatten()
@@ -3932,7 +3952,8 @@ mod tests {
 
     #[test]
     fn bucket_create_and_delete_use_admitted_fresh_tasks() {
-        let source = include_str!("bucket_usecase.rs");
+        let sources = bucket_usecase_sources();
+        let source = &sources;
         for (method, operation, inner_call) in [
             (
                 "execute_create_bucket",
@@ -3957,7 +3978,8 @@ mod tests {
 
     #[test]
     fn bucket_metadata_config_changes_notify_peer_metadata_reload() {
-        let source = include_str!("bucket_usecase.rs");
+        let sources = bucket_usecase_sources();
+        let source = &sources;
         for (method, operation, scanner_maintenance_change) in [
             ("execute_delete_bucket_policy", "delete bucket policy", false),
             ("execute_put_bucket_policy", "put bucket policy", false),

rustfs/src/app/multipart_usecase/abort.rs

@@ -0,0 +1,88 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! AbortMultipartUpload.
+
+use super::*;
+
+impl DefaultMultipartUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    pub async fn execute_abort_multipart_upload(
+        &self,
+        req: S3Request<AbortMultipartUploadInput>,
+    ) -> S3Result<S3Response<AbortMultipartUploadOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        record_s3_op(S3Operation::AbortMultipartUpload);
+        let mut opts = ObjectOptions::default();
+        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
+        let AbortMultipartUploadInput {
+            bucket, key, upload_id, ..
+        } = req.input;
+
+        reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        // Per the AWS S3 API reference, a malformed upload ID is reported as
+        // NoSuchUpload (404), the same as an upload that was aborted or
+        // completed. See multipart_upload_api_error.
+        match store
+            .abort_multipart_upload(bucket.as_str(), key.as_str(), upload_id.as_str(), &opts)
+            .await
+        {
+            Ok(_) => {
+                rustfs_scanner::record_dirty_usage_object_from_producer(
+                    &bucket,
+                    &key,
+                    rustfs_scanner::SegmentInvalidationProducerIdentity::AbortMultipartUpload,
+                );
+                Ok(S3Response::new(AbortMultipartUploadOutput { ..Default::default() }))
+            }
+            Err(err) => Err(multipart_upload_api_error(err)),
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::*;
+    use super::*;
+
+    #[tokio::test]
+    async fn execute_abort_multipart_upload_returns_internal_error_when_store_uninitialized() {
+        let input = AbortMultipartUploadInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .upload_id("upload-id".to_string())
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::DELETE);
+        let before = s3_op_total(S3Operation::AbortMultipartUpload);
+
+        let err = make_usecase().execute_abort_multipart_upload(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        assert_eq!(s3_op_total(S3Operation::AbortMultipartUpload), before + 1);
+    }
+}

rustfs/src/app/multipart_usecase/create.rs

@@ -0,0 +1,418 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! CreateMultipartUpload.
+
+use super::*;
+
+fn create_multipart_upload_metadata(
+    input_metadata: Option<HashMap<String, String>>,
+    headers: &HeaderMap,
+    tagging: Option<String>,
+    storage_class: Option<&crate::app::storage_api::s3::dto::StorageClass>,
+) -> HashMap<String, String> {
+    let mut metadata = input_metadata.unwrap_or_default();
+    namespace_reserved_user_metadata(&mut metadata);
+    extract_metadata_from_mime(headers, &mut metadata);
+
+    if let Some(tags) = tagging {
+        metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags);
+    }
+    if let Some(storage_class) = storage_class {
+        metadata.insert(AMZ_STORAGE_CLASS.to_owned(), storage_class.as_str().to_owned());
+    }
+
+    metadata
+}
+
+/// A multipart session advertises disk compression only when the staged-rollout
+/// switch (`RUSTFS_COMPRESSION_MULTIPART_ENABLED`) is on, the object key/headers
+/// qualify, AND the session is not an SSE-C ciphertext-passthrough replication
+/// session, which must preserve source bytes verbatim.
+///
+/// The rollout switch defaults to off so a rolling upgrade never creates new
+/// compressed multipart objects while pre-fix nodes (whose decompressor is not
+/// resumable) may still serve reads. Enable it once the fleet has converged on a
+/// fixed build; the default flips per the `multipart-compression-default-off-window`
+/// entry in docs/architecture/compat-cleanup-register.md.
+///
+/// Each part is compressed as an independent stream; the GET path decodes across part
+/// boundaries (see `ReadTransform::Compressed`), so the session may advertise
+/// object-level compression again.
+///
+/// Unlike single PUT there is no `MIN_DISK_COMPRESSIBLE_SIZE` floor here: the total
+/// object size is unknown at CreateMultipartUpload time, so tiny multipart objects pay
+/// the (harmless) framing overhead. This is a deliberate trade-off, not a bug.
+fn should_advertise_session_compression(multipart_enabled: bool, ciphertext_passthrough: bool, disk_compressible: bool) -> bool {
+    multipart_enabled && !ciphertext_passthrough && disk_compressible
+}
+
+impl DefaultMultipartUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    #[hotpath::measure(impl_type = "MultipartUsecase")]
+    pub async fn execute_create_multipart_upload(
+        &self,
+        req: S3Request<CreateMultipartUploadInput>,
+    ) -> S3Result<S3Response<CreateMultipartUploadOutput>> {
+        validate_new_object_key(&req.input.key, "CreateMultipartUpload")?;
+        let multipart_max_total_object_size = parse_presigned_multipart_max_total_object_size(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        let helper =
+            OperationHelper::new(&req, EventName::ObjectCreatedCreateMultipartUpload, S3Operation::CreateMultipartUpload)
+                .suppress_event();
+        let replication_authorized = replication_request_authorized(&req);
+        let CreateMultipartUploadInput {
+            bucket,
+            key,
+            tagging,
+            version_id,
+            storage_class,
+            server_side_encryption,
+            sse_customer_algorithm,
+            sse_customer_key,
+            sse_customer_key_md5,
+            ssekms_key_id,
+            object_lock_legal_hold_status,
+            object_lock_mode,
+            object_lock_retain_until_date,
+            metadata: input_metadata,
+            ..
+        } = req.input.clone();
+
+        let server_side_encryption = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
+        let ssekms_key_id = ssekms_key_id.or_else(|| {
+            req.headers
+                .get("x-amz-server-side-encryption-aws-kms-key-id")
+                .and_then(|value| value.to_str().ok())
+                .map(ToOwned::to_owned)
+        });
+
+        // Validate storage class if provided
+        if let Some(ref storage_class) = storage_class
+            && !is_valid_storage_class(storage_class.as_str())
+        {
+            return Err(s3_error!(InvalidStorageClass));
+        }
+
+        reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
+        validate_table_catalog_object_mutation(&bucket, &key).await?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        validate_archive_content_encoding(
+            &key,
+            req.headers.get("content-type").and_then(|value| value.to_str().ok()),
+            req.headers.get("content-encoding").and_then(|value| value.to_str().ok()),
+        )?;
+
+        let mut metadata = create_multipart_upload_metadata(input_metadata, &req.headers, tagging, storage_class.as_ref());
+        if let Some(limit) = multipart_max_total_object_size {
+            insert_str(&mut metadata, SUFFIX_MAX_TOTAL_OBJECT_SIZE, limit.to_string());
+        }
+
+        let has_explicit_object_lock_retention = object_lock_mode.is_some()
+            || object_lock_retain_until_date.is_some()
+            || has_replication_retention_update(&req.headers, replication_authorized);
+        let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
+        if let Some(object_lock_metadata) = build_put_like_object_lock_metadata(
+            &bucket,
+            &object_lock_config_state,
+            object_lock_legal_hold_status,
+            object_lock_mode,
+            object_lock_retain_until_date,
+        )? {
+            metadata.extend(object_lock_metadata);
+        }
+        apply_bucket_default_lock_retention(
+            &bucket,
+            &object_lock_config_state,
+            &mut metadata,
+            has_explicit_object_lock_retention,
+        )?;
+        let (header_sse_customer_algorithm, header_sse_customer_key, header_sse_customer_key_md5) =
+            extract_ssec_params_from_headers(&req.headers)?;
+        let sse_customer_algorithm = sse_customer_algorithm.or(header_sse_customer_algorithm);
+        let sse_customer_key = sse_customer_key.or(header_sse_customer_key);
+        let sse_customer_key_md5 = sse_customer_key_md5.or(header_sse_customer_key_md5);
+
+        // The session data key is generated here, so this is where a multipart upload is held
+        // to the KMS key it names. Parts and completion also authorize their own caller.
+        let session_principal = SseKmsPrincipal::from_request(&req);
+        let encryption_request = PrepareEncryptionRequest {
+            bucket: &bucket,
+            key: &key,
+            server_side_encryption,
+            ssekms_key_id,
+            ssekms_context: extract_ssekms_context_from_headers(&req.headers)?,
+            sse_customer_algorithm: sse_customer_algorithm.clone(),
+            sse_customer_key,
+            sse_customer_key_md5: sse_customer_key_md5.clone(),
+            principal: session_principal.as_ref(),
+        };
+
+        // SSE-C ciphertext passthrough: parts are already encrypted, so no
+        // session DEK is prepared; a session marker tells UploadPart to store
+        // the ciphertext verbatim instead of recovering encryption material.
+        let ciphertext_passthrough = replication_authorized
+            && get_header(&req.headers, SUFFIX_SOURCE_REPLICATION_REQUEST).as_deref() == Some("true")
+            && rustfs_utils::http::ssec_transport_to_stored_metadata(&req.headers).is_some();
+        if ciphertext_passthrough && let Some(metadata_sys) = self.bucket_metadata_sys() {
+            let check_result = map_quota_check_outcome(
+                &bucket,
+                QuotaChecker::new(metadata_sys)
+                    .check_quota(&bucket, QuotaOperation::PutObject, 0)
+                    .await,
+            )?;
+            if check_result.quota_limit.is_some() {
+                return Err(S3Error::with_message(
+                    S3ErrorCode::InvalidRequest,
+                    "SSE-C ciphertext replication is unavailable for quota-enabled buckets".to_string(),
+                ));
+            }
+        }
+        if ciphertext_passthrough {
+            insert_str(&mut metadata, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT, "true".to_string());
+        }
+
+        let prepared_material = if ciphertext_passthrough {
+            None
+        } else {
+            sse_prepare_encryption(encryption_request).await?
+        };
+        let (effective_sse, effective_kms_key_id) = match prepared_material {
+            Some(material) => {
+                let server_side_encryption = Some(material.server_side_encryption.clone());
+                let ssekms_key_id = material.response_kms_key_id();
+
+                let mut encryption_metadata = encryption_material_to_metadata(&material)?;
+                if material.key_kind == EncryptionKeyKind::Object {
+                    mark_encrypted_multipart_metadata(&mut encryption_metadata);
+                }
+                metadata.extend(encryption_metadata);
+
+                (server_side_encryption, ssekms_key_id)
+            }
+            None => (None, None),
+        };
+
+        if should_advertise_session_compression(
+            is_multipart_disk_compression_enabled(),
+            ciphertext_passthrough,
+            is_disk_compressible(&req.headers, &key),
+        ) {
+            rustfs_utils::http::insert_str(
+                &mut metadata,
+                rustfs_utils::http::SUFFIX_COMPRESSION,
+                compression_metadata_value(CompressionAlgorithm::default()),
+            );
+        }
+
+        let mt2 = metadata.clone();
+        let mut opts: ObjectOptions =
+            put_opts_with_replication_authorization(&bucket, &key, version_id, &req.headers, metadata, replication_authorized)
+                .await
+                .map_err(ApiError::from)?;
+        apply_bucket_generation_guard(&req, &bucket, &mut opts)?;
+
+        let dsc =
+            must_replicate_object(&bucket, &key, &mt2, "".to_string(), opts.delete_marker_replication_status(), opts.clone())
+                .await;
+        if dsc.replicate_any() {
+            insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_GENERATION, Uuid::new_v4().to_string());
+            insert_str(&mut opts.user_defined, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string());
+            insert_str(
+                &mut opts.user_defined,
+                SUFFIX_REPLICATION_STATUS,
+                dsc.pending_status().unwrap_or_default(),
+            );
+        }
+
+        if !super::super::object::shared::put_like_write_creates_new_version(&opts)
+            && super::super::object::shared::object_lock_checks_required_for_state(&object_lock_config_state)
+        {
+            let current_opts: ObjectOptions = internal_object_info_lookup_opts(
+                get_opts(&bucket, &key, opts.version_id.clone(), None, &req.headers)
+                    .await
+                    .map_err(ApiError::from)?,
+            );
+            match store.get_object_info(&bucket, &key, &current_opts).await {
+                Ok(existing_obj_info) => {
+                    validate_existing_object_lock_for_write(&object_lock_config_state, &existing_obj_info, &opts)?
+                }
+                Err(err) => {
+                    if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
+                        return Err(ApiError::from(err).into());
+                    }
+                }
+            }
+        }
+
+        let checksum_type = rustfs_rio::ChecksumType::from_header(&req.headers);
+        if checksum_type.is(rustfs_rio::ChecksumType::INVALID) {
+            return Err(s3_error!(InvalidArgument, "Invalid checksum type"));
+        } else if checksum_type.is_set() && !checksum_type.is(rustfs_rio::ChecksumType::TRAILING) {
+            opts.want_checksum = Some(rustfs_rio::Checksum {
+                checksum_type,
+                ..Default::default()
+            });
+        }
+        if effective_sse.is_some() && opts.want_checksum.is_some() {
+            insert_str(&mut opts.user_defined, SUFFIX_PLAINTEXT_CHECKSUM, "true".to_string());
+        }
+
+        let MultipartUploadResult {
+            upload_id,
+            checksum_algo,
+            checksum_type,
+        } = store
+            .new_multipart_upload(&bucket, &key, &opts)
+            .await
+            .map_err(ApiError::from)?;
+
+        let output = CreateMultipartUploadOutput {
+            bucket: Some(bucket),
+            key: Some(key),
+            upload_id: Some(upload_id),
+            server_side_encryption: effective_sse,
+            sse_customer_algorithm,
+            sse_customer_key_md5,
+            ssekms_key_id: effective_kms_key_id,
+            checksum_algorithm: checksum_algo.map(ChecksumAlgorithm::from),
+            checksum_type: checksum_type.map(ChecksumType::from),
+            ..Default::default()
+        };
+
+        let result = Ok(S3Response::new(output));
+        let _ = helper.complete(&result);
+        result
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::*;
+    use super::*;
+
+    #[test]
+    fn session_compression_is_advertised_only_for_non_passthrough_compressible_uploads() {
+        // (multipart_enabled, ciphertext_passthrough, disk_compressible, expected)
+        let cases = [
+            (true, false, false, false),
+            (true, false, true, true),
+            (true, true, false, false),
+            (true, true, true, false),
+            // The staged-rollout switch keeps multipart compression dark by
+            // default regardless of the other gates.
+            (false, false, true, false),
+            (false, false, false, false),
+            (false, true, true, false),
+            (false, true, false, false),
+        ];
+
+        for (multipart_enabled, ciphertext_passthrough, disk_compressible, expected) in cases {
+            assert_eq!(
+                should_advertise_session_compression(multipart_enabled, ciphertext_passthrough, disk_compressible),
+                expected,
+                "multipart_enabled={multipart_enabled} ciphertext_passthrough={ciphertext_passthrough} disk_compressible={disk_compressible}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn execute_create_multipart_upload_rejects_invalid_storage_class() {
+        let input = CreateMultipartUploadInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .storage_class(Some(StorageClass::from("invalid".to_string())))
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::POST);
+
+        let err = make_usecase().execute_create_multipart_upload(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidStorageClass);
+    }
+
+    #[test]
+    fn create_multipart_upload_metadata_persists_dto_storage_class_without_raw_header() {
+        let metadata = create_multipart_upload_metadata(
+            None,
+            &HeaderMap::new(),
+            None,
+            Some(&StorageClass::from_static("REDUCED_REDUNDANCY")),
+        );
+
+        assert_eq!(metadata.get(AMZ_STORAGE_CLASS), Some(&"REDUCED_REDUNDANCY".to_string()));
+    }
+
+    #[test]
+    fn create_multipart_upload_metadata_keeps_user_and_system_namespaces_separate() {
+        let mut headers = HeaderMap::new();
+        headers.insert("content-type", HeaderValue::from_static("application/octet-stream"));
+        headers.insert(AMZ_STORAGE_CLASS, HeaderValue::from_static("STANDARD"));
+        let input_metadata = HashMap::from([
+            ("content-type".to_string(), "user-content-type".to_string()),
+            (AMZ_STORAGE_CLASS.to_string(), "user-storage-class".to_string()),
+        ]);
+
+        let metadata = create_multipart_upload_metadata(
+            Some(input_metadata),
+            &headers,
+            Some("project=rustfs".to_string()),
+            Some(&StorageClass::from_static("REDUCED_REDUNDANCY")),
+        );
+
+        assert_eq!(metadata.get("content-type"), Some(&"application/octet-stream".to_string()));
+        assert_eq!(metadata.get("x-amz-meta-content-type"), Some(&"user-content-type".to_string()));
+        assert_eq!(metadata.get(AMZ_STORAGE_CLASS), Some(&"REDUCED_REDUNDANCY".to_string()));
+        assert_eq!(metadata.get("x-amz-meta-x-amz-storage-class"), Some(&"user-storage-class".to_string()));
+        assert_eq!(metadata.get(AMZ_OBJECT_TAGGING), Some(&"project=rustfs".to_string()));
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn create_multipart_rejects_ciphertext_replication_before_parts_are_staged() {
+        let (_store, bucket) = crate::app::gating_test_env::durable_quota_test_bucket("ciphertext-multipart-quota", 4096).await;
+        let usecase = DefaultMultipartUsecase::from_global();
+        let input = CreateMultipartUploadInput::builder()
+            .bucket(bucket)
+            .key("object".to_string())
+            .build()
+            .expect("create multipart request should build");
+        let mut request = build_request(input, Method::POST);
+        insert_header(&mut request.headers, SUFFIX_SOURCE_REPLICATION_REQUEST, "true");
+        request
+            .headers
+            .insert(rustfs_utils::http::REPLICATION_SSEC_ALGORITHM_HEADER, HeaderValue::from_static("AES256"));
+        request.extensions.insert(crate::storage::access::ReqInfo {
+            replication_request_authorized: true,
+            ..Default::default()
+        });
+
+        let err = usecase
+            .execute_create_multipart_upload(request)
+            .await
+            .expect_err("quota-enabled ciphertext multipart replication should fail before upload creation");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+}

rustfs/src/app/multipart_usecase/list.rs

@@ -0,0 +1,235 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! ListMultipartUploads and ListParts.
+
+use super::*;
+
+impl DefaultMultipartUsecase {
+    pub async fn execute_list_multipart_uploads(
+        &self,
+        req: S3Request<ListMultipartUploadsInput>,
+    ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
+        let table_list_access = req.extensions.get::<TableDataPlaneListAccess>().cloned();
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        let mut opts = ObjectOptions::default();
+        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
+        let ListMultipartUploadsInput {
+            bucket,
+            prefix,
+            delimiter,
+            key_marker,
+            upload_id_marker,
+            max_uploads,
+            encoding_type,
+            ..
+        } = req.input;
+
+        let ListMultipartUploadsParams {
+            prefix,
+            key_marker,
+            max_uploads,
+        } = parse_list_multipart_uploads_params(prefix, key_marker, max_uploads)?;
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        // `apply_bucket_generation_guard` tolerates a missing guard (only the S3
+        // access layer installs one), so resolve the current generation rather
+        // than failing the request. Listing is filtered by this value, so a
+        // stale one simply hides foreign-incarnation uploads, as intended.
+        let expected_incarnation_id = match opts.expected_bucket_incarnation_id {
+            Some(incarnation_id) => incarnation_id,
+            None => store.bucket_incarnation_id_from_disk(&bucket).await.map_err(ApiError::from)?,
+        };
+
+        let result = match table_list_access.as_ref() {
+            Some(access) => {
+                table_list_isolation::list_multipart_uploads(
+                    store.clone(),
+                    access,
+                    table_list_isolation::ListMultipartUploadsRequest {
+                        bucket: &bucket,
+                        prefix: &prefix,
+                        key_marker: key_marker.as_deref(),
+                        upload_id_marker: upload_id_marker.as_deref(),
+                        delimiter: delimiter.as_deref(),
+                        max_uploads,
+                        expected_incarnation_id,
+                    },
+                )
+                .await?
+            }
+            None => store
+                .list_multipart_uploads_for_bucket_incarnation(
+                    &bucket,
+                    &prefix,
+                    key_marker,
+                    upload_id_marker,
+                    delimiter,
+                    max_uploads,
+                    expected_incarnation_id,
+                )
+                .await
+                .map_err(ApiError::from)?,
+        };
+
+        Ok(S3Response::new(build_list_multipart_uploads_output(
+            bucket,
+            prefix,
+            result,
+            encoding_type.as_ref(),
+        )))
+    }
+
+    pub async fn execute_list_parts(&self, req: S3Request<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        let mut opts = ObjectOptions::default();
+        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
+        let ListPartsInput {
+            bucket,
+            key,
+            upload_id,
+            part_number_marker,
+            max_parts,
+            ..
+        } = req.input;
+
+        reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
+        let params = parse_list_parts_params(part_number_marker, max_parts)?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        let res = store
+            .list_object_parts(&bucket, &key, &upload_id, params.part_number_marker, params.max_parts, &opts)
+            .await
+            .map_err(multipart_upload_api_error)?;
+
+        Ok(S3Response::new(build_list_parts_output(res)))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::*;
+    use super::*;
+
+    #[tokio::test]
+    async fn execute_list_multipart_uploads_returns_internal_error_when_store_uninitialized() {
+        let input = ListMultipartUploadsInput::builder()
+            .bucket("bucket".to_string())
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+
+        let err = make_usecase().execute_list_multipart_uploads(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_list_multipart_uploads_rejects_invalid_key_marker_before_store_lookup() {
+        let input = ListMultipartUploadsInput::builder()
+            .bucket("bucket".to_string())
+            .prefix(Some("prefix/".to_string()))
+            .key_marker(Some("other/key".to_string()))
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+
+        let err = make_usecase().execute_list_multipart_uploads(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        assert_eq!(err.message(), Some("Invalid key marker"));
+    }
+
+    #[tokio::test]
+    async fn execute_list_multipart_uploads_rejects_invalid_max_uploads_before_store_lookup() {
+        let input = ListMultipartUploadsInput::builder()
+            .bucket("bucket".to_string())
+            .max_uploads(Some(0))
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+        let expected = "max-uploads must be between 1 and 1000";
+
+        let err = make_usecase().execute_list_multipart_uploads(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        assert_eq!(err.message(), Some(expected));
+    }
+
+    #[tokio::test]
+    async fn execute_list_parts_returns_internal_error_when_store_uninitialized() {
+        let input = ListPartsInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .upload_id("upload-id".to_string())
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+
+        let err = make_usecase().execute_list_parts(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_list_parts_rejects_negative_part_number_marker_before_store_lookup() {
+        let input = ListPartsInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .upload_id("upload-id".to_string())
+            .part_number_marker(Some(-1))
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+
+        let err = make_usecase().execute_list_parts(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        assert_eq!(err.message(), Some("part-number-marker must be non-negative"));
+    }
+
+    #[tokio::test]
+    async fn execute_list_parts_rejects_invalid_max_parts_before_store_lookup() {
+        let input = ListPartsInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .upload_id("upload-id".to_string())
+            .max_parts(Some(1001))
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::GET);
+
+        let err = make_usecase().execute_list_parts(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+        assert_eq!(err.message(), Some("max-parts must be between 1 and 1000"));
+    }
+}

rustfs/src/app/multipart_usecase/test_support.rs

@@ -0,0 +1,54 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Fixtures and test-only imports shared by the multipart use-case tests.
+
+use super::*;
+pub(super) use crate::app::storage_api::s3::dto::CompletedMultipartUpload;
+pub(super) use crate::app::storage_api::s3::dto::StorageClass;
+pub(super) use http::Extensions;
+pub(super) use http::Method;
+pub(super) use rustfs_filemeta::ObjectPartInfo;
+pub(super) use rustfs_utils::http::AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER;
+pub(super) use rustfs_utils::http::AMZ_OBJECT_LOCK_MODE_LOWER;
+pub(super) use rustfs_utils::http::AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER;
+pub(super) use std::io::Cursor;
+pub(super) use temp_env::async_with_vars;
+pub(super) use tokio::io::AsyncReadExt;
+
+pub(super) fn s3_op_total(op: S3Operation) -> u64 {
+    rustfs_io_metrics::s3_op_metrics_snapshot()
+        .into_iter()
+        .find(|snapshot| snapshot.op == op.as_str())
+        .map(|snapshot| snapshot.total)
+        .unwrap_or_default()
+}
+
+pub(super) fn build_request<T>(input: T, method: Method) -> S3Request<T> {
+    S3Request {
+        input,
+        method,
+        uri: Uri::from_static("/"),
+        headers: HeaderMap::new(),
+        extensions: Extensions::new(),
+        credentials: None,
+        region: None,
+        service: None,
+        trailing_headers: None,
+    }
+}
+
+pub(super) fn make_usecase() -> DefaultMultipartUsecase {
+    DefaultMultipartUsecase::without_context()
+}

rustfs/src/app/multipart_usecase/upload_part.rs

@@ -0,0 +1,823 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! UploadPart.
+
+use super::*;
+
+fn resolve_upload_part_size(content_length: Option<i64>, body: Option<&StreamingBlob>) -> rustfs_s3_types::S3Result<Option<i64>> {
+    if let Some(length) = content_length {
+        return Ok(Some(length));
+    }
+    body.and_then(|body| body.remaining_length().exact())
+        .map(i64::try_from)
+        .transpose()
+        .map_err(|_| rustfs_s3_types::s3_error!(UnexpectedContent))
+}
+
+fn require_upload_part_size(size: Option<i64>, capped: bool) -> rustfs_s3_types::S3Result<i64> {
+    match size {
+        Some(size) if size >= 0 => Ok(size),
+        Some(_) => Err(rustfs_s3_types::s3_error!(UnexpectedContent)),
+        None if capped => Err(rustfs_s3_types::s3_error!(UnexpectedContent)),
+        None => Err(rustfs_s3_types::S3Error::new(rustfs_s3_types::S3ErrorCode::MissingContentLength)),
+    }
+}
+
+fn upload_part_body_read_timeout(configured: Duration, capped: bool) -> Duration {
+    if capped {
+        configured.max(Duration::from_secs(rustfs_config::DEFAULT_HTTP_REQUEST_BODY_READ_TIMEOUT))
+    } else {
+        configured
+    }
+}
+
+/// Plaintext length of one SSE-C passthrough part, declared by the sender on
+/// UploadPart (backlog#2363).
+fn passthrough_part_actual_size(headers: &HeaderMap) -> Option<i64> {
+    get_header(headers, rustfs_utils::http::SUFFIX_REPLICATION_PART_ACTUAL_SIZE)?
+        .parse::<i64>()
+        .ok()
+        .filter(|size| *size > 0)
+}
+
+// Hoisted values are signed query parameters, not HTTP headers. Normalize only
+// after access control has verified the presigned request.
+fn normalize_presigned_part_checksums(req: &mut S3Request<UploadPartInput>) -> S3Result<()> {
+    if req.extensions.get::<VerifiedPresignedRequest>().is_none() {
+        return Ok(());
+    }
+    let Some(query) = req.uri.query() else {
+        return Ok(());
+    };
+    let mut seen = std::collections::HashSet::new();
+    for (name, value) in url::form_urlencoded::parse(query.as_bytes()) {
+        let field = match name.as_ref() {
+            "x-amz-checksum-crc32" => Some(&mut req.input.checksum_crc32),
+            "x-amz-checksum-crc32c" => Some(&mut req.input.checksum_crc32c),
+            "x-amz-checksum-crc64nvme" => Some(&mut req.input.checksum_crc64nvme),
+            "x-amz-checksum-sha1" => Some(&mut req.input.checksum_sha1),
+            "x-amz-checksum-sha256" => Some(&mut req.input.checksum_sha256),
+            "x-amz-sdk-checksum-algorithm" => None,
+            _ => continue,
+        };
+        if !seen.insert(name.clone()) {
+            return Err(S3Error::with_message(S3ErrorCode::InvalidRequest, "Duplicate checksum query parameter"));
+        }
+        if let Some(header) = req.headers.get(name.as_ref())
+            && header.as_bytes() != value.as_bytes()
+        {
+            return Err(S3Error::with_message(
+                S3ErrorCode::InvalidRequest,
+                "Conflicting checksum header and query parameter",
+            ));
+        }
+        let header = http::HeaderValue::from_str(&value)
+            .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid checksum query parameter"))?;
+        req.headers.insert(
+            http::header::HeaderName::from_bytes(name.as_bytes())
+                .map_err(|_| S3Error::with_message(S3ErrorCode::InvalidArgument, "Invalid checksum query parameter"))?,
+            header,
+        );
+        if let Some(field) = field {
+            *field = Some(value.into_owned());
+        } else {
+            req.input.checksum_algorithm = Some(ChecksumAlgorithm::from(value.into_owned()));
+        }
+    }
+    Ok(())
+}
+
+impl DefaultMultipartUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    #[hotpath::measure(impl_type = "MultipartUsecase")]
+    pub async fn execute_upload_part(&self, mut req: S3Request<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        normalize_presigned_part_checksums(&mut req)?;
+        let mut opts = ObjectOptions::default();
+        apply_bucket_generation_guard(&req, &req.input.bucket, &mut opts)?;
+        let part_principal = SseKmsPrincipal::from_request(&req);
+        let input = req.input;
+        let UploadPartInput {
+            body,
+            bucket,
+            key,
+            upload_id,
+            part_number,
+            content_length,
+            sse_customer_algorithm,
+            sse_customer_key,
+            sse_customer_key_md5,
+            // content_md5,
+            ..
+        } = input;
+
+        let part_id = parse_upload_part_number(part_number)?;
+
+        reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
+        validate_table_catalog_object_mutation(&bucket, &key).await?;
+
+        let size = resolve_upload_part_size(content_length, body.as_ref())?;
+        if let Some(size) = size {
+            reject_oversize_single_upload(size)?;
+        }
+        let body_stream = body.ok_or_else(|| s3_error!(IncompleteBody))?;
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+        let fi = store
+            .get_multipart_info(&bucket, &key, &upload_id, &opts)
+            .await
+            .map_err(multipart_upload_api_error)?;
+        authorize_sse_kms_multipart_write(part_principal.as_ref(), &fi.user_defined).await?;
+        let max_total_object_size = multipart_max_total_object_size(&fi.user_defined)?;
+        let mut size = require_upload_part_size(size, max_total_object_size.is_some())?;
+        if let Some(limit) = max_total_object_size
+            && u64::try_from(size).is_ok_and(|size| size > limit)
+        {
+            return Err(S3Error::new(S3ErrorCode::EntityTooLarge));
+        }
+        let admission_wait_started = rustfs_io_metrics::put_stage_timer();
+        let admission = self.concurrency_manager().admit_multipart_part(size).await;
+        rustfs_io_metrics::record_put_object_stage_duration_from("multipart_admission_wait", admission_wait_started);
+        let upload_part_admission = match admission
+            .map_err(|_| S3Error::with_message(S3ErrorCode::InternalError, "foreground write admission closed"))?
+        {
+            ForegroundWriteAdmission::Disabled => None,
+            ForegroundWriteAdmission::Admitted(permit) => {
+                counter!("rustfs.upload_part.foreground_admission.total", "result" => "admitted").increment(1);
+                Some(permit)
+            }
+            ForegroundWriteAdmission::Rejected => {
+                counter!("rustfs.upload_part.foreground_admission.total", "result" => "rejected").increment(1);
+                return Err(S3Error::with_message(
+                    S3ErrorCode::SlowDown,
+                    "foreground write concurrency limit reached, please reduce your request rate",
+                ));
+            }
+        };
+        let request_id = req
+            .extensions
+            .get::<super::super::storage_api::multipart_usecase::request_context::RequestContext>()
+            .map(|ctx| ctx.request_id.as_str())
+            .unwrap_or_default();
+        let timeout = upload_part_body_read_timeout(put_object_body_read_timeout(), max_total_object_size.is_some());
+        let (body_stream, progress_control) =
+            if let Some(control) = req.extensions.get::<super::super::object::request_body::BodyReadControl>() {
+                control.activate_upload(timeout, &bucket, &key, request_id, size.max(0) as u64);
+                (body_stream, control.clone())
+            } else {
+                super::super::object::request_body::guard_upload_progress(
+                    body_stream,
+                    timeout,
+                    &bucket,
+                    &key,
+                    request_id,
+                    size.max(0) as u64,
+                )
+            };
+
+        let ingress_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(std::time::Instant::now);
+
+        // Apply adaptive buffer sizing based on part size for optimal streaming performance.
+        // Uses workload profile configuration (enabled by default) to select appropriate buffer size.
+        // Buffer sizes range from 32KB to 4MB depending on part size and configured workload profile.
+        let buffer_size = get_buffer_size_opt_in(size);
+        let body = tokio::io::BufReader::with_capacity(
+            buffer_size,
+            StreamReader::new(body_stream.map(|f| f.map_err(s3s_body_error_to_io))),
+        );
+
+        // An SSE-C passthrough session stores ciphertext parts verbatim: a
+        // compression key restored on the session describes those stored
+        // bytes and must not add a second compression layer, and each part's
+        // plaintext length comes from the sender (backlog#2363).
+        let preserve_ciphertext = contains_key_str(&fi.user_defined, SUFFIX_REPLICATION_PRESERVE_CIPHERTEXT);
+        let is_disk_compressed = !preserve_ciphertext
+            && rustfs_utils::http::contains_key_str(&fi.user_defined, rustfs_utils::http::SUFFIX_COMPRESSION);
+
+        let actual_size = if preserve_ciphertext {
+            passthrough_part_actual_size(&req.headers).unwrap_or(size)
+        } else {
+            size
+        };
+
+        let mut md5hex = if let Some(base64_md5) = input.content_md5 {
+            let md5 = base64_simd::STANDARD
+                .decode_to_vec(base64_md5.as_bytes())
+                .map_err(|e| ApiError::from(StorageError::other(format!("Invalid content MD5: {e}"))))?;
+            Some(hex_simd::encode_to_string(&md5, hex_simd::AsciiCase::Lower))
+        } else {
+            None
+        };
+
+        let mut sha256hex = get_content_sha256_with_query(&req.headers, req.uri.query());
+
+        let mut write_plan = WritePlan::new();
+        let mut reader = if is_disk_compressed {
+            let algorithm = CompressionAlgorithm::default();
+            let mut hrd = HashReader::from_stream(body, size, actual_size, md5hex.take(), sha256hex.take(), false)
+                .map_err(ApiError::from)?;
+
+            if let Err(err) = hrd.add_checksum(&req.headers, trailer_source(req.trailing_headers.clone()), false) {
+                return Err(ApiError::from(err).into());
+            }
+
+            size = HashReader::SIZE_PRESERVE_LAYER;
+            write_plan = write_plan.with_compression(algorithm);
+            hrd
+        } else {
+            HashReader::from_stream(body, size, actual_size, md5hex, sha256hex, false).map_err(ApiError::from)?
+        };
+
+        if let Err(err) = reader.add_checksum(&req.headers, trailer_source(req.trailing_headers.clone()), size < 0) {
+            return Err(ApiError::from(err).into());
+        }
+        opts.want_checksum = reader.checksum();
+
+        // An SSE-C passthrough session stores ciphertext parts verbatim: no
+        // material recovery, no validation against the (absent) customer key.
+        let has_ssec = !preserve_ciphertext
+            && fi
+                .user_defined
+                .contains_key("x-amz-server-side-encryption-customer-algorithm");
+        let (server_side_encryption, ssekms_key_id) = if has_ssec || preserve_ciphertext {
+            (None, None)
+        } else {
+            let sse = fi
+                .user_defined
+                .get("x-amz-server-side-encryption")
+                .map(|s| {
+                    ServerSideEncryption::from_str(s)
+                        .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
+                })
+                .transpose()?;
+            let key_id = match sse.as_ref() {
+                Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS => fi
+                    .user_defined
+                    .get("x-amz-server-side-encryption-aws-kms-key-id")
+                    .map(|s| s.to_string()),
+                _ => None,
+            };
+            (sse, key_id)
+        };
+        if !preserve_ciphertext {
+            EncryptionRequest {
+                bucket: &bucket,
+                key: &key,
+                server_side_encryption: server_side_encryption.clone(),
+                ssekms_key_id: ssekms_key_id.clone(),
+                ssekms_context: None,
+                sse_customer_algorithm: sse_customer_algorithm.clone(),
+                sse_customer_key: sse_customer_key.clone(),
+                sse_customer_key_md5: sse_customer_key_md5.clone(),
+                content_size: actual_size,
+                principal: None,
+            }
+            .validate_multipart_ssec(&fi.user_defined)?;
+        }
+        let (requested_sse, requested_kms_key_id) = if has_ssec {
+            let ssec_material = sse_decryption(DecryptionRequest {
+                bucket: &bucket,
+                key: &key,
+                metadata: &fi.user_defined,
+                sse_customer_key: sse_customer_key.as_ref(),
+                sse_customer_key_md5: sse_customer_key_md5.as_ref(),
+                principal: None,
+            })
+            .await?
+            .ok_or_else(|| ApiError::from(StorageError::other("Missing SSE-C session material")))?;
+            let ssec_write = match ssec_material.key_kind {
+                EncryptionKeyKind::Object => WriteEncryption::multipart_object_key(ssec_material.key_bytes, part_id as u32),
+                EncryptionKeyKind::Direct => {
+                    WriteEncryption::multipart(ssec_material.key_bytes, ssec_material.base_nonce, part_id)
+                }
+            };
+            write_plan = write_plan.with_encryption(ssec_write);
+            (Some(ssec_material.server_side_encryption), ssec_material.kms_key_id)
+        } else if let Some(server_side_encryption) = server_side_encryption {
+            // Reuse the stored envelope after authorizing this caller against its
+            // pinned destination key above.
+            let managed_material = sse_decryption(DecryptionRequest {
+                bucket: &bucket,
+                key: &key,
+                metadata: &fi.user_defined,
+                sse_customer_key: None,
+                sse_customer_key_md5: None,
+                principal: None,
+            })
+            .await?
+            .ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?;
+            let managed_write = match managed_material.key_kind {
+                EncryptionKeyKind::Object => WriteEncryption::multipart_object_key(managed_material.key_bytes, part_id as u32),
+                EncryptionKeyKind::Direct => {
+                    WriteEncryption::multipart(managed_material.key_bytes, managed_material.base_nonce, part_id)
+                }
+            };
+            write_plan = write_plan.with_encryption(managed_write);
+            (Some(server_side_encryption), ssekms_key_id)
+        } else {
+            (None, None)
+        };
+
+        reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
+
+        {
+            use rustfs_rio::HashReaderMut;
+            let inner = reader.take_inner();
+            reader.inner =
+                rustfs_rio::boxed_reader(super::super::object::request_body::DemandReader::new(inner, progress_control));
+        }
+
+        let mut reader = PutObjReader::new(reader);
+
+        if let Some(stage_start) = ingress_stage_start {
+            rustfs_io_metrics::record_put_object_stage_duration(
+                "multipart_ingress_prepare",
+                stage_start.elapsed().as_secs_f64() * 1000.0,
+            );
+        }
+
+        let _upload_part_admission = upload_part_admission;
+        let store_write_started = rustfs_io_metrics::put_stage_timer();
+        let result = store
+            .put_object_part(&bucket, &key, &upload_id, part_id, &mut reader, &opts)
+            .await;
+        rustfs_io_metrics::record_put_object_stage_duration_from("multipart_store_write", store_write_started);
+        drop(_upload_part_admission);
+        let info = result.map_err(ApiError::from)?;
+
+        let mut checksum_crc32 = input.checksum_crc32;
+        let mut checksum_crc32c = input.checksum_crc32c;
+        let mut checksum_sha1 = input.checksum_sha1;
+        let mut checksum_sha256 = input.checksum_sha256;
+        let mut checksum_crc64nvme = input.checksum_crc64nvme;
+
+        if let Some(alg) = &input.checksum_algorithm
+            && let Some(Some(checksum_str)) = req.trailing_headers.as_ref().map(|trailer| {
+                let key = match alg.as_str() {
+                    ChecksumAlgorithm::CRC32 => rustfs_rio::ChecksumType::CRC32.key(),
+                    ChecksumAlgorithm::CRC32C => rustfs_rio::ChecksumType::CRC32C.key(),
+                    ChecksumAlgorithm::SHA1 => rustfs_rio::ChecksumType::SHA1.key(),
+                    ChecksumAlgorithm::SHA256 => rustfs_rio::ChecksumType::SHA256.key(),
+                    ChecksumAlgorithm::CRC64NVME => rustfs_rio::ChecksumType::CRC64_NVME.key(),
+                    _ => return None,
+                };
+                trailer.read(|headers| {
+                    headers
+                        .get(key.unwrap_or_default())
+                        .and_then(|value| value.to_str().ok().map(|s| s.to_string()))
+                })
+            })
+        {
+            match alg.as_str() {
+                ChecksumAlgorithm::CRC32 => checksum_crc32 = checksum_str,
+                ChecksumAlgorithm::CRC32C => checksum_crc32c = checksum_str,
+                ChecksumAlgorithm::SHA1 => checksum_sha1 = checksum_str,
+                ChecksumAlgorithm::SHA256 => checksum_sha256 = checksum_str,
+                ChecksumAlgorithm::CRC64NVME => checksum_crc64nvme = checksum_str,
+                _ => (),
+            }
+        }
+
+        // XXHash3/64/128 and SHA-512 have no typed UploadPartOutput field; echo the
+        // server-computed part checksum as a raw response header (#1261).
+        let upload_part_extra_checksum_headers = crate::app::object_usecase::additional_checksum_echo_pairs(&opts.want_checksum);
+
+        let output = UploadPartOutput {
+            server_side_encryption: requested_sse,
+            ssekms_key_id: requested_kms_key_id,
+            sse_customer_algorithm,
+            sse_customer_key_md5,
+            checksum_crc32,
+            checksum_crc32c,
+            checksum_sha1,
+            checksum_sha256,
+            checksum_crc64nvme,
+            e_tag: info.etag.map(|etag| to_s3s_etag(&etag)),
+            ..Default::default()
+        };
+
+        let mut response = S3Response::new(output);
+        crate::app::object_usecase::inject_additional_checksum_headers(
+            &mut response.headers,
+            &upload_part_extra_checksum_headers,
+        );
+        Ok(response)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::*;
+    use super::*;
+
+    fn presigned_checksum_request(query: &str) -> S3Request<UploadPartInput> {
+        let mut req = build_request(UploadPartInput::default(), Method::PUT);
+        req.uri = format!("/bucket/object?{query}").parse().unwrap();
+        req.extensions.insert(VerifiedPresignedRequest);
+        req
+    }
+
+    #[tokio::test]
+    async fn normalize_presigned_part_checksums_validates_body() {
+        // SHA256("abc"), including percent-encoded base64 punctuation.
+        let checksum = "ungWv48Bz+pBQUDeXa4iI7ADYaOWF3qctBD/YfIAFa0=";
+        let query =
+            "x-amz-checksum-sha256=ungWv48Bz%2BpBQUDeXa4iI7ADYaOWF3qctBD%2FYfIAFa0%3D&x-amz-sdk-checksum-algorithm=SHA256";
+        for payload in [b"abc", b"abd"] {
+            let mut req = presigned_checksum_request(query);
+            normalize_presigned_part_checksums(&mut req).unwrap();
+            assert_eq!(req.input.checksum_sha256.as_deref(), Some(checksum));
+            assert_eq!(
+                req.input.checksum_algorithm.as_ref().map(|algorithm| algorithm.as_str()),
+                Some(ChecksumAlgorithm::SHA256)
+            );
+            let mut reader = HashReader::from_stream(Cursor::new(payload), 3, 3, None, None, false).unwrap();
+            reader.add_checksum(&req.headers, None, false).unwrap();
+            assert_eq!(reader.content_crc_type(), Some(rustfs_rio::ChecksumType::SHA256));
+            let mut bytes = Vec::new();
+            let result = reader.read_to_end(&mut bytes).await;
+            if payload == b"abc" {
+                result.unwrap();
+                assert_eq!(bytes, payload);
+                assert_eq!(reader.content_crc().get("SHA256").map(String::as_str), Some(checksum));
+            } else {
+                let err = S3Error::from(ApiError::from(result.unwrap_err()));
+                assert_eq!(*err.code(), S3ErrorCode::BadDigest);
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn normalize_presigned_part_checksums_accepts_sdk_crc32_default() {
+        let mut req = presigned_checksum_request("x-amz-checksum-crc32=y%2FQ5Jg%3D%3D&x-amz-sdk-checksum-algorithm=CRC32");
+        normalize_presigned_part_checksums(&mut req).unwrap();
+        assert_eq!(req.input.checksum_crc32.as_deref(), Some("y/Q5Jg=="));
+        let mut reader = HashReader::from_stream(Cursor::new(b"123456789"), 9, 9, None, None, false).unwrap();
+        reader.add_checksum(&req.headers, None, false).unwrap();
+        let mut bytes = Vec::new();
+        reader.read_to_end(&mut bytes).await.unwrap();
+        assert_eq!(bytes, b"123456789");
+        assert_eq!(reader.content_crc().get("CRC32").map(String::as_str), Some("y/Q5Jg=="));
+    }
+
+    #[test]
+    fn normalize_presigned_part_checksums_rejects_ambiguous_values() {
+        for query in [
+            "x-amz-checksum-sha256=one&x-amz-checksum-sha256=two",
+            "x-amz-sdk-checksum-algorithm=SHA256&x-amz-sdk-checksum-algorithm=CRC32",
+        ] {
+            let mut req = presigned_checksum_request(query);
+            assert_eq!(
+                *normalize_presigned_part_checksums(&mut req).unwrap_err().code(),
+                S3ErrorCode::InvalidRequest
+            );
+        }
+        let mut req = presigned_checksum_request("x-amz-checksum-sha256=one");
+        req.headers.insert("x-amz-checksum-sha256", HeaderValue::from_static("two"));
+        assert_eq!(
+            *normalize_presigned_part_checksums(&mut req).unwrap_err().code(),
+            S3ErrorCode::InvalidRequest
+        );
+    }
+
+    #[test]
+    fn normalize_presigned_part_checksums_preserves_headers_and_rejects_invalid_values() {
+        let mut req = presigned_checksum_request("partNumber=1&uploadId=test");
+        req.input.checksum_sha256 = Some("existing".to_owned());
+        req.headers
+            .insert("x-amz-checksum-sha256", HeaderValue::from_static("existing"));
+        normalize_presigned_part_checksums(&mut req).unwrap();
+        assert_eq!(req.input.checksum_sha256.as_deref(), Some("existing"));
+        assert_eq!(req.headers["x-amz-checksum-sha256"], "existing");
+
+        let mut req = presigned_checksum_request("x-amz-checksum-sha256=%0D%0A");
+        assert_eq!(
+            *normalize_presigned_part_checksums(&mut req).unwrap_err().code(),
+            S3ErrorCode::InvalidArgument
+        );
+    }
+
+    #[test]
+    fn normalize_presigned_part_checksums_requires_verified_presign() {
+        let mut req = presigned_checksum_request("x-amz-checksum-sha256=one");
+        req.extensions.remove::<VerifiedPresignedRequest>();
+        normalize_presigned_part_checksums(&mut req).unwrap();
+        assert!(req.input.checksum_sha256.is_none());
+        assert!(req.headers.is_empty());
+    }
+
+    #[test]
+    fn resolve_upload_part_size_uses_normalized_logical_length() {
+        let result: rustfs_s3_types::S3Result<Option<i64>> = resolve_upload_part_size(Some(5242880), None);
+        assert_eq!(result.expect("DTO length"), Some(5242880));
+        assert_eq!(resolve_upload_part_size(None, None).expect("unknown length"), None);
+        let body = StreamingBlob::from(Bytes::from_static(b"abc"));
+        assert_eq!(resolve_upload_part_size(None, Some(&body)).expect("exact bytes"), Some(3));
+        assert_eq!(resolve_upload_part_size(Some(0), Some(&body)).expect("explicit length wins"), Some(0));
+        let body = StreamingBlob::wrap(futures::stream::iter([Ok::<_, std::io::Error>(Bytes::from_static(b"abc"))]));
+        assert_eq!(futures::Stream::size_hint(&body), (1, Some(1)), "the stream knows its item count");
+        assert_eq!(resolve_upload_part_size(None, Some(&body)).expect("unknown byte length"), None);
+    }
+
+    #[test]
+    fn upload_part_length_contract_rejects_unknown_and_all_negative_lengths() {
+        let result: rustfs_s3_types::S3Result<i64> = require_upload_part_size(None, false);
+        assert_eq!(
+            result.expect_err("ordinary unknown").code(),
+            &rustfs_s3_types::S3ErrorCode::MissingContentLength
+        );
+        assert_eq!(
+            require_upload_part_size(None, true).expect_err("capped unknown").code(),
+            &rustfs_s3_types::S3ErrorCode::UnexpectedContent
+        );
+        for capped in [false, true] {
+            for size in [-1, -2, i64::MIN] {
+                assert_eq!(
+                    require_upload_part_size(Some(size), capped).expect_err("negative").code(),
+                    &rustfs_s3_types::S3ErrorCode::UnexpectedContent
+                );
+            }
+            assert_eq!(require_upload_part_size(Some(0), capped).expect("zero is valid"), 0);
+        }
+    }
+
+    #[test]
+    fn upload_part_timeout_policy_preserves_disabled_and_capped_floor() {
+        for seconds in [0, 1, 299, 300, 601] {
+            let configured = Duration::from_secs(seconds);
+            assert_eq!(upload_part_body_read_timeout(configured, false), configured);
+            assert_eq!(upload_part_body_read_timeout(configured, true), Duration::from_secs(seconds.max(300)));
+        }
+    }
+
+    #[tokio::test]
+    async fn execute_upload_part_rejects_missing_body() {
+        let input = UploadPartInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .upload_id("upload-id".to_string())
+            .part_number(1)
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::PUT);
+
+        let err = make_usecase().execute_upload_part(req).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::IncompleteBody);
+    }
+
+    /// issue #7596: a part whose declared length exceeds the 5 GiB
+    /// single-request ceiling is rejected before the body is polled or the
+    /// store is consulted. Exact-cap and zero-length parts pass admission.
+    #[tokio::test]
+    async fn execute_upload_part_rejects_oversize_declared_part_before_reading_the_body() {
+        let ceiling = i64::try_from(rustfs_config::MAX_SINGLE_PUT_OBJECT_SIZE).expect("ceiling fits i64");
+
+        for (declared, expect_too_large) in [(ceiling + 1, true), (ceiling, false), (0, false)] {
+            let (body, polls) = crate::app::object::PollCountingBody::streaming_blob();
+            let input = UploadPartInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .upload_id("upload-id".to_string())
+                .part_number(1)
+                .body(Some(body))
+                .content_length(Some(declared))
+                .build()
+                .unwrap();
+            let req = build_request(input, Method::PUT);
+
+            let err = make_usecase().execute_upload_part(req).await.unwrap_err();
+            if expect_too_large {
+                assert_eq!(err.code(), &S3ErrorCode::EntityTooLarge, "declared {declared}");
+                assert_eq!(polls.load(std::sync::atomic::Ordering::SeqCst), 0, "body must not be polled");
+            } else {
+                assert_ne!(err.code(), &S3ErrorCode::EntityTooLarge, "declared {declared} must pass admission");
+            }
+        }
+    }
+
+    #[tokio::test]
+    async fn execute_upload_part_rejects_invalid_part_number_before_body_lookup() {
+        for part_number in [-1, 0, 10001] {
+            let input = UploadPartInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .upload_id("upload-id".to_string())
+                .part_number(part_number)
+                .build()
+                .unwrap();
+            let req = build_request(input, Method::PUT);
+
+            let err = make_usecase().execute_upload_part(req).await.unwrap_err();
+            assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+            assert_eq!(err.message(), Some("partNumber must be between 1 and 10000"));
+        }
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_upload_part_rejects_unknown_length_before_admission_or_body_polling() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
+
+        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!("upload-part-length-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket");
+        let concurrency_manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(
+            true,
+            1,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+            Duration::ZERO,
+        ));
+        let held = concurrency_manager
+            .admit_multipart_part(1024)
+            .await
+            .expect("hold the only permit");
+        let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), concurrency_manager);
+
+        for capped in [false, true] {
+            let mut options = ObjectOptions::default();
+            if capped {
+                insert_str(&mut options.user_defined, SUFFIX_MAX_TOTAL_OBJECT_SIZE, "1024".to_owned());
+            }
+            let upload = store
+                .new_multipart_upload(&bucket, "object", &options)
+                .await
+                .expect("upload session");
+            for content_length in [None, Some(-1)] {
+                for declared_chunk_encoding in [false, true] {
+                    let (body, polls) = crate::app::object::PollCountingBody::streaming_blob();
+                    let input = UploadPartInput::builder()
+                        .bucket(bucket.clone())
+                        .key("object".to_owned())
+                        .upload_id(upload.upload_id.clone())
+                        .part_number(1)
+                        .content_length(content_length)
+                        .body(Some(body))
+                        .build()
+                        .expect("part request");
+                    let mut request = build_request(input, Method::PUT);
+                    request
+                        .headers
+                        .insert("x-amz-decoded-content-length", HeaderValue::from_static("1024"));
+                    if declared_chunk_encoding {
+                        request
+                            .headers
+                            .insert("content-encoding", HeaderValue::from_static("aws-chunked"));
+                    }
+                    let error = usecase
+                        .execute_upload_part(request)
+                        .await
+                        .expect_err("unknown or negative logical size");
+                    assert_eq!(
+                        error.code(),
+                        &if capped || content_length.is_some() {
+                            S3ErrorCode::UnexpectedContent
+                        } else {
+                            S3ErrorCode::MissingContentLength
+                        }
+                    );
+                    assert_eq!(polls.load(std::sync::atomic::Ordering::Relaxed), 0);
+                }
+            }
+            let parts = store
+                .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+                .await
+                .expect("list rejected session");
+            assert!(parts.parts.is_empty());
+        }
+        drop(held);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_upload_part_progress_timeout_releases_admission_and_leaves_no_part() {
+        use crate::app::object::request_body::{BodyReadControl, ObservedBody};
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+        use http_body::Frame;
+        use http_body_util::StreamBody;
+        use tokio_stream::wrappers::UnboundedReceiverStream;
+        let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+        let ambient = crate::app::runtime_sources::current_app_context().expect("test app context");
+        let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+        let bucket = format!("part-progress-{}", Uuid::new_v4().simple());
+        store.make_bucket(&bucket, &MakeBucketOptions::default()).await.unwrap();
+        let manager = Arc::new(ConcurrencyManager::with_put_admission_for_test(true, 1, Duration::ZERO));
+        let usecase = DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::clone(&manager));
+        let upload = store
+            .new_multipart_upload(&bucket, "object", &ObjectOptions::default())
+            .await
+            .unwrap();
+        let (sender, receiver) = tokio::sync::mpsc::unbounded_channel::<Result<Frame<Bytes>, std::io::Error>>();
+        let control = BodyReadControl::default();
+        let body = StreamingBlob::from(crate::app::storage_api::s3::Body::http_body_unsync(ObservedBody::new(
+            StreamBody::new(UnboundedReceiverStream::new(receiver)),
+            control.clone(),
+        )));
+        let input = UploadPartInput::builder()
+            .bucket(bucket.clone())
+            .key("object".to_owned())
+            .upload_id(upload.upload_id.clone())
+            .part_number(1)
+            .content_length(Some(65536))
+            .body(Some(body))
+            .build()
+            .unwrap();
+        let mut request = build_request(input, Method::PUT);
+        request.extensions.insert(control.clone());
+        let owner = tokio::spawn(async move { usecase.execute_upload_part(request).await });
+        control.wait_until_pending_for_test().await;
+        assert!(matches!(
+            manager.admit_put_object(65536).await.unwrap(),
+            ForegroundWriteAdmission::Rejected
+        ));
+        control.advance_wait_for_test(Duration::from_secs(60));
+        sender.send(Ok(Frame::data(Bytes::from_static(b"x")))).unwrap();
+        assert_eq!(owner.await.unwrap().unwrap_err().code(), &S3ErrorCode::RequestTimeout);
+        assert!(matches!(
+            manager.admit_put_object(65536).await.unwrap(),
+            ForegroundWriteAdmission::Admitted(_)
+        ));
+        let parts = store
+            .list_object_parts(&bucket, "object", &upload.upload_id, None, 1000, &ObjectOptions::default())
+            .await
+            .unwrap();
+        assert!(parts.parts.is_empty());
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_upload_part_rejects_when_foreground_write_admission_is_full() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations, MakeBucketOptions};
+
+        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!("upload-part-admission-{}", Uuid::new_v4().simple());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("create upload part admission test bucket");
+        let upload = store
+            .new_multipart_upload(&bucket, "object", &ObjectOptions::default())
+            .await
+            .expect("create multipart upload");
+        let concurrency_manager = Arc::new(ConcurrencyManager::with_large_put_admission_for_test(
+            true,
+            1,
+            rustfs_config::DEFAULT_PUT_LARGE_FOREGROUND_ADMISSION_MIN_SIZE_BYTES,
+            Duration::ZERO,
+        ));
+        let held = concurrency_manager
+            .admit_multipart_part(1024)
+            .await
+            .expect("first multipart part should acquire the only permit");
+        let usecase =
+            DefaultMultipartUsecase::with_context_and_concurrency_manager(Some(context), Arc::clone(&concurrency_manager));
+        let body = StreamingBlob::wrap(futures::stream::pending::<Result<Bytes, std::io::Error>>());
+        let input = UploadPartInput::builder()
+            .bucket(bucket)
+            .key("object".to_string())
+            .upload_id(upload.upload_id)
+            .part_number(1)
+            .content_length(Some(1024))
+            .body(Some(body))
+            .build()
+            .expect("upload part input should build");
+
+        let err = usecase
+            .execute_upload_part(build_request(input, Method::PUT))
+            .await
+            .expect_err("full foreground write admission should reject before body ingest");
+        assert_eq!(err.code(), &S3ErrorCode::SlowDown);
+        drop(held);
+    }
+
+    mod body_read_tests;
+}

rustfs/src/app/multipart_usecase/upload_part_copy.rs

@@ -0,0 +1,556 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! UploadPartCopy.
+
+use super::*;
+
+fn normalize_copy_source_version_id(version_id: Option<&str>) -> S3Result<Option<String>> {
+    match version_id {
+        Some(v) => {
+            let trimmed = v.trim();
+            if trimmed.eq_ignore_ascii_case("null") {
+                Ok(Some(Uuid::nil().to_string()))
+            } else if Uuid::parse_str(trimmed).is_ok() {
+                Ok(Some(trimmed.to_string()))
+            } else {
+                Err(S3Error::with_message(
+                    S3ErrorCode::InvalidArgument,
+                    "Invalid version id specified in copy source",
+                ))
+            }
+        }
+        None => Ok(None),
+    }
+}
+
+fn format_copy_source_version_id(version_id: Option<Uuid>, src_versioned: bool, src_version_suspended: bool) -> Option<String> {
+    if src_versioned || src_version_suspended {
+        Some(match version_id {
+            Some(vid) if vid != Uuid::nil() => vid.to_string(),
+            _ => "null".to_string(),
+        })
+    } else {
+        None
+    }
+}
+
+/// Returns InvalidRange error if CopySourceRange end exceeds the source object size.
+/// Used by execute_upload_part_copy to reject out-of-bounds ranges per S3 spec.
+fn validate_copy_source_range_not_exceeds(range_spec: &HTTPRangeSpec, object_size: i64) -> S3Result<()> {
+    if range_spec.end >= object_size {
+        return Err(S3Error::with_message(
+            S3ErrorCode::InvalidRange,
+            "The requested range is not satisfiable".to_string(),
+        ));
+    }
+    Ok(())
+}
+
+impl DefaultMultipartUsecase {
+    #[instrument(level = "debug", skip(self, req))]
+    #[hotpath::measure(impl_type = "MultipartUsecase")]
+    pub async fn execute_upload_part_copy(
+        &self,
+        req: S3Request<UploadPartCopyInput>,
+    ) -> S3Result<S3Response<UploadPartCopyOutput>> {
+        reject_presigned_multipart_max_total_object_size_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedSigV4Request>().is_some(),
+        )?;
+        reject_presigned_put_max_content_length_for_other_operation(
+            &req.headers,
+            req.uri.query(),
+            req.extensions.get::<VerifiedPresignedRequest>().is_some(),
+        )?;
+        // Captured before `req.input` is destructured below.
+        let copy_principal = SseKmsPrincipal::from_request(&req);
+        let source_bucket = match &req.input.copy_source {
+            CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
+            CopySource::Outpost { .. } => return Err(s3_error!(NotImplemented)),
+            CopySource::Bucket { bucket, .. } => bucket.to_string(),
+        };
+        let mut source_generation_opts = ObjectOptions::default();
+        apply_copy_source_bucket_generation_guard(&req, &source_bucket, &mut source_generation_opts)?;
+        let expected_source_incarnation_id = source_generation_opts.expected_bucket_incarnation_id;
+        let mut destination_generation_opts = ObjectOptions::default();
+        apply_bucket_generation_guard(&req, &req.input.bucket, &mut destination_generation_opts)?;
+        let expected_destination_incarnation_id = destination_generation_opts.expected_bucket_incarnation_id;
+        let UploadPartCopyInput {
+            bucket,
+            key,
+            copy_source,
+            copy_source_range,
+            part_number,
+            upload_id,
+            copy_source_if_match,
+            copy_source_if_none_match,
+            sse_customer_algorithm,
+            sse_customer_key,
+            sse_customer_key_md5,
+            copy_source_sse_customer_algorithm,
+            copy_source_sse_customer_key,
+            copy_source_sse_customer_key_md5,
+            ..
+        } = req.input;
+
+        reject_veeam_sosapi_mutation(|| self.object_store(), &bucket, &key).await?;
+        let (src_bucket, src_key, raw_src_version_id) = match copy_source {
+            CopySource::AccessPoint { .. } => return Err(s3_error!(NotImplemented)),
+            CopySource::Outpost { .. } => return Err(s3_error!(NotImplemented)),
+            CopySource::Bucket {
+                bucket: ref src_bucket,
+                key: ref src_key,
+                version_id,
+            } => (src_bucket.to_string(), src_key.to_string(), version_id.map(|v| v.to_string())),
+        };
+        let src_version_id = normalize_copy_source_version_id(raw_src_version_id.as_deref())?;
+
+        let rs = if let Some(range_str) = copy_source_range {
+            Some(parse_copy_source_range(&range_str)?)
+        } else {
+            None
+        };
+
+        let part_id = parse_upload_part_number(part_number)?;
+
+        validate_table_catalog_object_mutation(&bucket, &key).await?;
+
+        let Some(store) = self.object_store() else {
+            return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
+        };
+
+        let (
+            source_bucket_lifecycle_guard,
+            current_source_incarnation_id,
+            destination_bucket_lifecycle_guard_storage,
+            current_destination_incarnation_id,
+        ) = store
+            .acquire_copy_bucket_incarnation_fences(
+                &src_bucket,
+                expected_source_incarnation_id,
+                &bucket,
+                expected_destination_incarnation_id,
+            )
+            .await
+            .map_err(ApiError::from)?;
+        let destination_bucket_lifecycle_guard = destination_bucket_lifecycle_guard_storage
+            .as_ref()
+            .unwrap_or(&source_bucket_lifecycle_guard);
+        if source_bucket_lifecycle_guard.is_lock_lost() || destination_bucket_lifecycle_guard.is_lock_lost() {
+            return Err(ApiError::from(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "copy_bucket_generation",
+                bucket: bucket.clone(),
+                object: key.clone(),
+                required: 1,
+                achieved: 0,
+            })
+            .into());
+        }
+        let mut dst_opts = ObjectOptions {
+            expected_bucket_incarnation_id: Some(current_destination_incarnation_id),
+            ..Default::default()
+        };
+        if src_bucket != bucket {
+            source_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
+        }
+        destination_bucket_lifecycle_guard.attach_to_object_options(&mut dst_opts);
+
+        let mp_info = store
+            .get_multipart_info(&bucket, &key, &upload_id, &dst_opts)
+            .await
+            .map_err(multipart_upload_api_error)?;
+        authorize_sse_kms_multipart_write(copy_principal.as_ref(), &mp_info.user_defined).await?;
+        let destination_size_limit = multipart_max_total_object_size(&mp_info.user_defined)?;
+        EncryptionRequest {
+            bucket: &bucket,
+            key: &key,
+            server_side_encryption: None,
+            ssekms_key_id: None,
+            ssekms_context: None,
+            sse_customer_algorithm: sse_customer_algorithm.clone(),
+            sse_customer_key: sse_customer_key.clone(),
+            sse_customer_key_md5: sse_customer_key_md5.clone(),
+            content_size: 0,
+            principal: None,
+        }
+        .validate_multipart_ssec(&mp_info.user_defined)?;
+
+        let src_versioned = BucketVersioningSys::prefix_enabled(&src_bucket, &src_key).await;
+        let src_version_suspended = BucketVersioningSys::prefix_suspended(&src_bucket, &src_key).await;
+        let mut src_opts = copy_src_opts(&src_bucket, &src_key, &req.headers).map_err(ApiError::from)?;
+        src_opts.version_id = src_version_id.clone();
+        src_opts.versioned = src_versioned;
+        src_opts.version_suspended = src_version_suspended;
+
+        let h = build_ssec_read_headers(
+            copy_source_sse_customer_algorithm.as_ref(),
+            copy_source_sse_customer_key.as_ref(),
+            copy_source_sse_customer_key_md5.as_ref(),
+        );
+        let get_opts = ObjectOptions {
+            version_id: src_opts.version_id.clone(),
+            versioned: src_opts.versioned,
+            version_suspended: src_opts.version_suspended,
+            expected_bucket_incarnation_id: Some(current_source_incarnation_id),
+            ..Default::default()
+        };
+        if source_bucket_lifecycle_guard.is_lock_lost() {
+            return Err(ApiError::from(StorageError::NamespaceLockQuorumUnavailable {
+                mode: "copy_source_bucket_generation",
+                bucket: src_bucket.clone(),
+                object: src_key.clone(),
+                required: 1,
+                achieved: 0,
+            })
+            .into());
+        }
+
+        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 = 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 resolved_src_version_id = format_copy_source_version_id(src_info.version_id, src_versioned, src_version_suspended);
+
+        if let Some(if_match) = copy_source_if_match {
+            if let Some(ref etag) = src_info.etag {
+                if let Some(strong_etag) = if_match.into_etag() {
+                    if ETag::Strong(etag.clone()) != strong_etag {
+                        return Err(s3_error!(PreconditionFailed));
+                    }
+                } else {
+                    return Err(s3_error!(PreconditionFailed));
+                }
+            } else {
+                return Err(s3_error!(PreconditionFailed));
+            }
+        }
+
+        if let Some(if_none_match) = copy_source_if_none_match
+            && let Some(ref etag) = src_info.etag
+            && let Some(strong_etag) = if_none_match.into_etag()
+            && ETag::Strong(etag.clone()) == strong_etag
+        {
+            return Err(s3_error!(PreconditionFailed));
+        }
+
+        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));
+            }
+            None => src_info.size,
+        };
+
+        let (_start_offset, length) = if let Some(ref range_spec) = rs {
+            // Copy-source ranges are expressed over the logical plaintext object.
+            // Encrypted (and compressed) objects have a larger or smaller physical
+            // representation, so validating against `size` rejects valid later parts.
+            validate_copy_source_range_not_exceeds(range_spec, source_logical_size)?;
+
+            range_spec
+                .get_offset_length(source_logical_size)
+                .map_err(|e| S3Error::with_message(S3ErrorCode::InvalidRange, e.to_string()))?
+        } else {
+            (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);
+
+        let actual_size = length;
+
+        let mut write_plan = WritePlan::new();
+        let mut reader = if is_disk_compressed {
+            let algorithm = CompressionAlgorithm::default();
+            let hrd = HashReader::from_stream(src_stream, length, actual_size, None, None, false).map_err(ApiError::from)?;
+            write_plan = write_plan.with_compression(algorithm);
+            hrd
+        } else {
+            HashReader::from_stream(src_stream, length, actual_size, None, None, false).map_err(ApiError::from)?
+        };
+
+        let server_side_encryption = mp_info
+            .user_defined
+            .get("x-amz-server-side-encryption")
+            .map(|s| {
+                ServerSideEncryption::from_str(s)
+                    .map_err(|e| ApiError::from(StorageError::other(format!("Invalid server-side encryption: {e}"))))
+            })
+            .transpose()?;
+        let has_ssec = mp_info
+            .user_defined
+            .contains_key("x-amz-server-side-encryption-customer-algorithm");
+        let ssekms_key_id = match server_side_encryption.as_ref() {
+            Some(sse) if sse.as_str() == ServerSideEncryption::AWS_KMS => mp_info
+                .user_defined
+                .get("x-amz-server-side-encryption-aws-kms-key-id")
+                .map(|s| s.to_string()),
+            _ => None,
+        };
+        let (requested_sse, requested_kms_key_id, dst_user_defined) = if has_ssec {
+            let ssec_material = sse_decryption(DecryptionRequest {
+                bucket: &bucket,
+                key: &key,
+                metadata: &mp_info.user_defined,
+                sse_customer_key: sse_customer_key.as_ref(),
+                sse_customer_key_md5: sse_customer_key_md5.as_ref(),
+                principal: None,
+            })
+            .await?
+            .ok_or_else(|| ApiError::from(StorageError::other("Missing SSE-C session material")))?;
+            let ssec_write = match ssec_material.key_kind {
+                EncryptionKeyKind::Object => WriteEncryption::multipart_object_key(ssec_material.key_bytes, part_id as u32),
+                EncryptionKeyKind::Direct => {
+                    WriteEncryption::multipart(ssec_material.key_bytes, ssec_material.base_nonce, part_id)
+                }
+            };
+            write_plan = write_plan.with_encryption(ssec_write);
+            (
+                Some(ssec_material.server_side_encryption),
+                ssec_material.kms_key_id,
+                mp_info.user_defined.clone(),
+            )
+        } else if let Some(server_side_encryption) = server_side_encryption {
+            // Reuse the destination envelope after authorizing this caller against
+            // its pinned key above. The source read is separately authorized.
+            let managed_material = sse_decryption(DecryptionRequest {
+                bucket: &bucket,
+                key: &key,
+                metadata: &mp_info.user_defined,
+                sse_customer_key: None,
+                sse_customer_key_md5: None,
+                principal: None,
+            })
+            .await?
+            .ok_or_else(|| ApiError::from(StorageError::other("Missing managed SSE session material")))?;
+            let managed_write = match managed_material.key_kind {
+                EncryptionKeyKind::Object => WriteEncryption::multipart_object_key(managed_material.key_bytes, part_id as u32),
+                EncryptionKeyKind::Direct => {
+                    WriteEncryption::multipart(managed_material.key_bytes, managed_material.base_nonce, part_id)
+                }
+            };
+            write_plan = write_plan.with_encryption(managed_write);
+            (Some(server_side_encryption), ssekms_key_id, mp_info.user_defined.clone())
+        } else {
+            (None, None, mp_info.user_defined.clone())
+        };
+
+        reader = write_plan.apply(reader, actual_size).map_err(ApiError::from)?;
+
+        if let Some(checksum_algorithm) = mp_info
+            .user_defined
+            .get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM)
+            .filter(|checksum_algorithm| !checksum_algorithm.is_empty())
+        {
+            let checksum_type = rustfs_rio::ChecksumType::from_string_with_obj_type(
+                checksum_algorithm,
+                mp_info
+                    .user_defined
+                    .get(rustfs_rio::RUSTFS_MULTIPART_CHECKSUM_TYPE)
+                    .map(String::as_str)
+                    .unwrap_or_default(),
+            );
+            if !checksum_type.is_set() {
+                return Err(ApiError::from(StorageError::other(format!(
+                    "Invalid multipart checksum type: {checksum_algorithm}"
+                )))
+                .into());
+            }
+            reader.add_calculated_checksum(checksum_type).map_err(ApiError::from)?;
+        }
+
+        let mut reader = PutObjReader::new(reader);
+
+        dst_opts.user_defined = dst_user_defined;
+
+        let part_info = store
+            .put_object_part(&bucket, &key, &upload_id, part_id, &mut reader, &dst_opts)
+            .await
+            .map_err(ApiError::from)?;
+
+        let copy_checksums = reader.as_hash_reader().content_crc();
+        let checksum_value = |checksum_type: rustfs_rio::ChecksumType| copy_checksums.get(&checksum_type.to_string()).cloned();
+
+        let copy_part_result = CopyPartResult {
+            checksum_crc32: checksum_value(rustfs_rio::ChecksumType::CRC32),
+            checksum_crc32c: checksum_value(rustfs_rio::ChecksumType::CRC32C),
+            checksum_sha1: checksum_value(rustfs_rio::ChecksumType::SHA1),
+            checksum_sha256: checksum_value(rustfs_rio::ChecksumType::SHA256),
+            checksum_crc64nvme: checksum_value(rustfs_rio::ChecksumType::CRC64_NVME),
+            e_tag: part_info.etag.map(|etag| to_s3s_xml_etag(&etag)),
+            last_modified: part_info.last_mod.map(Timestamp::from),
+            ..Default::default()
+        };
+
+        let output = UploadPartCopyOutput {
+            copy_part_result: Some(copy_part_result),
+            copy_source_version_id: resolved_src_version_id,
+            server_side_encryption: requested_sse,
+            ssekms_key_id: requested_kms_key_id,
+            sse_customer_algorithm,
+            sse_customer_key_md5,
+            ..Default::default()
+        };
+
+        Ok(S3Response::new(output))
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::super::test_support::*;
+    use super::*;
+
+    #[tokio::test]
+    async fn execute_upload_part_copy_returns_internal_error_when_store_uninitialized() {
+        let input = UploadPartCopyInput::builder()
+            .bucket("bucket".to_string())
+            .key("object".to_string())
+            .copy_source(CopySource::Bucket {
+                bucket: "src-bucket".into(),
+                key: "src-object".into(),
+                version_id: None,
+            })
+            .part_number(1)
+            .upload_id("upload-id".to_string())
+            .build()
+            .unwrap();
+        let req = build_request(input, Method::PUT);
+
+        let err = Box::pin(make_usecase().execute_upload_part_copy(req)).await.unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[tokio::test]
+    async fn execute_upload_part_copy_rejects_invalid_part_number_before_store_lookup() {
+        for part_number in [-1, 0, 10001] {
+            let input = UploadPartCopyInput::builder()
+                .bucket("bucket".to_string())
+                .key("object".to_string())
+                .copy_source(CopySource::Bucket {
+                    bucket: "src-bucket".into(),
+                    key: "src-object".into(),
+                    version_id: None,
+                })
+                .part_number(part_number)
+                .upload_id("upload-id".to_string())
+                .build()
+                .unwrap();
+            let req = build_request(input, Method::PUT);
+
+            let err = Box::pin(make_usecase().execute_upload_part_copy(req)).await.unwrap_err();
+            assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+            assert_eq!(err.message(), Some("partNumber must be between 1 and 10000"));
+        }
+    }
+
+    #[test]
+    fn test_validate_copy_source_range_not_exceeds_returns_invalid_range_when_range_exceeds() {
+        use super::validate_copy_source_range_not_exceeds;
+
+        let range_exceeds = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: 21,
+        };
+        let err = validate_copy_source_range_not_exceeds(&range_exceeds, 5).unwrap_err();
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRange);
+        assert!(err.to_string().contains("not satisfiable"));
+    }
+
+    #[test]
+    fn test_validate_copy_source_range_not_exceeds_ok_when_range_valid() {
+        use super::validate_copy_source_range_not_exceeds;
+
+        let range_valid = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 0,
+            end: 4,
+        };
+        assert!(validate_copy_source_range_not_exceeds(&range_valid, 5).is_ok());
+    }
+
+    #[test]
+    fn test_validate_copy_source_range_not_exceeds_ok_for_suffix_range() {
+        use super::validate_copy_source_range_not_exceeds;
+
+        let range_suffix = HTTPRangeSpec {
+            is_suffix_length: true,
+            start: -5,
+            end: -1,
+        };
+        assert!(validate_copy_source_range_not_exceeds(&range_suffix, 5).is_ok());
+    }
+
+    #[test]
+    fn upload_part_copy_normalizes_and_formats_source_version_id() {
+        assert_eq!(normalize_copy_source_version_id(None).expect("missing version id"), None);
+        assert_eq!(
+            normalize_copy_source_version_id(Some(" null ")).expect("null version id"),
+            Some(Uuid::nil().to_string())
+        );
+        let valid = Uuid::new_v4();
+        assert_eq!(
+            normalize_copy_source_version_id(Some(&format!(" {valid} "))).expect("valid uuid"),
+            Some(valid.to_string())
+        );
+        let err = normalize_copy_source_version_id(Some("not-a-uuid")).expect_err("invalid uuid should fail");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidArgument);
+
+        assert_eq!(format_copy_source_version_id(Some(Uuid::nil()), false, false), None);
+        assert_eq!(format_copy_source_version_id(None, true, false), Some("null".to_string()));
+        assert_eq!(format_copy_source_version_id(Some(Uuid::nil()), false, true), Some("null".to_string()));
+        assert_eq!(format_copy_source_version_id(Some(valid), true, false), Some(valid.to_string()));
+    }
+
+    mod copy_admission_tests;
+}

rustfs/src/app/storage_api.rs

@@ -56,6 +56,7 @@ pub(crate) mod s3 {
     pub(crate) use s3s::auth::{Credentials, SimpleAuth};
     #[cfg(test)]
     pub(crate) use s3s::config::{S3Config, StaticConfigProvider};
+    pub(crate) use s3s::dto;
     #[cfg(test)]
     pub(crate) use s3s::dto::{GetObjectInput, HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output};
     #[cfg(test)]
@@ -65,6 +66,8 @@ pub(crate) mod s3 {
     #[cfg(test)]
     pub(crate) use s3s::stream::upload_stream::{UploadStream, UploadStreamError};
     #[cfg(test)]
+    pub(crate) use s3s::xml;
+    #[cfg(test)]
     pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer};
     #[cfg(test)]
     pub(crate) use s3s::{Body, BodySizeLimitExceeded, S3, S3Response};

rustfs/src/guard_sources.rs

@@ -0,0 +1,287 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Test-only source reader for text guards that must survive a mechanical file split.
+//!
+//! A text guard asserts that some literal appears in production source. When it reads one file
+//! with `include_str!`, moving the guarded function into another file either breaks the guard or,
+//! worse, makes it stop seeing the code it guards. A guard built on [`SourceSet`] instead names
+//! the file it reads today plus the sibling directory a split moves that file's code into, and
+//! reads every `.rs` file of both at test time.
+//!
+//! The named file is required: a missing file panics with its path, so a guard never passes on
+//! absent input. The directory is optional while the split has not happened; once it exists,
+//! every `.rs` file under it (recursively) is read in byte order of its manifest-relative path,
+//! after the named file.
+
+use std::path::{Path, PathBuf};
+
+/// One source file of a [`SourceSet`].
+pub(crate) struct SourceFile {
+    /// Path relative to the `rustfs` package root, with `/` separators.
+    pub(crate) path: String,
+    pub(crate) text: String,
+}
+
+/// The ordered files a text guard reads: one required file, then the split directory.
+pub(crate) struct SourceSet {
+    files: Vec<SourceFile>,
+}
+
+impl SourceSet {
+    /// Reads `file` and every `.rs` file under `split_dir`, both relative to the `rustfs`
+    /// package root.
+    ///
+    /// # Panics
+    ///
+    /// Panics, naming the path, when `file` is missing or unreadable, when `split_dir` exists but
+    /// is not a directory, or when a file under `split_dir` cannot be read.
+    pub(crate) fn read(file: &str, split_dir: &str) -> Self {
+        Self::read_under(Path::new(env!("CARGO_MANIFEST_DIR")), file, split_dir)
+    }
+
+    fn read_under(root: &Path, file: &str, split_dir: &str) -> Self {
+        let required = root.join(file);
+        let text = std::fs::read_to_string(&required)
+            .unwrap_or_else(|err| panic!("guard source {} must exist and be readable: {err}", required.display()));
+        let mut files = vec![SourceFile {
+            path: file.to_owned(),
+            text,
+        }];
+
+        let dir = root.join(split_dir);
+        if dir.exists() {
+            assert!(dir.is_dir(), "guard split path {} must be a directory", dir.display());
+            let mut paths = Vec::new();
+            collect_rust_files(&dir, &mut paths);
+            let mut split_files: Vec<SourceFile> = paths
+                .into_iter()
+                .map(|path| {
+                    let text = std::fs::read_to_string(&path)
+                        .unwrap_or_else(|err| panic!("guard source {} must be readable: {err}", path.display()));
+                    let relative = path
+                        .strip_prefix(root)
+                        .unwrap_or_else(|_| panic!("guard source {} must be under {}", path.display(), root.display()));
+                    let path = relative
+                        .components()
+                        .map(|component| component.as_os_str().to_string_lossy())
+                        .collect::<Vec<_>>()
+                        .join("/");
+                    SourceFile { path, text }
+                })
+                .collect();
+            split_files.sort_by(|left, right| left.path.cmp(&right.path));
+            files.extend(split_files);
+        }
+
+        Self { files }
+    }
+
+    /// Every file of the set, the required file first.
+    pub(crate) fn files(&self) -> &[SourceFile] {
+        &self.files
+    }
+
+    /// Whether any one file of the set contains `needle`.
+    pub(crate) fn contains(&self, needle: &str) -> bool {
+        self.files.iter().any(|file| file.text.contains(needle))
+    }
+
+    /// Non-overlapping occurrences of `needle`, summed over the files; a match never spans two
+    /// files.
+    pub(crate) fn count(&self, needle: &str) -> usize {
+        self.files.iter().map(|file| file.text.matches(needle).count()).sum()
+    }
+
+    /// The text after `marker` in the one file that contains it.
+    ///
+    /// # Panics
+    ///
+    /// Panics unless `marker` occurs exactly once across the whole set.
+    pub(crate) fn after_unique_marker(&self, marker: &str) -> &str {
+        let occurrences = self.count(marker);
+        assert_eq!(occurrences, 1, "marker {marker:?} must occur exactly once across {}", self.describe());
+        self.files
+            .iter()
+            .find_map(|file| file.text.split_once(marker).map(|(_, rest)| rest))
+            .unwrap_or_else(|| unreachable!("a counted marker is found"))
+    }
+
+    /// Manifest-relative paths of the set, for panic messages.
+    pub(crate) fn describe(&self) -> String {
+        self.files
+            .iter()
+            .map(|file| file.path.as_str())
+            .collect::<Vec<_>>()
+            .join(", ")
+    }
+}
+
+fn collect_rust_files(dir: &Path, out: &mut Vec<PathBuf>) {
+    let entries =
+        std::fs::read_dir(dir).unwrap_or_else(|err| panic!("guard split directory {} must be readable: {err}", dir.display()));
+    for entry in entries {
+        let entry = entry.unwrap_or_else(|err| panic!("guard split directory {} must be listable: {err}", dir.display()));
+        let path = entry.path();
+        if path.is_dir() {
+            collect_rust_files(&path, out);
+        } else if path.extension().is_some_and(|extension| extension == "rs") {
+            out.push(path);
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::SourceSet;
+    use std::fs;
+    use std::path::Path;
+
+    fn write(root: &Path, relative: &str, text: &str) {
+        let path = root.join(relative);
+        fs::create_dir_all(path.parent().expect("fixture path has a parent")).expect("create fixture dir");
+        fs::write(path, text).expect("write fixture");
+    }
+
+    fn panic_message(run: impl FnOnce() + std::panic::UnwindSafe) -> String {
+        let payload = std::panic::catch_unwind(run).expect_err("the guard input must be rejected");
+        payload
+            .downcast_ref::<String>()
+            .cloned()
+            .or_else(|| payload.downcast_ref::<&str>().map(|message| (*message).to_owned()))
+            .unwrap_or_default()
+    }
+
+    #[test]
+    fn absent_split_directory_reads_only_the_required_file() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(set.describe(), "src/app/usecase.rs");
+        assert!(set.contains("guarded();"));
+        assert_eq!(set.count("guarded();"), 1);
+    }
+
+    #[test]
+    fn split_directory_files_follow_the_required_file_in_path_order() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/zeta.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/alpha.rs", "guarded();\nguarded();\n");
+        write(root.path(), "src/app/usecase/tests/nested.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase/notes.txt", "guarded();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(
+            set.describe(),
+            "src/app/usecase.rs, src/app/usecase/alpha.rs, src/app/usecase/tests/nested.rs, src/app/usecase/zeta.rs"
+        );
+        assert_eq!(set.count("guarded();"), 5, "non-.rs files are not source");
+    }
+
+    #[test]
+    fn missing_required_file_fails_even_when_the_split_directory_exists() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase/moved.rs", "guarded();\n");
+        let root_path = root.path().to_path_buf();
+
+        let message = panic_message(move || {
+            SourceSet::read_under(&root_path, "src/app/usecase.rs", "src/app/usecase");
+        });
+
+        assert!(message.contains("src/app/usecase.rs"), "panic must name the missing file: {message}");
+    }
+
+    #[test]
+    fn split_path_that_is_a_file_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guarded();\n");
+        write(root.path(), "src/app/usecase", "not a directory\n");
+        let root_path = root.path().to_path_buf();
+
+        let message = panic_message(move || {
+            SourceSet::read_under(&root_path, "src/app/usecase.rs", "src/app/usecase");
+        });
+
+        assert!(message.contains("must be a directory"), "{message}");
+    }
+
+    #[test]
+    fn count_does_not_join_a_needle_across_two_files() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "guard");
+        write(root.path(), "src/app/usecase/tail.rs", "ed();\n");
+
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        assert_eq!(set.count("guarded();"), 0);
+        assert!(!set.contains("guarded();"));
+    }
+
+    #[test]
+    fn unique_marker_absent_from_the_set_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn other() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("exactly once"), "{message}");
+    }
+
+    #[test]
+    fn unique_marker_in_two_files_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn guarded() {}\n");
+        write(root.path(), "src/app/usecase/copy.rs", "fn guarded() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("src/app/usecase/copy.rs"), "panic must name the set: {message}");
+    }
+
+    #[test]
+    fn unique_marker_twice_in_one_file_fails() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn guarded() {}\nfn guarded() {}\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let message = panic_message(move || {
+            set.after_unique_marker("fn guarded");
+        });
+
+        assert!(message.contains("exactly once"), "{message}");
+    }
+
+    #[test]
+    fn unique_marker_returns_the_rest_of_its_own_file_only() {
+        let root = tempfile::tempdir().expect("tempdir");
+        write(root.path(), "src/app/usecase.rs", "fn head() {}\n");
+        write(root.path(), "src/app/usecase/moved.rs", "fn guarded() { body(); }\n");
+        write(root.path(), "src/app/usecase/next.rs", "fn later() { other(); }\n");
+        let set = SourceSet::read_under(root.path(), "src/app/usecase.rs", "src/app/usecase");
+
+        let rest = set.after_unique_marker("fn guarded");
+
+        assert_eq!(rest, "() { body(); }\n");
+    }
+}

rustfs/src/lib.rs

@@ -89,6 +89,8 @@ pub mod diagnose;
 pub mod embedded;
 pub mod error;
 pub mod foreground_observation;
+#[cfg(test)]
+pub(crate) mod guard_sources;
 pub mod init;
 pub mod inspect;
 pub(crate) mod kms_deletion_gate;

rustfs/src/site_replication/tests.rs

@@ -1334,11 +1334,11 @@ fn test_delete_bucket_broadcast_fences_target_membership_through_delivery() {
         "a failed destructive delivery must remain visible for operator repair"
     );
 
-    let usecase = include_str!("../app/bucket_usecase.rs");
+    let usecase = crate::guard_sources::SourceSet::read("src/app/bucket_usecase.rs", "src/app/bucket_usecase");
     let delete = usecase
-        .split("async fn execute_delete_bucket_inner")
-        .nth(1)
-        .and_then(|rest| rest.split("pub async fn execute_head_bucket").next())
+        .after_unique_marker("async fn execute_delete_bucket_inner")
+        .split("pub async fn execute_head_bucket")
+        .next()
         .expect("delete bucket usecase");
     assert!(
         delete

rustfs/src/storage/ecfs.rs

@@ -2276,13 +2276,13 @@ impl S3 for FS {
 
 /// Every mutating S3 operation that has no bucket notification event reaches the audit
 /// stream only through `audit_only_operation`; a new such operation, or a dropped
-/// wrapper, must fail here rather than silently go unaudited.
+/// wrapper, must fail here rather than silently go unaudited. The wrappers are looked
+/// up in this file and in `legacy_edge/`, where a split of this file moves them.
 #[cfg(test)]
 mod audit_coverage_tests {
+    use crate::guard_sources::SourceSet;
     use rustfs_s3_ops::S3Operation;
 
-    const SOURCE: &str = include_str!("ecfs.rs");
-
     fn is_read_only(op: S3Operation) -> bool {
         let name = op.as_str().trim_start_matches("s3:");
         ["Get", "Head", "List", "Select"]
@@ -2300,9 +2300,15 @@ mod audit_coverage_tests {
         expected.push(S3Operation::AbortMultipartUpload);
         assert_eq!(expected.len(), 22, "an operation was added or removed: {expected:?}");
 
+        let source = SourceSet::read("src/storage/ecfs.rs", "src/storage/legacy_edge");
         for op in expected {
             let wrapper = format!("audit_only_operation(req, S3Operation::{op:?},");
-            assert!(SOURCE.contains(&wrapper), "{} is not run through audit_only_operation", op.as_str());
+            assert!(
+                source.contains(&wrapper),
+                "{} is not run through audit_only_operation in {}",
+                op.as_str(),
+                source.describe()
+            );
         }
     }
 }

rustfs/src/table_catalog/tests.rs

@@ -14478,31 +14478,31 @@ fn object_mutation_entrypoints_call_reserved_prefix_guard() {
 
 #[test]
 fn multipart_mutation_entrypoints_call_reserved_prefix_guard() {
-    let source = include_str!("../app/multipart_usecase.rs");
+    let source = crate::guard_sources::SourceSet::read("src/app/multipart_usecase.rs", "src/app/multipart_usecase");
 
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;")
-            .count(),
-        4
+        source.count("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
+        4,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
 }
 
 #[test]
 fn object_metadata_mutation_entrypoints_call_reserved_prefix_guard() {
-    let source = include_str!("../storage/ecfs.rs");
+    let source = crate::guard_sources::SourceSet::read("src/storage/ecfs.rs", "src/storage/legacy_edge");
 
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &object).await?;")
-            .count(),
-        2
+        source.count("validate_table_catalog_object_mutation(&bucket, &object).await?;"),
+        2,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
     assert_eq!(
-        source
-            .matches("validate_table_catalog_object_mutation(&bucket, &key).await?;")
-            .count(),
-        2
+        source.count("validate_table_catalog_object_mutation(&bucket, &key).await?;"),
+        2,
+        "reserved-prefix guard calls across {}",
+        source.describe()
     );
 }
 

scripts/README.md

@@ -38,6 +38,7 @@ their issue closes.
 | `check_no_tokio_io_uring.sh` | ci-gate | Keeps tokio's io-uring backend disabled | ci.yml Quick Checks |
 | `check_s3s_footprint.sh` | ci-gate | Exact, lower-only baseline on the direct s3s surface ahead of the gateway migration (reads `.config/s3s-footprint-baseline.txt`; `s3_error_lines` leaves out lines whose file imports the macro from `rustfs-s3-types`, rule in the script header); `--mode allowlist` fails on any Rust file referencing s3s outside `.config/s3s-edge-allowlist.txt`, `--dry-run` lists them | ci.yml Quick Checks; `make pre-commit`; `make script-tests` (`--self-test`) |
 | `check_test_count_ratchet.sh` | ci-gate | Per-crate test-count floor and `#[ignore]` ceiling against `.config/test-count-baseline.txt`; no crate may fall below its tests row or rise above its ignored row (extra tests and new crates pass with a note) | ci.yml Quick Checks; `make pre-commit` |
+| `guard_source_set.sh` | ci-gate (library) | Sourced helper that resolves a text guard's input as a required file plus the directory a mechanical split moves its code into (rustfs/backlog#2765); a missing required file fails | Sourced by `check_architecture_migration_rules.sh` and `check_logging_guardrails.sh`; contract tests `test_guard_source_set.sh` (`make script-tests`) |
 | `check_unsafe_code_allowances.sh` | ci-gate | Unsafe-code allowance ledger guard | ci.yml Quick Checks |
 | `layer-dependency-baseline.txt` | ci-gate (data) | Committed baseline consumed by `check_layer_dependencies.sh` | [Architecture guard troubleshooting](../docs/operations/architecture-guard-troubleshooting.md) |
 | `static.sh` | ci-gate | Static-build helper executed inside image builds | `Dockerfile.source`, `Dockerfile.decommission-local` |
@@ -57,6 +58,9 @@ their issue closes.
 | `test_object_generation_protocol_model.py` | dev-tool | Bounded single-slot, four-voter promise/accept design model and recovery/retirement examples; no production runtime evidence | `python3 scripts/test_object_generation_protocol_model.py`; [Generation contract](../docs/architecture/unified-object-generation.md#executable-design-model-and-acceptance-mapping) |
 | `junit_diff.py` | dev-tool | Case-by-case pass/fail/skip diff of two JUnit XML reports (reference vs candidate stack); exit 1 on any difference or stale `--allow` entry, 2 on unreadable input | `python3 scripts/junit_diff.py a.xml b.xml`; backlog T3.1 |
 | `test_junit_diff.py` | ci-gate | Contract tests and fixtures (`scripts/tests/junit_diff/`) for `junit_diff.py` | `make script-tests` |
+| `s3-tests/reference_parity.py` | ci-gate | Plans the reference parity targets, lists a run's failures for the single rerun, and compares every candidate's s3-tests JUnit against the pinned reference per node id with a tolerated-difference file | `.github/workflows/reference-parity.yml`; backlog T3.2 |
+| `s3-tests/test_reference_parity.py` | ci-gate | Contract tests for `reference_parity.py` | `make script-tests` |
+| `test_reference_binary.py` | ci-gate | Contract tests for `.github/actions/reference-binary/reference_binary.py` (pin resolution and verified unpacking) | `make script-tests` |
 | `e2e-run.sh` | ci-gate | Boots a rustfs server and runs the `s3s-e2e` black-box conformance tool against it | ci.yml `e2e-tests` jobs; `docs/testing/README.md` |
 | `run_ecstore_validation_suite.sh` | dev-tool | ecstore black-box validation suite (`quick`/`full`/`destructive`/`fuzz` profiles) | `docs/testing/README.md`, `docs/testing/ecstore-validation-suite-design.md` |
 | `run_e2e_tests.sh` | dev-tool | Local `e2e_test` crate runner (starts a server, applies filters, cleans up) | `crates/e2e_test/README.md` |

scripts/check_architecture_migration_rules.sh

@@ -118,6 +118,28 @@ require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "Do
 TMP_DIR="$(mktemp -d)"
 trap 'rm -rf "$TMP_DIR"' EXIT
 
+# Rules about one of the large files below read that file plus the sibling
+# directory a mechanical split moves its code into (rustfs/backlog#2765), so a
+# split neither breaks nor silently blinds them. The *_SPLIT_SOURCES arrays hold
+# only the directory part, for rules that select the file itself by glob.
+# shellcheck source=scripts/guard_source_set.sh
+source "${ROOT_DIR}/scripts/guard_source_set.sh"
+guard_source_set "$ROOT_DIR" rustfs/src/storage/ecfs.rs rustfs/src/storage/legacy_edge ||
+  report_failure "ECFS guard source set is unreadable"
+ECFS_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/bucket_usecase.rs rustfs/src/app/bucket_usecase ||
+  report_failure "bucket usecase guard source set is unreadable"
+BUCKET_USECASE_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+BUCKET_USECASE_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/multipart_usecase.rs rustfs/src/app/multipart_usecase ||
+  report_failure "multipart usecase guard source set is unreadable"
+MULTIPART_USECASE_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+MULTIPART_USECASE_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+guard_source_set "$ROOT_DIR" rustfs/src/app/lifecycle_transition_api_test.rs rustfs/src/app/lifecycle_transition_api_test ||
+  report_failure "lifecycle transition API test guard source set is unreadable"
+LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES=("${GUARD_SOURCES[@]}")
+LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES=("${GUARD_SOURCES[@]:1}")
+
 PR_TYPES_FILE="${TMP_DIR}/pr_types.txt"
 PR_TYPE_HITS_FILE="${TMP_DIR}/pr_type_hits.txt"
 SOURCE_MARKERS_FILE="${TMP_DIR}/source_markers.txt"
@@ -2157,7 +2179,7 @@ fi
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::app::context::(?:\{[^}]*\b(?:resolve_object_store_handle|resolve_buffer_config|resolve_internode_metrics|resolve_local_node_name|resolve_action_credentials|resolve_notify_interface|resolve_performance_metrics|resolve_encryption_service|resolve_region|resolve_ready_iam_handle)\b|(?:resolve_object_store_handle|resolve_buffer_config|resolve_internode_metrics|resolve_local_node_name|resolve_action_credentials|resolve_notify_interface|resolve_performance_metrics|resolve_encryption_service|resolve_region|resolve_ready_iam_handle)\b)' \
     rustfs/src/storage/access.rs \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     rustfs/src/storage/ecfs_extend.rs \
     rustfs/src/storage/helper.rs \
     rustfs/src/storage/concurrency/manager.rs \
@@ -2175,7 +2197,7 @@ fi
 (
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::app::(?:bucket_usecase|multipart_usecase|object_usecase|object\b)|Default(?:Bucket|Multipart|Object)Usecase::from_global\(\)' \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     --glob '*.rs' || true
 ) >"$RUSTFS_STORAGE_ECFS_USECASE_BYPASS_HITS_FILE"
 
@@ -2228,7 +2250,10 @@ fi
 
 (
   cd "$ROOT_DIR"
-  rg -n --with-filename 'use crate::storage::\*;' rustfs/src/app --glob '*_usecase.rs' --glob 'object/*.rs' || true
+  # A glob containing a slash is anchored at the search root, so object files
+  # need the leading **/ to match under rustfs/src/app.
+  rg -n --with-filename 'use crate::storage::\*;' rustfs/src/app --glob '*_usecase.rs' --glob '**/object/*.rs' \
+    "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
 ) >"$RUSTFS_APP_USECASE_STORAGE_WILDCARD_HITS_FILE"
 
 if [[ -s "$RUSTFS_APP_USECASE_STORAGE_WILDCARD_HITS_FILE" ]]; then
@@ -2246,7 +2271,8 @@ fi
 
 (
   cd "$ROOT_DIR"
-  rg -n --with-filename 'crate::storage::s3_api::|use crate::storage::s3_api|super::s3_api::|use super::s3_api' rustfs/src/app --glob '*_usecase.rs' --glob 'object/*.rs' || true
+  rg -n --with-filename 'crate::storage::s3_api::|use crate::storage::s3_api|super::s3_api::|use super::s3_api' rustfs/src/app --glob '*_usecase.rs' --glob '**/object/*.rs' \
+    "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
 ) >"$RUSTFS_APP_USECASE_S3_API_BYPASS_HITS_FILE"
 
 if [[ -s "$RUSTFS_APP_USECASE_S3_API_BYPASS_HITS_FILE" ]]; then
@@ -2258,13 +2284,14 @@ fi
   {
     rg -n --with-filename \
       '(use crate::storage::(access|helper|options|request_context|sse|timeout_wrapper|head_prefix|concurrency|ecfs)|crate::storage::sse::EncryptionKeyKind|use crate::storage::\{|use crate::storage::[A-Z])' \
-      rustfs/src/app/select_object.rs rustfs/src/app/*_usecase.rs rustfs/src/app/object || true
+      rustfs/src/app/select_object.rs rustfs/src/app/*_usecase.rs rustfs/src/app/object \
+      "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" || true
     rg -n --with-filename \
       'use super::(?:\{[^}]*\b(?:DynReader|HashReader|WriteEncryption|WritePlan|DecryptReader|EncryptReader|HardLimitReader|boxed_reader|wrap_reader|compression_metadata_value|is_disk_compressible|MIN_DISK_COMPRESSIBLE_SIZE|get_lock_acquire_timeout|is_valid_storage_class|StorageError|DiskError|is_all_buckets_not_found|is_err_bucket_not_found|is_err_object_not_found|is_err_version_not_found)\b|(?:object_api_utils::to_s3s_etag|storageclass|StorageError|DiskError|DynReader|HashReader|WriteEncryption|WritePlan|DecryptReader|EncryptReader|HardLimitReader|boxed_reader|wrap_reader|compression_metadata_value|is_disk_compressible|MIN_DISK_COMPRESSIBLE_SIZE|get_lock_acquire_timeout|is_valid_storage_class|is_all_buckets_not_found|is_err_bucket_not_found|is_err_object_not_found|is_err_version_not_found)\b)' \
-      rustfs/src/app/bucket_usecase.rs rustfs/src/app/object_usecase.rs rustfs/src/app/object rustfs/src/app/multipart_usecase.rs rustfs/src/app/lifecycle_transition_api_test.rs || true
+      "${BUCKET_USECASE_GUARD_SOURCES[@]}" rustfs/src/app/object_usecase.rs rustfs/src/app/object "${MULTIPART_USECASE_GUARD_SOURCES[@]}" "${LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES[@]}" || true
     rg -n --with-filename \
       'use super::(?:\{[^}]*\b(?:AppObjectLockConfigExt|AppReplicationConfigExt|AppVersioningConfigExt|predict_lifecycle_expiration|validate_restore_request|bucket_target_sys|lifecycle|metadata|metadata_sys|object_lock|policy_sys|quota|replication|tagging|target|utils|versioning_sys|transition_api|ObjectInfo|ObjectOptions)\b|(?:AppObjectLockConfigExt|AppReplicationConfigExt|AppVersioningConfigExt|predict_lifecycle_expiration|validate_restore_request|bucket_target_sys|lifecycle|metadata|metadata_sys|object_lock|policy_sys|quota|replication|tagging|target|utils|versioning_sys|transition_api|ObjectInfo|ObjectOptions)\b)|super::(?:lifecycle|metadata_sys|object_lock|quota|replication|tagging|target|utils|versioning_sys|transition_api)::|super::super::(?:metadata_sys|lifecycle|target)::' \
-      rustfs/src/app/bucket_usecase.rs rustfs/src/app/object_usecase.rs rustfs/src/app/object rustfs/src/app/multipart_usecase.rs rustfs/src/app/lifecycle_transition_api_test.rs rustfs/src/app/capacity_dirty_scope_test.rs rustfs/src/app/context.rs rustfs/src/app/context/handles.rs rustfs/src/app/context/interfaces.rs rustfs/src/app/context/runtime_sources.rs || true
+      "${BUCKET_USECASE_GUARD_SOURCES[@]}" rustfs/src/app/object_usecase.rs rustfs/src/app/object "${MULTIPART_USECASE_GUARD_SOURCES[@]}" "${LIFECYCLE_TRANSITION_TEST_GUARD_SOURCES[@]}" rustfs/src/app/capacity_dirty_scope_test.rs rustfs/src/app/context.rs rustfs/src/app/context/handles.rs rustfs/src/app/context/interfaces.rs rustfs/src/app/context/runtime_sources.rs || true
   }
 ) >"$RUSTFS_APP_USECASE_STORAGE_API_BYPASS_HITS_FILE"
 
@@ -2277,6 +2304,7 @@ fi
   rg -n --with-filename \
     '(use crate::storage::(access|request_context|ecfs)|use crate::storage::\{[^}]*Storage(ObjectInfo|ObjectOptions|PutObjReader)|crate::storage::StorageObjectOptions)' \
     rustfs/src/app/*_test.rs \
+    "${LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES[@]}" \
     rustfs/src/admin/router.rs \
     rustfs/src/admin/console.rs \
     rustfs/src/admin/handlers/heal.rs \
@@ -4561,7 +4589,7 @@ fi
   cd "$ROOT_DIR"
   rg -n --with-filename 'crate::storage::core_storage_compat' \
     rustfs/src/storage/mod.rs \
-    rustfs/src/storage/ecfs.rs \
+    "${ECFS_GUARD_SOURCES[@]}" \
     rustfs/src/storage/ecfs_extend.rs \
     rustfs/src/storage/access.rs \
     rustfs/src/storage/head_prefix.rs \
@@ -4577,7 +4605,8 @@ fi
   cd "$ROOT_DIR"
   {
     rg -n --with-filename 'crate::app::usecase_storage_compat' \
-      rustfs/src/app/*.rs || true
+      rustfs/src/app/*.rs "${BUCKET_USECASE_SPLIT_SOURCES[@]}" "${MULTIPART_USECASE_SPLIT_SOURCES[@]}" \
+      "${LIFECYCLE_TRANSITION_TEST_SPLIT_SOURCES[@]}" || true
     rg -n --with-filename 'crate::admin::router_storage_compat' \
       rustfs/src/admin/router.rs || true
     rg -n --with-filename 'crate::storage::core_storage_compat' \

scripts/check_layer_dependencies.sh

@@ -22,7 +22,8 @@ classify_source_layer() {
     printf 'composition'
   elif [[ "$file" == rustfs/src/app/* ]]; then
     printf 'app'
-  elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]] ||
+  elif [[ "$file" == rustfs/src/admin/* ]] || [[ "$file" == rustfs/src/storage/ecfs.rs ]] ||
+    [[ "$file" == rustfs/src/storage/legacy_edge/* ]] || [[ "$file" == rustfs/src/storage/s3_api/* ]] ||
     [[ "$file" == rustfs/src/storage/s3_compat/* ]]; then
     printf 'interface'
   elif [[ "$file" == rustfs/src/* ]]; then
@@ -55,6 +56,7 @@ classify_target_layer() {
     storage)
       storage_path="${path#storage::}"
       if [[ "$storage_path" == "ecfs" ]] || [[ "$storage_path" == ecfs::* ]] ||
+        [[ "$storage_path" == "legacy_edge" ]] || [[ "$storage_path" == legacy_edge::* ]] ||
         [[ "$storage_path" == "s3_api" ]] || [[ "$storage_path" == s3_api::* ]] ||
         [[ "$storage_path" == "s3_compat" ]] || [[ "$storage_path" == s3_compat::* ]]; then
         printf 'interface'
@@ -137,6 +139,60 @@ run_layer_model_self_tests() {
   assert_dependency_direction 'allowed' "$s3_compat_source" "$app_target"
   assert_dependency_direction 'reverse' "$app_source" "$s3_compat_target"
   assert_dependency_direction 'reverse' "$storage_source" "$s3_compat_target"
+
+  # A mechanical split moves ecfs.rs code into storage/legacy_edge/ and the
+  # usecase files' code into same-named directories; each keeps its layer.
+  assert_source_layer 'interface' rustfs/src/storage/ecfs.rs
+  assert_source_layer 'interface' rustfs/src/storage/legacy_edge/bucket_core.rs
+  assert_source_layer 'interface' rustfs/src/storage/legacy_edge/tests/object_meta.rs
+  assert_source_layer 'infra' rustfs/src/storage/legacy_edge.rs
+  assert_source_layer 'infra' rustfs/src/storage/legacy_edge_helper.rs
+  assert_source_layer 'app' rustfs/src/app/bucket_usecase/cors.rs
+  assert_source_layer 'app' rustfs/src/app/multipart_usecase/complete.rs
+  assert_target_layer 'interface' storage::legacy_edge
+  assert_target_layer 'interface' storage::legacy_edge::object_meta
+  assert_target_layer 'infra' storage::legacy_edge_helper
+  assert_dependency_direction 'allowed' "$(classify_source_layer rustfs/src/storage/legacy_edge/bucket_core.rs)" "$app_target"
+  assert_dependency_direction 'reverse' "$app_source" "$(classify_target_layer storage::legacy_edge::bucket_core)"
+  assert_dependency_direction 'reverse' "$storage_source" "$(classify_target_layer storage::legacy_edge)"
+  assert_dedicated_test_file 'yes' rustfs/src/app/lifecycle_transition_api_test.rs
+  assert_dedicated_test_file 'yes' rustfs/src/app/lifecycle_transition_api_test/put.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/lifecycle_transition_api_test_support.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/lifecycle_transition_api.rs
+  assert_dedicated_test_file 'no' rustfs/src/app/bucket_usecase/cors.rs
+  assert_dedicated_test_file 'no' rustfs/src/storage/legacy_edge/bucket_core.rs
+}
+
+assert_source_layer() {
+  local expected="$1" file="$2" actual
+
+  actual="$(classify_source_layer "$file")"
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: %s classified as %s (expected %s)\n' "$file" "$actual" "$expected" >&2
+    exit 1
+  fi
+}
+
+assert_target_layer() {
+  local expected="$1" path="$2" actual
+
+  actual="$(classify_target_layer "$path")"
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: crate::%s classified as %s (expected %s)\n' "$path" "$actual" "$expected" >&2
+    exit 1
+  fi
+}
+
+assert_dedicated_test_file() {
+  local expected="$1" file="$2" actual='no'
+
+  if is_dedicated_test_file "$file"; then
+    actual='yes'
+  fi
+  if [[ "$actual" != "$expected" ]]; then
+    printf 'Layer dependency guard self-test failed: %s dedicated test file = %s (expected %s)\n' "$file" "$actual" "$expected" >&2
+    exit 1
+  fi
 }
 
 normalize_import_group_item() {
@@ -251,11 +307,20 @@ normalize_import_path() {
   printf '%s\n' "$path"
 }
 
+# The guard is file-scoped: dedicated test modules are excluded, while inline
+# #[cfg(test)] imports remain subject to the source file's layer. The
+# lifecycle_transition_api_test/ directory is where a mechanical split of
+# lifecycle_transition_api_test.rs moves its tests (rustfs/backlog#2765).
+is_dedicated_test_file() {
+  local file="$1"
+
+  [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]] ||
+    [[ "$file" == rustfs/src/app/lifecycle_transition_api_test/* ]]
+}
+
 emit_crate_use_statements() {
   (cd "$ROOT_DIR" && rg --files -g '*.rs' rustfs/src | while IFS= read -r file; do
-    # The guard is file-scoped: dedicated test modules are excluded, while
-    # inline #[cfg(test)] imports remain subject to the source file's layer.
-    if [[ "$file" == *_test.rs ]] || [[ "$file" == */tests/* ]]; then
+    if is_dedicated_test_file "$file"; then
       continue
     fi
     perl -0777 -ne '

scripts/check_logging_guardrails.sh

@@ -1094,11 +1094,29 @@ trace_hot_spans=(
   "rustfs/src/app/object:execute_get_object"
 )
 
+# A hot span in a file that is due for a mechanical split (rustfs/backlog#2765)
+# is also looked up in the directory the split moves its functions into; the
+# named file itself stays required.
+# shellcheck source=scripts/guard_source_set.sh
+source "${repo_root}/scripts/guard_source_set.sh"
+
 for hot_span in "${trace_hot_spans[@]}"; do
   file="${hot_span%%:*}"
   function="${hot_span##*:}"
+  hot_span_sources=("$file")
+  split_dir=""
+  case "$file" in
+    rustfs/src/app/bucket_usecase.rs) split_dir="rustfs/src/app/bucket_usecase" ;;
+  esac
+  if [[ -n "$split_dir" ]]; then
+    if ! guard_source_set "$repo_root" "$file" "$split_dir"; then
+      echo "❌ logging guardrail violation: $file::$function guard source is missing" >&2
+      exit 1
+    fi
+    hot_span_sources=("${GUARD_SOURCES[@]}")
+  fi
   trace_span_pattern="#\\[(tracing::)?instrument\\([^]]*level = \\\"trace\\\"[^]]*\\)\\]([[:space:]]+#\\[[^]]+\\])*[[:space:]]+(pub(\\([^)]*\\))?[[:space:]]+)?(super[[:space:]]+)?async fn ${function}\\b"
-  if ! rg -U "$trace_span_pattern" "$file" >/dev/null; then
+  if ! rg -U "$trace_span_pattern" "${hot_span_sources[@]}" >/dev/null; then
     echo "❌ logging guardrail violation: $file::$function must remain TRACE-only" >&2
     exit 1
   fi

scripts/codemods/split_multipart_usecase.py

@@ -0,0 +1,994 @@
+#!/usr/bin/env python3
+"""Split rustfs/src/app/multipart_usecase.rs into one file per operation.
+
+Preparatory refactor P3 of the s3s to rustfs-gateway migration. A pure code
+move: every top-level item, every method of `impl DefaultMultipartUsecase` and
+every test is assigned to a target file by the explicit tables below, and is
+moved verbatim with its attributes and doc comments. Nothing is renamed and no
+body changes, with these exceptions, each mechanical and listed here so a
+reviewer knows where the output differs from the input:
+
+  * the three test fixtures (`s3_op_total`, `build_request`, `make_usecase`)
+    gain `pub(super)` because they move to `test_support.rs`;
+  * the test module's `use` items become `pub(super) use` re-exports in
+    `test_support.rs` (names the parent already imports are not repeated), so
+    every per-operation test module reaches them with one glob;
+  * code-position `s3s::` paths in the new files become
+    `crate::app::storage_api::s3::`, the sanctioned s3s funnel, so the number
+    of files naming s3s (scripts/check_s3s_footprint.sh) does not grow. The
+    funnel gains two re-exports, `dto` and `xml`, in rustfs/src/app/storage_api.rs;
+  * the nextest filter that names the one test it pins follows the test to its
+    new module path.
+
+Layout produced (parent keeps the struct, its constructors and accessors, the
+shared helpers and the test fixtures):
+
+  multipart_usecase.rs                   shared state and helpers
+  multipart_usecase/test_support.rs      fixtures and test-only imports
+  multipart_usecase/{abort,create,list,upload_part,complete,upload_part_copy}.rs
+                                         one operation: its helpers, its
+                                         `impl DefaultMultipartUsecase` block,
+                                         its tests
+  multipart_usecase/upload_part/tests/body_read_tests.rs
+  multipart_usecase/upload_part_copy/tests/copy_admission_tests.rs
+                                         the two former test files, unchanged
+
+Usage:
+  scripts/codemods/split_multipart_usecase.py [--no-fmt] [--report]
+  scripts/codemods/split_multipart_usecase.py --self-test
+
+The input is the unsplit tree. A tree that is already split is refused, as is
+any top-level item, method or test without an assignment, any table entry that
+matches nothing, and any private item that a file other than its own refers
+to. Re-running therefore means restoring the pre-split files first:
+
+  git checkout <pre-split-commit> -- rustfs/src/app/multipart_usecase.rs \\
+      rustfs/src/app/multipart_usecase rustfs/src/app/storage_api.rs .config/nextest.toml
+  git clean -fd rustfs/src/app/multipart_usecase
+
+and then running this script. rustfmt formats the written files unless
+--no-fmt is given. --report prints the item to file assignment and the
+cross-file reference check and writes nothing.
+
+Exit status: 0 on success, 1 when the input cannot be split as asked (the
+message says why), 2 for a usage error.
+"""
+
+import argparse
+import re
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+APP = "rustfs/src/app"
+SOURCE = f"{APP}/multipart_usecase.rs"
+OUT_DIR = f"{APP}/multipart_usecase"
+STORAGE_API = f"{APP}/storage_api.rs"
+NEXTEST = ".config/nextest.toml"
+STRUCT = "DefaultMultipartUsecase"
+FUNNEL = "crate::app::storage_api::s3::"
+
+PARENT = "parent"
+SUPPORT = "test_support"
+OPS = ("abort", "create", "list", "upload_part", "complete", "upload_part_copy")
+OP_DOC = {
+    "abort": "AbortMultipartUpload.",
+    "create": "CreateMultipartUpload.",
+    "list": "ListMultipartUploads and ListParts.",
+    "upload_part": "UploadPart.",
+    "complete": "CompleteMultipartUpload.",
+    "upload_part_copy": "UploadPartCopy.",
+}
+
+# Test files that already existed, moved unchanged: (old path, operation).
+OLD_TEST_FILES = {
+    "body_read_tests": (f"{OUT_DIR}/tests/body_read_tests.rs", "upload_part"),
+    "copy_admission_tests": (f"{OUT_DIR}/tests/copy_admission_tests.rs", "upload_part_copy"),
+}
+
+# ---------------------------------------------------------------------------
+# Assignment tables: item name -> target file. Every name must be listed; the
+# script fails on a missing or an unused entry.
+# ---------------------------------------------------------------------------
+
+def assigned(target, names):
+    return {name: target for name in names.split()}
+
+
+# Top-level functions (everything that is not an import, the struct, its impl
+# or the test module). A helper stays in the parent when more than one
+# operation, or a cross-operation test, uses it.
+FUNCTIONS = {
+    **assigned(
+        PARENT,
+        """
+        merge_part_encryption_metadata multipart_plaintext_size multipart_part_numbers
+        multipart_max_total_object_size validate_table_catalog_object_mutation
+        internal_object_info_lookup_opts multipart_upload_api_error
+        """,
+    ),
+    **assigned(
+        "upload_part_copy",
+        "normalize_copy_source_version_id format_copy_source_version_id validate_copy_source_range_not_exceeds",
+    ),
+    **assigned(
+        "complete",
+        """
+        validate_complete_multipart_parts normalize_complete_multipart_parts complete_part_from_s3
+        has_complete_multipart_object_lock_headers validate_complete_multipart_checksum_type
+        quota_accounting_object_size encode_s3_path extract_request_scheme extract_request_host
+        build_complete_multipart_location
+        """,
+    ),
+    **assigned("create", "create_multipart_upload_metadata should_advertise_session_compression"),
+    **assigned(
+        "upload_part",
+        """
+        resolve_upload_part_size require_upload_part_size upload_part_body_read_timeout
+        passthrough_part_actual_size normalize_presigned_part_checksums
+        """,
+    ),
+}
+
+# Methods of `impl DefaultMultipartUsecase`.
+METHODS = {
+    **assigned(
+        PARENT,
+        "without_context from_global with_context with_context_and_concurrency_manager"
+        " bucket_metadata_sys object_store object_data_cache concurrency_manager",
+    ),
+    **assigned("abort", "execute_abort_multipart_upload"),
+    **assigned("complete", "execute_complete_multipart_upload execute_complete_multipart_upload_owned"),
+    **assigned("create", "execute_create_multipart_upload"),
+    **assigned("upload_part", "execute_upload_part"),
+    **assigned("list", "execute_list_multipart_uploads execute_list_parts"),
+    **assigned("upload_part_copy", "execute_upload_part_copy"),
+}
+
+# Items of `mod tests`. A test that drives more than one operation, and the
+# tests of the shared helpers, stay in the parent's own `mod tests`.
+TESTS = {
+    **assigned(SUPPORT, "s3_op_total build_request make_usecase"),
+    **assigned(
+        PARENT,
+        """
+        internal_object_info_lookup_opts_drops_http_preconditions
+        merge_part_encryption_metadata_keeps_source_metadata_unchanged
+        managed_multipart_roundtrip_preserves_session_nonce_between_parts
+        multipart_upload_rejects_reserved_directory_markers
+        multipart_max_total_object_size_reads_compatible_internal_metadata
+        multipart_upload_api_error_reports_malformed_upload_id_as_no_such_upload
+        multipart_handlers_reject_another_callers_denied_kms_upload
+        """,
+    ),
+    **assigned("abort", "execute_abort_multipart_upload_returns_internal_error_when_store_uninitialized"),
+    **assigned(
+        "create",
+        """
+        session_compression_is_advertised_only_for_non_passthrough_compressible_uploads
+        execute_create_multipart_upload_rejects_invalid_storage_class
+        create_multipart_upload_metadata_persists_dto_storage_class_without_raw_header
+        create_multipart_upload_metadata_keeps_user_and_system_namespaces_separate
+        create_multipart_rejects_ciphertext_replication_before_parts_are_staged
+        """,
+    ),
+    **assigned(
+        "list",
+        """
+        execute_list_multipart_uploads_returns_internal_error_when_store_uninitialized
+        execute_list_multipart_uploads_rejects_invalid_key_marker_before_store_lookup
+        execute_list_multipart_uploads_rejects_invalid_max_uploads_before_store_lookup
+        execute_list_parts_returns_internal_error_when_store_uninitialized
+        execute_list_parts_rejects_negative_part_number_marker_before_store_lookup
+        execute_list_parts_rejects_invalid_max_parts_before_store_lookup
+        """,
+    ),
+    **assigned(
+        "upload_part",
+        """
+        presigned_checksum_request
+        normalize_presigned_part_checksums_validates_body
+        normalize_presigned_part_checksums_accepts_sdk_crc32_default
+        normalize_presigned_part_checksums_rejects_ambiguous_values
+        normalize_presigned_part_checksums_preserves_headers_and_rejects_invalid_values
+        normalize_presigned_part_checksums_requires_verified_presign
+        resolve_upload_part_size_uses_normalized_logical_length
+        upload_part_length_contract_rejects_unknown_and_all_negative_lengths
+        upload_part_timeout_policy_preserves_disabled_and_capped_floor
+        execute_upload_part_rejects_missing_body
+        execute_upload_part_rejects_oversize_declared_part_before_reading_the_body
+        execute_upload_part_rejects_invalid_part_number_before_body_lookup
+        execute_upload_part_rejects_unknown_length_before_admission_or_body_polling
+        execute_upload_part_progress_timeout_releases_admission_and_leaves_no_part
+        execute_upload_part_rejects_when_foreground_write_admission_is_full
+        """,
+    ),
+    **assigned(
+        "upload_part_copy",
+        """
+        execute_upload_part_copy_returns_internal_error_when_store_uninitialized
+        execute_upload_part_copy_rejects_invalid_part_number_before_store_lookup
+        test_validate_copy_source_range_not_exceeds_returns_invalid_range_when_range_exceeds
+        test_validate_copy_source_range_not_exceeds_ok_when_range_valid
+        test_validate_copy_source_range_not_exceeds_ok_for_suffix_range
+        upload_part_copy_normalizes_and_formats_source_version_id
+        """,
+    ),
+    **assigned(
+        "complete",
+        """
+        upload_metadata_with_checksum_type headers_with_checksum_type
+        complete_multipart_checksum_type_matching_the_upload_is_accepted
+        complete_multipart_checksum_type_contradicting_the_upload_is_rejected
+        complete_multipart_checksum_type_without_both_sides_is_accepted
+        quota_accounting_uses_logical_size_when_available
+        quota_accounting_fails_closed_only_when_quota_is_configured
+        test_build_complete_multipart_location_uses_forwarded_proto_and_encodes_key
+        test_build_complete_multipart_location_falls_back_to_uri_authority_and_scheme
+        test_build_complete_multipart_location_returns_path_without_host
+        execute_complete_multipart_upload_rejects_missing_parts_payload
+        execute_complete_multipart_upload_rejects_missing_parts_list
+        compressed_complete_records_logical_quota_usage_and_overwrite_delta
+        authorized_replica_only_multipart_complete_preserves_persisted_anti_cascade_state
+        authorized_replica_only_multipart_complete_preserves_persisted_anti_cascade_state_inner
+        ssec_completion_and_replay_require_key_before_disclosing_plaintext_checksums
+        assert_multipart_etag_listings
+        kms_multipart_completion_and_replay_preserve_etag_in_versioned_listings
+        conditional_multipart_upload_if_none_match_returns_precondition_failed_when_object_exists
+        conditional_multipart_upload_if_none_match_returns_precondition_failed_when_object_exists_inner
+        concurrent_completions_share_durable_bucket_quota_reservations
+        multipart_completion_commits_after_quota_fleet_proof_loss_before_rename
+        data_movement_multipart_completion_has_zero_quota_growth
+        cancelled_multipart_completion_emits_once_and_replay_does_not_emit
+        rejected_empty_parts_preserve_existing_object_and_staging
+        execute_complete_multipart_upload_allows_duplicate_part_numbers_by_using_last_occurrence
+        execute_complete_multipart_upload_rejects_out_of_order_parts
+        validate_complete_multipart_parts_enforces_part_number_range
+        normalize_complete_multipart_parts_keeps_last_duplicate_part
+        execute_complete_multipart_upload_rejects_object_lock_headers
+        """,
+    ),
+}
+
+# Test module items that are not code to assign.
+TEST_MODS = ("body_read_tests", "copy_admission_tests")
+
+CODE, COMMENT, LITERAL = 0, 1, 2
+IDENT_CHARS = frozenset("abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_")
+CHAR_LITERAL = re.compile(r"'(?:\\(?:[nrt\\0'\"]|x[0-9a-fA-F]{2}|u\{[0-9a-fA-F_]{1,6}\})|[^\\'\n])'")
+RAW_STRING_START = re.compile(r'r(#*)"')
+# A relative path in code moved one module down: `super::object::..` named a sibling of
+# multipart_usecase, and from an operation file that is `super::super::object::..`.
+SUPER_PATH = re.compile(r"(?<![A-Za-z0-9_:])super::(?!\*)")
+S3S_PATH = re.compile(r"(?<![\"A-Za-z0-9_])s3s::")
+VIS = r"(?:pub(?:\([^)]*\))?\s+)?"
+FN_HEAD = re.compile(r"^" + VIS + r"(?:const\s+)?(?:async\s+)?(?:unsafe\s+)?fn\s+([A-Za-z_][A-Za-z0-9_]*)")
+STRUCT_HEAD = re.compile(r"^" + VIS + r"struct\s+([A-Za-z_][A-Za-z0-9_]*)")
+IMPL_HEAD = re.compile(r"^impl\s+([A-Za-z_][A-Za-z0-9_]*)\s*\{")
+MOD_BLOCK_HEAD = re.compile(r"^" + VIS + r"mod\s+([A-Za-z_][A-Za-z0-9_]*)\s*\{")
+MOD_DECL_HEAD = re.compile(r"^" + VIS + r"mod\s+([A-Za-z_][A-Za-z0-9_]*)\s*;")
+USE_HEAD = re.compile(r"^" + VIS + r"use\s")
+
+
+class SplitError(Exception):
+    """The input cannot be split as asked: exit 1."""
+
+
+# ---------------------------------------------------------------------------
+# Lexing: blank comments and literals so structure can be read from the rest.
+# ---------------------------------------------------------------------------
+
+
+def classify(text):
+    """Return one kind byte per character: CODE, COMMENT or LITERAL."""
+    kinds = bytearray(len(text))
+    i, n = 0, len(text)
+
+    def mark(start, end, kind):
+        kinds[start:end] = bytes([kind]) * (end - start)
+
+    while i < n:
+        ch = text[i]
+        if text.startswith("//", i):
+            end = text.find("\n", i)
+            end = n if end < 0 else end
+            mark(i, end, COMMENT)
+            i = end
+        elif text.startswith("/*", i):
+            depth, j = 1, i + 2
+            while j < n and depth:
+                if text.startswith("/*", j):
+                    depth, j = depth + 1, j + 2
+                elif text.startswith("*/", j):
+                    depth, j = depth - 1, j + 2
+                else:
+                    j += 1
+            mark(i, j, COMMENT)
+            i = j
+        elif ch == "r" and (i == 0 or text[i - 1] not in IDENT_CHARS or (text[i - 1] == "b" and (i == 1 or text[i - 2] not in IDENT_CHARS))):
+            raw = RAW_STRING_START.match(text, i)
+            if raw is None:
+                i += 1
+                continue
+            closing = '"' + raw.group(1)
+            end = text.find(closing, raw.end())
+            end = n if end < 0 else end + len(closing)
+            mark(i, end, LITERAL)
+            i = end
+        elif ch == '"':
+            j = i + 1
+            while j < n and text[j] != '"':
+                j += 2 if text[j] == "\\" else 1
+            mark(i, min(j + 1, n), LITERAL)
+            i = j + 1
+        elif ch == "'":
+            literal = CHAR_LITERAL.match(text, i)
+            if literal is None:  # a lifetime or a loop label
+                i += 1
+                continue
+            mark(i, literal.end(), LITERAL)
+            i = literal.end()
+        else:
+            i += 1
+    return kinds
+
+
+def code_only(text):
+    """`text` with comments and literals replaced by spaces, newlines kept."""
+    kinds = classify(text)
+    return "".join(ch if kinds[i] == CODE or ch == "\n" else " " for i, ch in enumerate(text))
+
+
+# ---------------------------------------------------------------------------
+# Item splitting. The input is rustfmt output, so an item is the run of lines
+# from its first attribute or comment to the line on which brackets balance
+# and the line ends in `}` or `;`.
+# ---------------------------------------------------------------------------
+
+
+class Item:
+    def __init__(self, lines, code_lines):
+        self.lines = lines  # original lines, leading comments and attributes included
+        self.code_lines = code_lines  # the same lines with comments and literals blanked
+        self.head = next((line.strip() for line in code_lines if line.strip() and not line.strip().startswith("#[")), "")
+        self.cfg_test = any(line.strip() == "#[cfg(test)]" for line in lines)
+
+    @property
+    def text(self):
+        return "\n".join(self.lines)
+
+    @property
+    def code(self):
+        return "\n".join(self.code_lines)
+
+    def name(self):
+        for pattern in (FN_HEAD, STRUCT_HEAD, IMPL_HEAD, MOD_BLOCK_HEAD, MOD_DECL_HEAD):
+            match = pattern.match(self.head)
+            if match:
+                return match.group(1)
+        return None
+
+    def is_use(self):
+        return bool(USE_HEAD.match(self.head))
+
+
+def split_items(lines, code_lines, what):
+    """Cut `lines` into items; blank lines separate them."""
+    items, i, n = [], 0, len(lines)
+    while i < n:
+        if not lines[i].strip():
+            i += 1
+            continue
+        start, balance, j = i, 0, i
+        while j < n:
+            stripped = code_lines[j].rstrip()
+            balance += sum(stripped.count(c) for c in "([{") - sum(stripped.count(c) for c in ")]}")
+            if balance == 0 and stripped.endswith(("}", ";")):
+                break
+            j += 1
+        if j >= n:
+            raise SplitError(f"{what}: an item starting at line {start + 1} never closes")
+        items.append(Item(lines[start : j + 1], code_lines[start : j + 1]))
+        i = j + 1
+    return items
+
+
+def canonical(items, separator):
+    return separator.join(item.text for item in items)
+
+
+# ---------------------------------------------------------------------------
+# Use trees.
+# ---------------------------------------------------------------------------
+
+
+def split_top(inner):
+    parts, depth, current = [], 0, []
+    for ch in inner:
+        if ch == "{":
+            depth += 1
+        elif ch == "}":
+            depth -= 1
+        if ch == "," and depth == 0:
+            parts.append("".join(current))
+            current = []
+        else:
+            current.append(ch)
+    parts.append("".join(current))
+    return [p.strip() for p in parts if p.strip()]
+
+
+def flatten_use(tree, prefix=""):
+    """[(full path, bound name)] for a use tree; a glob binds `*`, `as _` binds `_`."""
+    tree = tree.strip()
+    brace = tree.find("{")
+    if brace >= 0:
+        head = tree[:brace].strip()
+        inner = tree[brace + 1 : tree.rindex("}")]
+        base = prefix + head
+        found = []
+        for part in split_top(inner):
+            found.extend(flatten_use(part, base))
+        return found
+    alias = None
+    if " as " in tree:
+        tree, alias = (side.strip() for side in tree.split(" as ", 1))
+    path = prefix + tree
+    last = path.split("::")[-1]
+    if last == "self":
+        path = path[: -len("::self")]
+        last = path.split("::")[-1]
+    return [(path, alias if alias else last)]
+
+
+def use_paths(item):
+    body = re.sub(r"^\s*(?:#\[[^\n]*\]\s*)*", "", item.code, flags=re.S)
+    body = re.sub(r"^" + VIS + r"use\s+", "", body.strip())
+    body = body.rstrip().rstrip(";")
+    return flatten_use(body)
+
+
+# ---------------------------------------------------------------------------
+# The parsed input.
+# ---------------------------------------------------------------------------
+
+
+class Parsed:
+    pass
+
+
+def parse(text):
+    lines = text.split("\n")
+    if lines and lines[-1] == "":
+        lines.pop()
+    code_lines = code_only("\n".join(lines)).split("\n")
+
+    # Leading licence and `//!` lines are the file preamble.
+    first = 0
+    while first < len(lines) and (lines[first].startswith("//") or not lines[first].strip()):
+        first += 1
+    preamble = lines[:first]
+    items = split_items(lines[first:], code_lines[first:], "multipart_usecase.rs")
+
+    parsed = Parsed()
+    parsed.preamble = preamble
+    parsed.imports, parsed.functions, parsed.body = [], [], []
+    parsed.struct = parsed.impl = parsed.tests = None
+    for item in items:
+        if not item.is_use():
+            parsed.body.append(item)
+        name = item.name()
+        if item.is_use():
+            if parsed.functions or parsed.struct or parsed.impl:
+                raise SplitError("a `use` item follows code: imports are expected first")
+            parsed.imports.append(item)
+        elif STRUCT_HEAD.match(item.head) and name == STRUCT:
+            parsed.struct = item
+        elif IMPL_HEAD.match(item.head) and name == STRUCT:
+            if parsed.impl is not None:
+                raise SplitError(f"more than one `impl {STRUCT}` block")
+            parsed.impl = item
+        elif MOD_BLOCK_HEAD.match(item.head) and name == "tests":
+            parsed.tests = item
+        elif FN_HEAD.match(item.head):
+            item.after_impl = parsed.impl is not None
+            parsed.functions.append(item)
+        elif MOD_DECL_HEAD.match(item.head):
+            raise SplitError(f"{SOURCE} already declares {item.head!r}: it is split already; restore the pre-split files first")
+        else:
+            raise SplitError(f"unrecognised top-level item: {item.head!r}")
+    for what, value in (("struct", parsed.struct), ("impl", parsed.impl), ("tests module", parsed.tests)):
+        if value is None:
+            raise SplitError(f"the {what} was not found; is the input already split?")
+
+    # Spacing must be the canonical one, or the output would not be a move.
+    rebuilt = "\n".join(preamble) + "\n" + "\n".join(i.text for i in parsed.imports) + "\n\n"
+    rebuilt += canonical(parsed.body, "\n\n")
+    if "\n".join(lines) != "\n".join(rebuilt.split("\n")):
+        raise SplitError("the item layout is not the one this script reassembles (spacing differs from rustfmt's)")
+
+    parsed.impl_methods = body_items(parsed.impl, "impl")
+    parsed.test_items = body_items(parsed.tests, "mod tests")
+    return parsed
+
+
+def body_items(block, what):
+    """Items between the opening line and the closing brace of `block`."""
+    opening = next(i for i, line in enumerate(block.code_lines) if line.strip() and not line.strip().startswith("#["))
+    inner, inner_code = block.lines[opening + 1 : -1], block.code_lines[opening + 1 : -1]
+    items = split_items(inner, inner_code, what)
+    return items
+
+
+# ---------------------------------------------------------------------------
+# Assignment and checks.
+# ---------------------------------------------------------------------------
+
+
+def assign(items, table, what):
+    placed = {}
+    seen = set()
+    for item in items:
+        name = item.name()
+        if name is None:
+            raise SplitError(f"{what}: cannot name the item {item.head!r}")
+        if name in seen:
+            raise SplitError(f"{what}: two items are named {name}")
+        seen.add(name)
+        target = table.get(name)
+        if target is None:
+            raise SplitError(f"{what}: no assignment for {name} ({item.head[:60]!r})")
+        placed[name] = (target, item)
+    unused = sorted(set(table) - seen)
+    if unused:
+        raise SplitError(f"{what}: assignments that match nothing: {', '.join(unused)}")
+    return placed
+
+
+def check_targets(table, what, allowed):
+    for name, target in table.items():
+        if target not in allowed:
+            raise SplitError(f"{what}: {name} is assigned to unknown file {target!r}")
+
+
+def word(name):
+    return re.compile(r"(?<![A-Za-z0-9_])" + re.escape(name) + r"(?![A-Za-z0-9_])")
+
+
+def referenced_from(name, texts, own):
+    """Files other than `own` whose code mentions `name`."""
+    pattern = word(name)
+    return sorted(f for f, code in texts.items() if f != own and pattern.search(code))
+
+
+def plan(parsed):
+    """Decide the content of every output file; returns (pieces, report)."""
+    check_targets(FUNCTIONS, "functions", {PARENT, *OPS})
+    check_targets(METHODS, "methods", {PARENT, *OPS})
+    check_targets(TESTS, "tests", {PARENT, SUPPORT, *OPS})
+    functions = assign(parsed.functions, FUNCTIONS, "functions")
+    methods = assign(parsed.impl_methods, METHODS, "methods")
+
+    test_code_items = []
+    test_uses, test_mods, test_glob = [], [], None
+    for item in parsed.test_items:
+        if item.is_use():
+            if item.head.replace(" ", "") == "usesuper::*;":
+                test_glob = item
+            else:
+                test_uses.append(item)
+        elif MOD_DECL_HEAD.match(item.head):
+            if item.name() not in TEST_MODS:
+                raise SplitError(f"mod tests: unexpected module declaration {item.head!r}")
+            test_mods.append(item)
+        else:
+            test_code_items.append(item)
+    if test_glob is None:
+        raise SplitError("mod tests does not start with `use super::*;`")
+    if sorted(m.name() for m in test_mods) != sorted(TEST_MODS):
+        raise SplitError("mod tests: the declared test modules differ from TEST_MODS")
+    tests = assign(test_code_items, TESTS, "tests")
+
+    for placed in (functions, methods):
+        for target, item in placed.values():
+            if target != PARENT:
+                push_down(item)
+
+    # Piece lists per target file.
+    pieces = {f: {"functions": [], "methods": [], "tests": []} for f in (PARENT, SUPPORT, *OPS)}
+    for _, (target, item) in functions.items():
+        pieces[target]["functions"].append(item)
+    for _, (target, item) in methods.items():
+        pieces[target]["methods"].append(item)
+    for _, (target, item) in tests.items():
+        pieces[target]["tests"].append(item)
+    if pieces[SUPPORT]["functions"] or pieces[SUPPORT]["methods"]:
+        raise SplitError("test_support holds tests-module fixtures only")
+    pieces["_uses"] = test_uses
+    pieces["_mods"] = test_mods
+    return pieces, (functions, methods, tests)
+
+
+def per_file_code(parsed, pieces, extra):
+    """Code of each output file for the reference check (parent keeps imports)."""
+    texts = {}
+    for target in (PARENT, SUPPORT, *OPS):
+        parts = []
+        if target == PARENT:
+            parts += [i.code for i in parsed.imports] + [parsed.struct.code]
+        parts += [i.code for i in pieces[target]["functions"]]
+        parts += [i.code for i in pieces[target]["methods"]]
+        parts += [i.code for i in pieces[target]["tests"]]
+        texts[target] = "\n".join(parts)
+    for target, code in extra.items():
+        texts[target] += "\n" + code
+    return texts
+
+
+def reference_check(pieces, texts):
+    """A private item moved out of the parent must stay inside its own file.
+
+    Items kept in the parent and fixtures in test_support are visible to every
+    child through the globs, so only items placed in an operation file are
+    checked. A name that is also a local binding elsewhere would show up here as
+    a false alarm; the message names the files so the table can be adjusted.
+    """
+    problems = []
+    for group in ("functions", "methods", "tests"):
+        for target in OPS:
+            for item in pieces[target][group]:
+                name = item.name()
+                if name is None or not FN_HEAD.match(item.head):
+                    continue
+                if re.match(r"^pub(\(|\s)", item.head) and group == "methods":
+                    continue  # a public method is reached through the type
+                users = referenced_from(name, texts, target)
+                if users:
+                    problems.append(f"{name} is placed in {target} but also mentioned in {', '.join(users)}")
+    return problems
+
+
+# ---------------------------------------------------------------------------
+# Rendering.
+# ---------------------------------------------------------------------------
+
+
+def rewrite_code(text, pattern, replacement):
+    """Replace `pattern` matches that sit in code, never in comments or literals."""
+    kinds = classify(text)
+    out, last = [], 0
+    for match in pattern.finditer(text):
+        if kinds[match.start()] != CODE:
+            continue
+        out.append(text[last : match.start()])
+        out.append(replacement)
+        last = match.end()
+    out.append(text[last:])
+    return "".join(out)
+
+
+def rewrite_s3s(text):
+    """Send code-position `s3s::` paths through the sanctioned funnel."""
+    return rewrite_code(text, S3S_PATH, FUNNEL)
+
+
+def push_down(item):
+    """Make the relative paths of an item that leaves the parent module point where they did."""
+    text = rewrite_code(item.text, SUPER_PATH, "super::super::")
+    item.lines = text.split("\n")
+    item.code_lines = code_only(text).split("\n")
+
+
+def licence(preamble):
+    block = []
+    for line in preamble:
+        if not line.startswith("//") or line.startswith("//!"):
+            break
+        block.append(line)
+    return "\n".join(block)
+
+
+def doc_line(preamble):
+    return next(line for line in preamble if line.startswith("//!"))
+
+
+def join_items(items):
+    return "\n\n".join(item.text for item in items)
+
+
+def render_operation(op, pieces, header, tests_extra_mods):
+    out = [header, f"//! {OP_DOC[op]}", "", "use super::*;", ""]
+    body = []
+    if pieces[op]["functions"]:
+        body.append(join_items(pieces[op]["functions"]))
+    if pieces[op]["methods"]:
+        body.append(f"impl {STRUCT} {{\n{join_items(pieces[op]['methods'])}\n}}")
+    if pieces[op]["tests"] or tests_extra_mods:
+        inner = [join_items(pieces[op]["tests"])] if pieces[op]["tests"] else []
+        inner += [f"    mod {name};" for name in tests_extra_mods]
+        body.append(
+            "#[cfg(test)]\nmod tests {\n    use super::super::test_support::*;\n    use super::*;\n\n"
+            + "\n\n".join(inner)
+            + "\n}"
+        )
+    out.append("\n\n".join(body))
+    return rewrite_s3s("\n".join(out) + "\n")
+
+
+def render_test_support(parsed, pieces, header, parent_bound):
+    out = [header, "//! Fixtures and test-only imports shared by the multipart use-case tests.", "", "use super::*;"]
+    reexports, notes = [], []
+    for item in pieces["_uses"]:
+        for path, name in use_paths(item):
+            if name in parent_bound:
+                notes.append((name, parent_bound[name], path))
+                continue
+            reexports.append(f"pub(super) use {path};")
+    out.extend(reexports)
+    out.append("")
+    fixtures = []
+    for item in pieces[SUPPORT]["tests"]:
+        lines = list(item.lines)
+        for index, line in enumerate(lines):
+            if FN_HEAD.match(line.strip()) and not line.startswith("    #"):
+                lines[index] = re.sub(r"^(\s*)", r"\1pub(super) ", line, count=1)
+                break
+        fixtures.append("\n".join(line[4:] if line.startswith("    ") else line for line in lines))
+    out.append("\n\n".join(fixtures))
+    return rewrite_s3s("\n".join(out) + "\n"), notes
+
+
+def render_parent(parsed, pieces, header):
+    out = [header, "\n".join(line for line in parsed.preamble if line.startswith("//!")), ""]
+    out.append("\n".join(i.text for i in parsed.imports))
+    out.append("")
+    mods = [f"mod {op};" for op in sorted(OPS + (SUPPORT,))]
+    mods = [("#[cfg(test)]\n" + m) if SUPPORT in m else m for m in mods]
+    out.append("\n".join(mods))
+    out.append("")
+    before = [i for i in pieces[PARENT]["functions"] if not i.after_impl]
+    after = [i for i in pieces[PARENT]["functions"] if i.after_impl]
+    chunks = [join_items(before)] if before else []
+    chunks.append(
+        f"{parsed.struct.text}\n\n{parsed.impl.lines[0]}\n{join_items(pieces[PARENT]['methods'])}\n}}"
+    )
+    if after:
+        chunks.append(join_items(after))
+    if pieces[PARENT]["tests"]:
+        chunks.append(
+            rewrite_s3s(
+                "#[cfg(test)]\nmod tests {\n    use super::test_support::*;\n    use super::*;\n\n"
+                + join_items(pieces[PARENT]["tests"])
+                + "\n}"
+            )
+        )
+    out.append("\n\n".join(chunks))
+    return "\n".join(out) + "\n"
+
+
+# ---------------------------------------------------------------------------
+# Driver.
+# ---------------------------------------------------------------------------
+
+
+def bound_names(imports):
+    bound = {}
+    for item in imports:
+        for path, name in use_paths(item):
+            if name not in ("*", "_"):
+                bound[name] = path
+    return bound
+
+
+def build(root):
+    source_path = root / SOURCE
+    if not source_path.is_file():
+        raise SplitError(f"{SOURCE} is missing")
+    text = source_path.read_text(encoding="utf-8")
+    parsed = parse(text)
+    pieces, (functions, methods, tests) = plan(parsed)
+
+    extra = {}
+    old_texts = {}
+    for name, (path, op) in OLD_TEST_FILES.items():
+        file = root / path
+        if not file.is_file():
+            raise SplitError(f"{path} is missing")
+        old_texts[name] = file.read_text(encoding="utf-8")
+        extra[op] = extra.get(op, "") + "\n" + code_only(old_texts[name])
+    texts = per_file_code(parsed, pieces, extra)
+    problems = reference_check(pieces, texts)
+    if problems:
+        raise SplitError("cross-file references:\n  " + "\n  ".join(problems))
+
+    header = licence(parsed.preamble) + "\n"
+    files = {}
+    parent_bound = bound_names(parsed.imports)
+    support, notes = render_test_support(parsed, pieces, header, parent_bound)
+    files[f"{OUT_DIR}/test_support.rs"] = support
+    for op in OPS:
+        mods = [name for name, (_, owner) in OLD_TEST_FILES.items() if owner == op]
+        files[f"{OUT_DIR}/{op}.rs"] = render_operation(op, pieces, header, mods)
+    for name, (path, op) in OLD_TEST_FILES.items():
+        files[f"{OUT_DIR}/{op}/tests/{name}.rs"] = old_texts[name]
+    files[SOURCE] = render_parent(parsed, pieces, header)
+    return parsed, pieces, files, notes, (functions, methods, tests)
+
+
+def check_footprint(files):
+    for path, text in files.items():
+        if path == SOURCE:
+            continue
+        for number, line in enumerate(text.split("\n"), 1):
+            if S3S_PATH.search(line):
+                raise SplitError(f"{path}:{number} still names s3s directly: {line.strip()}")
+    return None
+
+
+FUNNEL_NAMES = {"dto", "xml", "S3", "Body", "S3Request", "S3ErrorCode", "S3Response", "S3Error", "S3Result"}
+
+
+def check_funnel_names(parsed_sources, files):
+    seen = set()
+    original = "\n".join(parsed_sources)
+    for match in S3S_PATH.finditer(original):
+        tail = re.match(r"[A-Za-z_][A-Za-z0-9_]*", original[match.end() :])
+        if tail:
+            seen.add(tail.group(0))
+    unknown = sorted(seen - FUNNEL_NAMES)
+    if unknown:
+        raise SplitError(f"s3s paths with no funnel re-export: {', '.join(unknown)}")
+
+
+def edit_storage_api(text):
+    if "pub(crate) use s3s::dto;" in text:
+        raise SplitError(f"{STORAGE_API} already re-exports s3s::dto")
+    anchor = "    #[cfg(test)]\n    pub(crate) use s3s::dto::{GetObjectInput, HeadObjectInput, ListObjectsV2Input, ListObjectsV2Output};\n"
+    if text.count(anchor) != 1:
+        raise SplitError(f"{STORAGE_API}: the s3 module's dto re-export anchor was not found exactly once")
+    text = text.replace(anchor, "    pub(crate) use s3s::dto;\n" + anchor)
+    anchor = "    #[cfg(test)]\n    pub(crate) use s3s::xml::{Serialize as XmlSerialize, Serializer as XmlSerializer};\n"
+    if text.count(anchor) != 1:
+        raise SplitError(f"{STORAGE_API}: the s3 module's xml re-export anchor was not found exactly once")
+    return text.replace(anchor, "    #[cfg(test)]\n    pub(crate) use s3s::xml;\n" + anchor)
+
+
+def edit_nextest(text, tests):
+    """Point the filter that pins one test at the module the test now lives in."""
+    replaced = 0
+    for name, (target, _) in tests.items():
+        needle = f"app::multipart_usecase::tests::{name}"
+        if needle in text:
+            moved = "app::multipart_usecase::tests" if target == PARENT else f"app::multipart_usecase::{target}::tests"
+            replaced += text.count(needle)
+            text = text.replace(needle, f"{moved}::{name}")
+    if replaced == 0:
+        raise SplitError(f"{NEXTEST} names no multipart_usecase test: has the filter moved?")
+    return text
+
+
+def run(root, fmt, report):
+    parsed, pieces, files, notes, (functions, methods, tests) = build(root)
+    check_funnel_names([i.text for i in parsed.functions + parsed.impl_methods + parsed.test_items], files)
+    check_footprint(files)
+    if report:
+        print_report(parsed, pieces, files, notes, (functions, methods, tests))
+        return 0
+
+    storage = edit_storage_api((root / STORAGE_API).read_text(encoding="utf-8"))
+    nextest = edit_nextest((root / NEXTEST).read_text(encoding="utf-8"), tests)
+
+    for path, text in files.items():
+        target = root / path
+        target.parent.mkdir(parents=True, exist_ok=True)
+        target.write_text(text, encoding="utf-8")
+    for name, (path, op) in OLD_TEST_FILES.items():
+        (root / path).unlink()
+    old_dir = root / OUT_DIR / "tests"
+    if old_dir.is_dir() and not any(old_dir.iterdir()):
+        old_dir.rmdir()
+    (root / STORAGE_API).write_text(storage, encoding="utf-8")
+    (root / NEXTEST).write_text(nextest, encoding="utf-8")
+
+    if fmt:
+        written = [str(root / p) for p in files if p != SOURCE] + [str(root / SOURCE), str(root / STORAGE_API)]
+        result = subprocess.run(["rustfmt", "--edition", "2024", *written], check=False, cwd=root)
+        if result.returncode != 0:
+            raise SplitError("rustfmt failed on the written files")
+    summary(files, pieces)
+    return 0
+
+
+def print_report(parsed, pieces, files, notes, assigned):
+    for target in (PARENT, SUPPORT, *OPS):
+        counts = {k: len(pieces[target][k]) for k in ("functions", "methods", "tests")}
+        print(f"{target}: {counts}")
+    for name, parent_path, test_path in notes:
+        print(f"note: test import {name} is already bound in the parent ({parent_path}); test path {test_path}")
+    for path, text in sorted(files.items()):
+        print(f"{path}: {len(text.splitlines())} lines")
+
+
+def summary(files, pieces):
+    for path, text in sorted(files.items()):
+        print(f"wrote {path}: {len(text.splitlines())} lines")
+    for target in (PARENT, SUPPORT, *OPS):
+        fns = len(pieces[target]["functions"]) + len(pieces[target]["methods"])
+        tests = sum(1 for i in pieces[target]["tests"] if re.search(r"#\[(tokio::)?test", i.text))
+        print(f"{target}: {fns} production fns, {tests} tests")
+
+
+def main(argv):
+    parser = argparse.ArgumentParser(description=__doc__.split("\n\n", 1)[0])
+    parser.add_argument("--no-fmt", action="store_true", help="do not run rustfmt on the written files")
+    parser.add_argument("--report", action="store_true", help="print the assignment and write nothing")
+    parser.add_argument("--self-test", action="store_true", help="run the script's own tests")
+    args = parser.parse_args(argv)
+    if args.self_test:
+        suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTest)
+        return 0 if unittest.TextTestRunner(verbosity=1).run(suite).wasSuccessful() else 1
+    try:
+        return run(ROOT, not args.no_fmt, args.report)
+    except SplitError as error:
+        print(f"split_multipart_usecase: {error}", file=sys.stderr)
+        return 1
+
+
+class SelfTest(unittest.TestCase):
+    SAMPLE = "\n".join(
+        [
+            "// licence",
+            "",
+            "//! doc",
+            "",
+            "use a::b;",
+            "",
+            "/// doc",
+            "fn one(",
+            "    x: u8,",
+            ") -> u8 {",
+            '    let s = "}";',
+            "    x",
+            "}",
+            "",
+            "#[cfg(test)]",
+            "mod tests {",
+            "    use super::*;",
+            "",
+            "    #[test]",
+            "    fn t() {}",
+            "}",
+        ]
+    )
+
+    def test_items_split_on_balance_not_on_braces_in_strings(self):
+        lines = self.SAMPLE.split("\n")[6:]
+        items = split_items(lines, code_only("\n".join(lines)).split("\n"), "sample")
+        self.assertEqual([i.name() for i in items], ["one", "tests"])
+        self.assertTrue(items[1].cfg_test)
+
+    def test_use_tree_flattening(self):
+        paths = flatten_use("a::{b, c::{d as e, f}, g::*}")
+        self.assertEqual(paths, [("a::b", "b"), ("a::c::d", "e"), ("a::c::f", "f"), ("a::g::*", "*")])
+
+    def test_s3s_rewrite_skips_strings_and_identifiers(self):
+        text = 'let a = s3s::dto::X; let b = "s3s::dto"; let c = rs3s::y;'
+        self.assertEqual(rewrite_s3s(text), f'let a = {FUNNEL}dto::X; let b = "s3s::dto"; let c = rs3s::y;')
+
+    def test_unassigned_item_fails(self):
+        lines = self.SAMPLE.split("\n")[6:13]
+        items = split_items(lines, code_only("\n".join(lines)).split("\n"), "sample")
+        with self.assertRaises(SplitError):
+            assign(items, {}, "functions")
+        with self.assertRaises(SplitError):
+            assign(items, {"one": PARENT, "ghost": PARENT}, "functions")
+
+    def test_unclosed_item_fails(self):
+        with self.assertRaises(SplitError):
+            split_items(["fn a() {"], ["fn a() {"], "sample")
+
+
+if __name__ == "__main__":
+    sys.exit(main(sys.argv[1:]))

scripts/guard_source_set.sh

@@ -0,0 +1,51 @@
+#!/usr/bin/env bash
+#
+# Source-set helper for text guards that must survive a mechanical file split.
+# Sourced by check_architecture_migration_rules.sh and check_logging_guardrails.sh;
+# contract tests: scripts/test_guard_source_set.sh.
+#
+# A guard that greps one named file stops seeing the code it guards once that
+# file is split into a sibling directory. guard_source_set resolves the set of
+# files such a guard is about instead:
+#
+#   guard_source_set <root> <required-file> [<split-dir>...]
+#
+# On success it sets the global array GUARD_SOURCES to <required-file> followed
+# by every `.rs` file under each <split-dir> that exists, in byte order of path
+# (LC_ALL=C), all relative to <root>. An absent <split-dir> contributes nothing.
+# It returns 1 with a message on stderr when <required-file> is not a regular
+# file (a guard whose input is missing must fail, not pass) or when a
+# <split-dir> exists but is not a directory. On failure GUARD_SOURCES still
+# holds <required-file>, so a caller that records the failure and continues
+# hands the missing path to its search tool rather than nothing.
+
+guard_source_set() {
+  local root="$1" required="$2" split_dir path listing
+  shift 2
+  GUARD_SOURCES=("$required")
+
+  if [[ ! -f "${root}/${required}" ]]; then
+    printf 'guard source %s is missing\n' "$required" >&2
+    return 1
+  fi
+
+  for split_dir in "$@"; do
+    if [[ ! -e "${root}/${split_dir}" ]]; then
+      continue
+    fi
+    if [[ ! -d "${root}/${split_dir}" ]]; then
+      printf 'guard split path %s is not a directory\n' "$split_dir" >&2
+      return 1
+    fi
+    if ! listing="$(set -o pipefail; cd "$root" && find "$split_dir" -type f -name '*.rs' | LC_ALL=C sort)"; then
+      printf 'guard split directory %s cannot be listed\n' "$split_dir" >&2
+      return 1
+    fi
+    while IFS= read -r path; do
+      if [[ -n "$path" ]]; then
+        GUARD_SOURCES+=("$path")
+      fi
+    done <<<"$listing"
+  done
+  return 0
+}

scripts/s3-tests/README.md

@@ -509,7 +509,9 @@ pass/fail table in the job summary.
   incomplete parameterized cases.
 - `api_coverage.py` — quantifies S3 API surface coverage by comparing the
   s3s `S3` trait (at the revision pinned in Cargo.toml) against the methods
-  RustFS overrides in `impl S3 for FS`:
+  RustFS overrides in `impl S3 for FS`, found exactly once across
+  `rustfs/src/storage/ecfs.rs` and `rustfs/src/storage/legacy_edge/`
+  (contract tests: `test_api_coverage.py`, run by `make script-tests`):
 
   ```bash
   python3 scripts/s3-tests/api_coverage.py            # summary + missing ops

scripts/s3-tests/api_coverage.py

@@ -17,7 +17,8 @@
 
 Compares the operations defined by the s3s `S3` trait (at the revision pinned
 in the workspace Cargo.toml) against the methods RustFS overrides in
-`impl S3 for FS` (rustfs/src/storage/ecfs.rs). Trait methods that are not
+`impl S3 for FS` (rustfs/src/storage/ecfs.rs, or the rustfs/src/storage/legacy_edge/
+directory a mechanical split of that file moves code into). Trait methods that are not
 overridden fall back to the s3s default implementation, which returns
 NotImplemented — so "overridden" is a faithful proxy for "implemented".
 
@@ -36,6 +37,8 @@ import sys
 import tempfile
 
 ECFS_PATH = "rustfs/src/storage/ecfs.rs"
+ECFS_SPLIT_DIR = "rustfs/src/storage/legacy_edge"
+IMPL_MARKER = "impl S3 for FS {"
 TRAIT_RELPATH = "crates/s3s/src/s3_trait.rs"
 
 
@@ -97,14 +100,28 @@ def trait_operations(trait_source: pathlib.Path) -> dict[str, str]:
     return ops
 
 
-def implemented_operations(ecfs_source: pathlib.Path) -> set[str]:
-    src = ecfs_source.read_text(encoding="utf-8")
-    marker = "impl S3 for FS {"
-    start = src.find(marker)
-    if start < 0:
-        raise RuntimeError(f"`{marker}` not found in {ecfs_source}")
-    block = extract_block(src, start)
-    return set(re.findall(r"async fn (\w+)", block))
+def ecfs_sources(root: pathlib.Path) -> list[pathlib.Path]:
+    """Return ecfs.rs (required) followed by every .rs file under the split directory, in path order."""
+    required = root / ECFS_PATH
+    if not required.is_file():
+        raise RuntimeError(f"{required} is missing")
+    split_dir = root / ECFS_SPLIT_DIR
+    if not split_dir.exists():
+        return [required]
+    if not split_dir.is_dir():
+        raise RuntimeError(f"{split_dir} is not a directory")
+    return [required, *sorted(split_dir.rglob("*.rs"), key=lambda path: path.relative_to(root).as_posix())]
+
+
+def implemented_operations(root: pathlib.Path) -> set[str]:
+    """Return the methods of the one `impl S3 for FS` block across the ECFS source set."""
+    blocks = []
+    for source in ecfs_sources(root):
+        src = source.read_text(encoding="utf-8")
+        blocks += [extract_block(src, match.start()) for match in re.finditer(re.escape(IMPL_MARKER), src)]
+    if len(blocks) != 1:
+        raise RuntimeError(f"`{IMPL_MARKER}` must occur exactly once under {ECFS_PATH} and {ECFS_SPLIT_DIR}, found {len(blocks)}")
+    return set(re.findall(r"async fn (\w+)", blocks[0]))
 
 
 def main() -> int:
@@ -122,7 +139,7 @@ def main() -> int:
     trait_source = args.s3s_trait or locate_trait_source(git_url, rev)
 
     all_ops = trait_operations(trait_source)
-    impl_ops = implemented_operations(root / ECFS_PATH)
+    impl_ops = implemented_operations(root)
 
     stale = sorted(impl_ops - all_ops.keys())
     implemented = sorted(op for op in all_ops if op in impl_ops)

scripts/s3-tests/reference_parity.py

@@ -0,0 +1,329 @@
+#!/usr/bin/env python3
+# 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.
+"""Plan and judge the s3-tests reference parity lane.
+
+Subcommands:
+  targets    validate the target list and print it as a JSON matrix
+  export-env validate one target's extra environment and print KEY=VALUE lines
+  failed     print the failed test names of one run, in implemented_tests.txt form
+  compare    compare every candidate against the reference, per pytest node id
+
+`compare` reads <reports>/<target>/selected-nodeids.txt, junit.xml and the
+optional rerun-junit.xml of every target. Each node id is reduced to pass /
+fail / skip / missing; a case that failed and then passed its single rerun is a
+pass and is listed as flaky. A tolerated entry must match a real difference.
+
+Exit codes: 0 no untolerated difference and no stale entry; 1 a difference, a
+stale tolerated entry or an incomplete run; 2 missing or malformed input.
+"""
+
+import argparse
+import json
+from pathlib import Path
+import re
+import sys
+import tomllib
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+import junit_diff  # noqa: E402  (sibling script, not a package)
+
+PASS, FAIL, SKIP, MISSING = junit_diff.PASS, junit_diff.FAIL, junit_diff.SKIP, junit_diff.MISSING
+NAME = re.compile(r"[a-z][a-z0-9-]{0,39}")
+ENV_KEY = re.compile(r"RUSTFS_[A-Z0-9_]+")
+ENV_VALUE = re.compile(r"[A-Za-z0-9._,:/=+-]*")
+ISSUE = re.compile(r"https://github\.com/rustfs/[A-Za-z0-9_.-]+/issues/[1-9][0-9]*")
+NODEID = re.compile(r"[A-Za-z0-9_/]+\.py(::[A-Za-z0-9_]+)+(\[[^\]\s]*\])?")
+TEST_FILE_PREFIX = "s3tests/functional/test_s3.py::"
+BINARIES = ("reference", "candidate")
+
+
+class InputError(Exception):
+    """Input that cannot be trusted; maps to exit code 2."""
+
+
+# --- target list -----------------------------------------------------------
+
+def load_targets(path):
+    try:
+        data = tomllib.loads(Path(path).read_text(encoding="utf-8"))
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: target list is missing") from exc
+    except (OSError, UnicodeDecodeError, tomllib.TOMLDecodeError) as exc:
+        raise InputError(f"{path}: cannot read target list: {exc}") from exc
+    if data.get("schema") != 1 or type(data.get("schema")) is not int or set(data) != {"schema", "targets"}:
+        raise InputError(f"{path}: unsupported target list schema")
+    return validate_targets(data["targets"], str(path))
+
+
+def validate_targets(targets, where="targets"):
+    if not isinstance(targets, list) or len(targets) < 2:
+        raise InputError(f"{where}: need one reference and at least one candidate")
+    result, names = [], set()
+    for item in targets:
+        if not isinstance(item, dict) or not set(item) <= {"name", "binary", "env"}:
+            raise InputError(f"{where}: invalid target entry {item!r}")
+        name, binary, env = item.get("name"), item.get("binary"), item.get("env", {})
+        if not isinstance(name, str) or not NAME.fullmatch(name) or name in names:
+            raise InputError(f"{where}: invalid or duplicate target name {name!r}")
+        if binary not in BINARIES:
+            raise InputError(f"{where}: target {name} has unknown binary {binary!r}")
+        validate_env(env, f"{where}: target {name}")
+        names.add(name)
+        result.append({"name": name, "binary": binary, "env": dict(env)})
+    if [t["binary"] for t in result].count("reference") != 1:
+        raise InputError(f"{where}: exactly one target must use the reference binary")
+    return result
+
+
+def validate_env(env, where):
+    if not isinstance(env, dict):
+        raise InputError(f"{where}: env must be a table")
+    for key, value in env.items():
+        if not isinstance(key, str) or not ENV_KEY.fullmatch(key):
+            raise InputError(f"{where}: env key {key!r} must match RUSTFS_[A-Z0-9_]+")
+        if not isinstance(value, str) or not ENV_VALUE.fullmatch(value):
+            raise InputError(f"{where}: env value for {key} has unsupported characters")
+    return env
+
+
+# --- node ids and reports --------------------------------------------------
+
+def nodeid_to_case(nodeid):
+    """Map a pytest node id to the classname::name id that junit_diff uses."""
+    if not NODEID.fullmatch(nodeid):
+        raise InputError(f"invalid pytest node id {nodeid!r}")
+    path, *parts = nodeid.split("::")
+    module = path[: -len(".py")].replace("/", ".")
+    classname = ".".join([module, *parts[:-1]])
+    return junit_diff.case_id(classname, parts[-1])
+
+
+def read_selected(path):
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: selected node id list is missing") from exc
+    except (OSError, UnicodeDecodeError) as exc:
+        raise InputError(f"{path}: cannot read selected node ids: {exc}") from exc
+    nodeids = [line.strip() for line in lines if line.strip()]
+    if not nodeids:
+        raise InputError(f"{path}: selected node id list is empty")
+    if len(set(nodeids)) != len(nodeids):
+        raise InputError(f"{path}: duplicate selected node id")
+    by_case = {}
+    for nodeid in nodeids:
+        case = nodeid_to_case(nodeid)
+        if case in by_case:
+            raise InputError(f"{path}: node ids {by_case[case]} and {nodeid} collide")
+        by_case[case] = nodeid
+    return nodeids, by_case
+
+
+def target_states(directory):
+    """Return (nodeids, {nodeid: state}, [flaky nodeids]) for one target run."""
+    directory = Path(directory)
+    if not directory.is_dir():
+        raise InputError(f"{directory}: report directory is missing")
+    nodeids, by_case = read_selected(directory / "selected-nodeids.txt")
+    try:
+        first, _ = junit_diff.parse_report(str(directory / "junit.xml"))
+    except junit_diff.InputError as exc:
+        raise InputError(str(exc)) from exc
+    unexpected = sorted(set(first) - set(by_case))
+    if unexpected:
+        raise InputError(f"{directory}: junit.xml has unselected cases: {', '.join(unexpected[:5])}")
+    states = {by_case[case]: state for case, state in first.items()}
+    flaky = []
+    rerun_path = directory / "rerun-junit.xml"
+    if rerun_path.exists():
+        try:
+            rerun, _ = junit_diff.parse_report(str(rerun_path))
+        except junit_diff.InputError as exc:
+            raise InputError(str(exc)) from exc
+        for case, state in sorted(rerun.items()):
+            nodeid = by_case.get(case)
+            if nodeid is None or states.get(nodeid) != FAIL:
+                raise InputError(f"{rerun_path}: {case} was rerun without failing first")
+            if state == PASS:
+                flaky.append(nodeid)
+            states[nodeid] = PASS if state == PASS else FAIL
+    for nodeid in nodeids:
+        states.setdefault(nodeid, MISSING)
+    return nodeids, states, flaky
+
+
+def failed_names(directory):
+    _, states, _ = target_states(directory)
+    names = []
+    for nodeid, state in states.items():
+        if state == FAIL:
+            if not nodeid.startswith(TEST_FILE_PREFIX):
+                raise InputError(f"cannot express {nodeid} as an implemented_tests.txt entry")
+            names.append(nodeid[len(TEST_FILE_PREFIX):])
+    return sorted(names)
+
+
+# --- tolerated differences ---------------------------------------------------
+
+def read_tolerated(path, candidates):
+    try:
+        lines = Path(path).read_text(encoding="utf-8").splitlines()
+    except FileNotFoundError as exc:
+        raise InputError(f"{path}: tolerated-difference file is missing") from exc
+    except (OSError, UnicodeDecodeError) as exc:
+        raise InputError(f"{path}: cannot read tolerated differences: {exc}") from exc
+    entries = {}
+    for number, line in enumerate(lines, 1):
+        if not line.strip() or line.lstrip().startswith("#"):
+            continue
+        fields = [part.strip() for part in line.split(" | ")]
+        if len(fields) != 4 or not all(fields):
+            raise InputError(f"{path}:{number}: expected '<target> | <node id> | <issue URL> | <reason>'")
+        target, nodeid, issue, _reason = fields
+        if target not in candidates:
+            raise InputError(f"{path}:{number}: {target!r} is not a candidate target")
+        nodeid_to_case(nodeid)
+        if not ISSUE.fullmatch(issue):
+            raise InputError(f"{path}:{number}: tracking issue must be a https://github.com/rustfs/<repo>/issues/<n> URL")
+        if (target, nodeid) in entries:
+            raise InputError(f"{path}:{number}: duplicate entry for {target} {nodeid}")
+        entries[(target, nodeid)] = issue
+    return entries
+
+
+# --- comparison ----------------------------------------------------------------
+
+def count(states):
+    result = {PASS: 0, FAIL: 0, SKIP: 0, MISSING: 0}
+    for state in states.values():
+        result[state] += 1
+    return result
+
+
+def compare(targets, reports, tolerated_path):
+    reference = next(t["name"] for t in targets if t["binary"] == "reference")
+    candidates = [t["name"] for t in targets if t["binary"] == "candidate"]
+    tolerated = read_tolerated(tolerated_path, set(candidates))
+    runs = {t["name"]: target_states(Path(reports) / t["name"]) for t in targets}
+    selected = runs[reference][0]
+    for name, (nodeids, _, _) in runs.items():
+        if nodeids != selected:
+            raise InputError(f"{name}: selected node ids differ from the {reference} run")
+    ref_states = runs[reference][1]
+    differences = []
+    for name in candidates:
+        states = runs[name][1]
+        for nodeid in selected:
+            if ref_states[nodeid] != states[nodeid]:
+                issue = tolerated.get((name, nodeid))
+                differences.append({"target": name, "nodeid": nodeid, "reference": ref_states[nodeid],
+                                    "candidate": states[nodeid], "tolerated": issue})
+    seen = {(d["target"], d["nodeid"]) for d in differences}
+    stale = [{"target": t, "nodeid": n, "issue": i} for (t, n), i in sorted(tolerated.items()) if (t, n) not in seen]
+    incomplete = {name: [n for n in selected if run[1][n] == MISSING] for name, run in runs.items()}
+    return {
+        "reference": reference,
+        "candidates": candidates,
+        "selected": len(selected),
+        "counts": {name: count(run[1]) for name, run in runs.items()},
+        "flaky": {name: run[2] for name, run in runs.items()},
+        "differences": differences,
+        "stale_tolerated": stale,
+        "incomplete": {name: ids for name, ids in incomplete.items() if ids},
+    }
+
+
+def failed(summary):
+    return bool(summary["stale_tolerated"] or summary["incomplete"]
+                or any(not d["tolerated"] for d in summary["differences"]))
+
+
+def render(summary):
+    out = ["## s3-tests reference parity", ""]
+    out.append(f"Reference target `{summary['reference']}`; {summary['selected']} selected node ids per target.")
+    out += ["", "| Target | pass | fail | skip | missing | flaky (passed on rerun) |", "|---|---:|---:|---:|---:|---:|"]
+    for name, c in summary["counts"].items():
+        out.append(f"| `{name}` | {c[PASS]} | {c[FAIL]} | {c[SKIP]} | {c[MISSING]} | {len(summary['flaky'][name])} |")
+    diffs = summary["differences"]
+    out += ["", f"### Differences ({len(diffs)})", ""]
+    if diffs:
+        out += ["| Target | Node id | reference | candidate | Tolerated |", "|---|---|---|---|---|"]
+        for d in diffs:
+            out.append(f"| `{d['target']}` | `{d['nodeid']}` | {d['reference']} | {d['candidate']} | {d['tolerated'] or 'no'} |")
+    else:
+        out.append("No differences.")
+    if summary["stale_tolerated"]:
+        out += ["", "### Stale tolerated entries (no matching difference; remove them)", ""]
+        out += [f"- `{s['target']}` `{s['nodeid']}` ({s['issue']})" for s in summary["stale_tolerated"]]
+    if summary["incomplete"]:
+        out += ["", "### Incomplete runs (selected node ids without a result)", ""]
+        out += [f"- `{name}`: {len(ids)} missing, first `{ids[0]}`" for name, ids in summary["incomplete"].items()]
+    flaky = [(name, ids) for name, ids in summary["flaky"].items() if ids]
+    if flaky:
+        out += ["", "### Flaky (failed, then passed the single rerun; not a difference)", ""]
+        out += [f"- `{name}`: " + ", ".join(f"`{i}`" for i in ids) for name, ids in flaky]
+    out += ["", "Result: **" + ("FAIL" if failed(summary) else "PASS") + "**", ""]
+    return "\n".join(out)
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
+    sub = parser.add_subparsers(dest="mode", required=True)
+    tgt = sub.add_parser("targets")
+    tgt.add_argument("--config", required=True)
+    env = sub.add_parser("export-env")
+    env.add_argument("--target-json", required=True)
+    fail = sub.add_parser("failed")
+    fail.add_argument("--report", required=True, help="one target's report directory")
+    cmp_ = sub.add_parser("compare")
+    cmp_.add_argument("--targets-json", required=True)
+    cmp_.add_argument("--reports", required=True)
+    cmp_.add_argument("--tolerated", required=True)
+    cmp_.add_argument("--markdown", help="write the summary table here (appended)")
+    cmp_.add_argument("--json", help="write the machine-readable summary here")
+    args = parser.parse_args(argv)
+    try:
+        if args.mode == "targets":
+            print(json.dumps(load_targets(args.config), separators=(",", ":")))
+        elif args.mode == "export-env":
+            target = json.loads(args.target_json)
+            if not isinstance(target, dict):
+                raise InputError("target must be a JSON object")
+            for key, value in sorted(validate_env(target.get("env", {}), "target").items()):
+                print(f"{key}={value}")
+        elif args.mode == "failed":
+            for name in failed_names(args.report):
+                print(name)
+        else:
+            targets = validate_targets(json.loads(args.targets_json))
+            summary = compare(targets, args.reports, args.tolerated)
+            text = render(summary)
+            print(text)
+            if args.markdown:
+                with open(args.markdown, "a", encoding="utf-8") as handle:
+                    handle.write(text + "\n")
+            if args.json:
+                with open(args.json, "w", encoding="utf-8") as handle:
+                    json.dump(summary, handle, indent=2, sort_keys=True)
+                    handle.write("\n")
+            return 1 if failed(summary) else 0
+    except (InputError, json.JSONDecodeError) as exc:
+        print(f"error: {exc}", file=sys.stderr)
+        return 2
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/s3-tests/test_api_coverage.py

@@ -0,0 +1,93 @@
+#!/usr/bin/env python3
+# Copyright 2024 RustFS Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+"""Contract tests for how api_coverage.py finds `impl S3 for FS` across the ECFS source set."""
+
+from __future__ import annotations
+
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent))
+
+import api_coverage  # noqa: E402
+
+IMPL = "impl S3 for FS {\n    async fn get_object(&self) {}\n    async fn put_object(&self) {}\n}\n"
+
+
+class ImplementedOperationsTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.root = Path(self._tmp.name)
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def write(self, relative: str, text: str) -> None:
+        path = self.root / relative
+        path.parent.mkdir(parents=True, exist_ok=True)
+        path.write_text(text, encoding="utf-8")
+
+    def test_reads_the_impl_from_ecfs_when_the_split_directory_is_absent(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n" + IMPL)
+
+        self.assertEqual(api_coverage.implemented_operations(self.root), {"get_object", "put_object"})
+
+    def test_finds_the_impl_after_it_moves_into_the_split_directory(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/bucket_core.rs", "impl FS {\n    async fn helper(&self) {}\n}\n")
+
+        self.assertEqual(api_coverage.implemented_operations(self.root), {"get_object", "put_object"})
+
+    def test_missing_ecfs_fails_even_when_the_split_directory_holds_the_impl(self) -> None:
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "ecfs.rs is missing"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_an_impl_in_both_ecfs_and_the_split_directory_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, IMPL)
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/s3_impl.rs", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "exactly once.*found 2"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_no_impl_anywhere_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/bucket_core.rs", "impl FS {}\n")
+
+        with self.assertRaisesRegex(RuntimeError, "found 0"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_an_impl_in_a_non_rust_file_of_the_split_directory_is_not_source(self) -> None:
+        self.write(api_coverage.ECFS_PATH, "struct FS;\n")
+        self.write(f"{api_coverage.ECFS_SPLIT_DIR}/notes.md", IMPL)
+
+        with self.assertRaisesRegex(RuntimeError, "found 0"):
+            api_coverage.implemented_operations(self.root)
+
+    def test_a_split_path_that_is_a_file_fails(self) -> None:
+        self.write(api_coverage.ECFS_PATH, IMPL)
+        self.write(api_coverage.ECFS_SPLIT_DIR, "stray\n")
+
+        with self.assertRaisesRegex(RuntimeError, "not a directory"):
+            api_coverage.implemented_operations(self.root)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/s3-tests/test_reference_parity.py

@@ -0,0 +1,332 @@
+#!/usr/bin/env python3
+# 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.
+"""Contract tests for reference_parity.py: target list, rerun merge, tolerated entries and exit codes."""
+
+import json
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+
+SCRIPT = Path(__file__).resolve().parent / "reference_parity.py"
+REPO = Path(__file__).resolve().parents[2]
+PREFIX = "s3tests/functional/test_s3.py::"
+CLASSNAME = "s3tests.functional.test_s3"
+TARGETS = [{"name": "reference", "binary": "reference", "env": {}},
+           {"name": "candidate-legacy", "binary": "candidate", "env": {}}]
+ISSUE = "https://github.com/rustfs/rustfs/issues/1"
+
+
+def junit(states):
+    cases = []
+    for name, state in states.items():
+        body = {"pass": "", "fail": "<failure message='x'/>", "error": "<error message='x'/>",
+                "skip": "<skipped/>"}[state]
+        cases.append(f'<testcase classname="{CLASSNAME}" name="{name}">{body}</testcase>')
+    return f'<testsuites><testsuite name="pytest">{"".join(cases)}</testsuite></testsuites>'
+
+
+def run(*args):
+    proc = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False)
+    return proc.returncode, proc.stdout, proc.stderr
+
+
+class Lane:
+    """A temporary reports directory with one sub-directory per target."""
+
+    def __init__(self, case):
+        self.temp = tempfile.TemporaryDirectory()
+        case.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.reports = self.root / "reports"
+        self.tolerated = self.root / "tolerated.txt"
+        self.tolerated.write_text("# none\n")
+
+    def target(self, name, states, selected=None, rerun=None):
+        directory = self.reports / name
+        directory.mkdir(parents=True)
+        names = selected if selected is not None else list(states)
+        (directory / "selected-nodeids.txt").write_text("".join(PREFIX + n + "\n" for n in names))
+        (directory / "junit.xml").write_text(junit(states))
+        if rerun is not None:
+            (directory / "rerun-junit.xml").write_text(junit(rerun))
+        return directory
+
+    def compare(self, targets=None):
+        summary = self.root / "summary.json"
+        code, out, err = run("compare", "--targets-json", json.dumps(targets or TARGETS),
+                             "--reports", str(self.reports), "--tolerated", str(self.tolerated),
+                             "--json", str(summary))
+        data = json.loads(summary.read_text()) if summary.exists() else None
+        return code, out, err, data
+
+
+BASE = {"test_a": "pass", "test_b": "pass", "test_c": "skip"}
+
+
+class CompareTests(unittest.TestCase):
+    # --- negative cases ---
+
+    def test_pass_versus_fail_is_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"], [{"target": "candidate-legacy", "nodeid": PREFIX + "test_a",
+                                                "reference": "pass", "candidate": "fail", "tolerated": None}])
+        self.assertIn("Result: **FAIL**", out)
+
+    def test_error_counts_as_fail(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_b": "error"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["candidate"], "fail")
+
+    def test_reference_fail_candidate_pass_is_also_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", {**BASE, "test_a": "fail"})
+        lane.target("candidate-legacy", BASE)
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["reference"], "fail")
+
+    def test_skip_versus_pass_is_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_c": "pass"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["nodeid"], PREFIX + "test_c")
+
+    def test_case_absent_from_candidate_report_is_missing_and_incomplete(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {"test_a": "pass", "test_c": "skip"}, selected=list(BASE))
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"][0]["candidate"], "missing")
+        self.assertEqual(data["incomplete"], {"candidate-legacy": [PREFIX + "test_b"]})
+        self.assertIn("Incomplete runs", out)
+
+    def test_case_missing_from_both_reports_still_fails(self):
+        lane = Lane(self)
+        partial = {"test_a": "pass", "test_c": "skip"}
+        lane.target("reference", partial, selected=list(BASE))
+        lane.target("candidate-legacy", partial, selected=list(BASE))
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["differences"], [])
+        self.assertEqual(set(data["incomplete"]), {"reference", "candidate-legacy"})
+
+    def test_failure_that_also_fails_the_rerun_stays_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_a": "fail"})
+        code, _, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["flaky"]["candidate-legacy"], [])
+
+    def test_rerun_of_a_case_that_did_not_fail_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_b": "pass"})
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("rerun without failing first", err)
+
+    def test_missing_target_report_fails_rather_than_skips(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        code, _, err, data = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIsNone(data)
+        self.assertIn("candidate-legacy: report directory is missing", err)
+
+    def test_missing_junit_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        (lane.target("candidate-legacy", BASE) / "junit.xml").unlink()
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("cannot read", err)
+
+    def test_different_selected_lists_are_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {"test_a": "pass", "test_b": "pass"})
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("selected node ids differ", err)
+
+    def test_unselected_case_in_junit_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_extra": "pass"}, selected=list(BASE))
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("unselected cases", err)
+
+    def test_stale_tolerated_entry_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        lane.tolerated.write_text(f"candidate-legacy | {PREFIX}test_a | {ISSUE} | known drift\n")
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 1)
+        self.assertEqual(data["stale_tolerated"][0]["nodeid"], PREFIX + "test_a")
+        self.assertIn("Stale tolerated entries", out)
+
+    def test_tolerated_entry_for_another_target_does_not_cover_the_difference(self):
+        lane = Lane(self)
+        targets = TARGETS + [{"name": "candidate-gateway", "binary": "candidate", "env": {}}]
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        lane.target("candidate-gateway", {**BASE, "test_a": "fail"})
+        lane.tolerated.write_text(f"candidate-gateway | {PREFIX}test_a | {ISSUE} | gateway gap\n")
+        code, _, _, data = lane.compare(targets)
+        self.assertEqual(code, 1)
+        untolerated = [d["target"] for d in data["differences"] if not d["tolerated"]]
+        self.assertEqual(untolerated, ["candidate-legacy"])
+
+    def test_tolerated_entry_without_issue_url_is_rejected(self):
+        for line in (f"candidate-legacy | {PREFIX}test_a | rustfs#1 | reason",
+                     f"candidate-legacy | {PREFIX}test_a | {ISSUE} | ",
+                     f"candidate-legacy | {PREFIX}test_a | {ISSUE}",
+                     f"reference | {PREFIX}test_a | {ISSUE} | reason",
+                     f"candidate-legacy | test_a | {ISSUE} | reason"):
+            with self.subTest(line=line):
+                lane = Lane(self)
+                lane.target("reference", BASE)
+                lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+                lane.tolerated.write_text(line + "\n")
+                code, _, err, _ = lane.compare()
+                self.assertEqual(code, 2, err)
+
+    def test_duplicate_tolerated_entry_is_rejected(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        entry = f"candidate-legacy | {PREFIX}test_a | {ISSUE} | reason\n"
+        lane.tolerated.write_text(entry * 2)
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("duplicate entry", err)
+
+    def test_missing_tolerated_file_fails(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        lane.tolerated.unlink()
+        code, _, err, _ = lane.compare()
+        self.assertEqual(code, 2)
+        self.assertIn("tolerated-difference file is missing", err)
+
+    # --- positive cases ---
+
+    def test_identical_runs_pass(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", BASE)
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["counts"]["candidate-legacy"], {"pass": 2, "fail": 0, "skip": 1, "missing": 0})
+        self.assertIn("No differences.", out)
+        self.assertIn("Result: **PASS**", out)
+
+    def test_failure_that_passes_the_rerun_is_flaky_not_a_difference(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"}, rerun={"test_a": "pass"})
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["flaky"]["candidate-legacy"], [PREFIX + "test_a"])
+        self.assertIn("Flaky", out)
+
+    def test_tolerated_difference_passes_and_is_labelled(self):
+        lane = Lane(self)
+        lane.target("reference", BASE)
+        lane.target("candidate-legacy", {**BASE, "test_a": "fail"})
+        lane.tolerated.write_text(f"# comment\ncandidate-legacy | {PREFIX}test_a | {ISSUE} | known drift\n")
+        code, out, _, data = lane.compare()
+        self.assertEqual(code, 0)
+        self.assertEqual(data["differences"][0]["tolerated"], ISSUE)
+        self.assertIn(ISSUE, out)
+
+
+class TargetTests(unittest.TestCase):
+    def write(self, text):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        path = Path(temp.name) / "targets.toml"
+        path.write_text(text)
+        return path
+
+    def test_invalid_target_lists_are_rejected(self):
+        ref = '[[targets]]\nname = "reference"\nbinary = "reference"\n'
+        cand = '[[targets]]\nname = "candidate-legacy"\nbinary = "candidate"\n'
+        for label, text in {
+            "no schema": ref + cand,
+            "only reference": "schema = 1\n" + ref,
+            "two references": "schema = 1\n" + ref + ref.replace('"reference"\nbinary', '"other"\nbinary'),
+            "no reference": "schema = 1\n" + cand + cand.replace("candidate-legacy", "candidate-two"),
+            "duplicate name": "schema = 1\n" + ref + cand + cand,
+            "unknown binary": "schema = 1\n" + ref + cand.replace('binary = "candidate"', 'binary = "main"'),
+            "bad name": "schema = 1\n" + ref + cand.replace("candidate-legacy", "Candidate Legacy"),
+            "foreign env key": "schema = 1\n" + ref + cand + 'env = { PATH = "/tmp" }\n',
+            "env injection": "schema = 1\n" + ref + cand + 'env = { RUSTFS_X = "a\\nB=c" }\n',
+            "unknown key": "schema = 1\n" + ref + cand + 'stack = "gateway"\n',
+        }.items():
+            with self.subTest(label):
+                code, _, _ = run("targets", "--config", str(self.write(text)))
+                self.assertEqual(code, 2)
+
+    def test_missing_target_file_fails(self):
+        code, _, err = run("targets", "--config", "/nonexistent/targets.toml")
+        self.assertEqual(code, 2)
+        self.assertIn("target list is missing", err)
+
+    def test_export_env_rejects_foreign_keys(self):
+        code, _, _ = run("export-env", "--target-json", json.dumps({"env": {"LD_PRELOAD": "x"}}))
+        self.assertEqual(code, 2)
+
+    def test_repository_target_list_is_valid(self):
+        code, out, err = run("targets", "--config", str(REPO / ".config/reference-parity-targets.toml"))
+        self.assertEqual(code, 0, err)
+        self.assertEqual(json.loads(out), TARGETS)
+
+    def test_export_env_prints_validated_lines(self):
+        code, out, _ = run("export-env", "--target-json", json.dumps({"env": {"RUSTFS_S3_STACK": "gateway"}}))
+        self.assertEqual((code, out), (0, "RUSTFS_S3_STACK=gateway\n"))
+
+
+class FailedTests(unittest.TestCase):
+    def test_failed_lists_only_failures_in_test_list_form(self):
+        lane = Lane(self)
+        directory = lane.target("reference", {**BASE, "test_a": "fail", "test_d": "error"})
+        code, out, _ = run("failed", "--report", str(directory))
+        self.assertEqual((code, out), (0, "test_a\ntest_d\n"))
+
+    def test_failed_rejects_a_missing_report(self):
+        code, _, err = run("failed", "--report", "/nonexistent/report")
+        self.assertEqual(code, 2)
+        self.assertIn("report directory is missing", err)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_guard_source_set.sh

@@ -0,0 +1,153 @@
+#!/usr/bin/env bash
+# Contract tests for scripts/guard_source_set.sh, the source-set helper that
+# lets text guards read a file plus the directory a mechanical split moves its
+# code into (rustfs/backlog#2765).
+#
+# Each case builds a throwaway tree; nothing in the real workspace is read.
+
+set -euo pipefail
+
+SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)"
+# shellcheck source=scripts/guard_source_set.sh
+source "${GUARD_SOURCE_SET_SCRIPT:-$SCRIPT_DIR/guard_source_set.sh}"
+TMP_ROOT="$(mktemp -d)"
+trap 'chmod -R u+rwx "$TMP_ROOT" 2>/dev/null || true; rm -rf "$TMP_ROOT"' EXIT
+
+passed=0
+skipped=0
+case_no=0
+FX=""
+STATUS=0
+ERR=""
+
+fail() {
+  echo "FAIL: $1" >&2
+  exit 1
+}
+
+new_case() {
+  case_no=$((case_no + 1))
+  FX="$TMP_ROOT/case$case_no"
+  mkdir -p "$FX"
+}
+
+put() {
+  mkdir -p "$(dirname "$FX/$1")"
+  printf 'fn guarded() {}\n' >"$FX/$1"
+}
+
+# run_set <args...>: call guard_source_set on the case tree, capturing status and stderr.
+run_set() {
+  local err_file="$FX/.stderr"
+  STATUS=0
+  guard_source_set "$FX" "$@" 2>"$err_file" || STATUS=$?
+  ERR="$(cat "$err_file")"
+}
+
+sources_are() {
+  local expected actual
+  expected="$(printf '%s\n' "$@")"
+  actual="$(printf '%s\n' "${GUARD_SOURCES[@]}")"
+  [[ "$actual" == "$expected" ]] || fail "case $case_no: sources were [$actual], expected [$expected]"
+}
+
+ok() {
+  passed=$((passed + 1))
+}
+
+# --- positive cases ---------------------------------------------------------
+
+# An absent split directory contributes nothing.
+new_case
+put src/usecase.rs
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 0 ]] || fail "absent split dir: status $STATUS ($ERR)"
+sources_are src/usecase.rs
+ok
+
+# A present split directory adds its .rs files, recursively, in byte order,
+# after the required file; non-.rs files are not source.
+new_case
+put src/usecase.rs
+put src/usecase/zeta.rs
+put src/usecase/B.rs
+put src/usecase/a.rs
+put src/usecase/tests/nested.rs
+printf 'not rust\n' >"$FX/src/usecase/notes.md"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 0 ]] || fail "present split dir: status $STATUS ($ERR)"
+sources_are src/usecase.rs src/usecase/B.rs src/usecase/a.rs src/usecase/tests/nested.rs src/usecase/zeta.rs
+ok
+
+# Several split directories are read in argument order; an empty one adds nothing.
+new_case
+put src/ecfs.rs
+put src/legacy_edge/b.rs
+mkdir -p "$FX/src/empty"
+put src/other/a.rs
+run_set src/ecfs.rs src/legacy_edge src/empty src/missing src/other
+[[ "$STATUS" == 0 ]] || fail "several split dirs: status $STATUS ($ERR)"
+sources_are src/ecfs.rs src/legacy_edge/b.rs src/other/a.rs
+ok
+
+# --- negative cases ---------------------------------------------------------
+
+# A missing required file fails and names the file.
+new_case
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "missing required file: status $STATUS"
+[[ "$ERR" == *"src/usecase.rs"* ]] || fail "missing required file: stderr does not name it: $ERR"
+sources_are src/usecase.rs
+ok
+
+# A missing required file fails even when the split directory holds the code:
+# the split must keep the file, and a guard must not pass on half its input.
+new_case
+put src/usecase/moved.rs
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "missing required file with split dir: status $STATUS"
+[[ "$ERR" == *"is missing"* ]] || fail "missing required file with split dir: $ERR"
+ok
+
+# A required path that is a directory is not a source file.
+new_case
+mkdir -p "$FX/src/usecase.rs"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "required directory: status $STATUS"
+ok
+
+# A split path that exists as a regular file fails rather than being skipped.
+new_case
+put src/usecase.rs
+printf 'stray\n' >"$FX/src/usecase"
+run_set src/usecase.rs src/usecase
+[[ "$STATUS" == 1 ]] || fail "split path is a file: status $STATUS"
+[[ "$ERR" == *"not a directory"* ]] || fail "split path is a file: $ERR"
+ok
+
+# A later split path that is a file still fails after an earlier valid one.
+new_case
+put src/usecase.rs
+put src/usecase/a.rs
+printf 'stray\n' >"$FX/src/second"
+run_set src/usecase.rs src/usecase src/second
+[[ "$STATUS" == 1 ]] || fail "second split path is a file: status $STATUS"
+ok
+
+# An unlistable split directory fails rather than contributing nothing.
+new_case
+put src/usecase.rs
+put src/usecase/hidden/a.rs
+chmod 000 "$FX/src/usecase/hidden"
+if [[ "$(id -u)" == 0 ]]; then
+  echo "SKIP: unlistable split directory case needs a non-root user (root ignores directory permissions)"
+  skipped=$((skipped + 1))
+else
+  run_set src/usecase.rs src/usecase
+  [[ "$STATUS" == 1 ]] || fail "unlistable split dir: status $STATUS"
+  [[ "$ERR" == *"cannot be listed"* ]] || fail "unlistable split dir: $ERR"
+  ok
+fi
+chmod 755 "$FX/src/usecase/hidden"
+
+echo "guard_source_set contract tests: $passed passed, $skipped skipped"

scripts/test_reference_binary.py

@@ -0,0 +1,157 @@
+#!/usr/bin/env python3
+# 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.
+"""Exercise the reference-binary pin resolver and unpacker without network access."""
+
+import hashlib
+from pathlib import Path
+import subprocess
+import sys
+import tempfile
+import unittest
+import zipfile
+
+ROOT = Path(__file__).resolve().parents[1]
+SCRIPT = ROOT / ".github/actions/reference-binary/reference_binary.py"
+PIN = '''schema = 1
+repository = "rustfs/rustfs"
+version = "1.0.5"
+tag = "1.0.5"
+
+[platforms.linux-x86_64-gnu]
+asset = "rustfs-linux-x86_64-gnu-v1.0.5.zip"
+sha256 = "{sha}"
+executable = "rustfs"
+'''
+SHA = "a" * 64
+
+
+def run(*args):
+    proc = subprocess.run([sys.executable, str(SCRIPT), *args], capture_output=True, text=True, check=False)
+    return proc.returncode, proc.stdout, proc.stderr
+
+
+class Base(unittest.TestCase):
+    def setUp(self):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        self.root = Path(temp.name)
+
+    def pin(self, text):
+        path = self.root / "pin.toml"
+        path.write_text(text)
+        return path
+
+    def archive(self, members):
+        path = self.root / "rustfs.zip"
+        with zipfile.ZipFile(path, "w") as bundle:
+            for name, data in members.items():
+                bundle.writestr(name, data)
+        return path, hashlib.sha256(path.read_bytes()).hexdigest()
+
+
+class ResolveTests(Base):
+    def test_missing_pin_file_fails(self):
+        code, _, err = run("resolve", "--pin", str(self.root / "absent.toml"), "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 1)
+        self.assertIn("reference pin file is missing", err)
+
+    def test_unpinned_platform_fails(self):
+        code, _, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-aarch64-gnu")
+        self.assertEqual(code, 1)
+        self.assertIn("no pin for platform", err)
+
+    def test_malformed_pins_fail(self):
+        good = PIN.format(sha=SHA)
+        for label, text in {
+            "short digest": good.replace(SHA, "a" * 63),
+            "uppercase digest": good.replace(SHA, "A" * 64),
+            "no digest": good.replace(f'sha256 = "{SHA}"\n', ""),
+            "path in asset": good.replace('asset = "rustfs-', 'asset = "../rustfs-'),
+            "non-zip asset": good.replace("v1.0.5.zip", "v1.0.5.deb"),
+            "asset of another version": good.replace("v1.0.5.zip", "v1.0.4.zip"),
+            "path in executable": good.replace('executable = "rustfs"', 'executable = "bin/rustfs"'),
+            "wrong schema": good.replace("schema = 1", "schema = 2"),
+            "boolean schema": good.replace("schema = 1", "schema = true"),
+            "no repository": good.replace('repository = "rustfs/rustfs"\n', ""),
+            "unknown key": good + 'url = "https://example.invalid"\n',
+            "not toml": "schema = \n",
+        }.items():
+            with self.subTest(label):
+                code, _, _ = run("resolve", "--pin", str(self.pin(text)), "--platform", "linux-x86_64-gnu")
+                self.assertEqual(code, 1)
+
+    def test_resolves_identity_as_step_outputs(self):
+        code, out, err = run("resolve", "--pin", str(self.pin(PIN.format(sha=SHA))), "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 0, err)
+        self.assertEqual(out.splitlines(), ["repository=rustfs/rustfs", "version=1.0.5", "tag=1.0.5",
+                                            "asset=rustfs-linux-x86_64-gnu-v1.0.5.zip", f"sha256={SHA}",
+                                            "executable=rustfs"])
+
+    def test_repository_pin_is_valid(self):
+        code, out, err = run("resolve", "--pin", str(ROOT / ".config/reference-binary.toml"),
+                             "--platform", "linux-x86_64-gnu")
+        self.assertEqual(code, 0, err)
+        self.assertIn("version=1.0.5", out.splitlines())
+
+
+class UnpackTests(Base):
+    def unpack(self, archive, sha):
+        return run("unpack", "--archive", str(archive), "--sha256", sha, "--executable", "rustfs",
+                   "--destination", str(self.root / "out"))
+
+    def test_digest_mismatch_fails_and_extracts_nothing(self):
+        archive, _ = self.archive({"rustfs": b"binary"})
+        code, _, err = self.unpack(archive, SHA)
+        self.assertEqual(code, 1)
+        self.assertIn("does not match the pinned digest", err)
+        self.assertFalse((self.root / "out").exists())
+
+    def test_missing_archive_fails(self):
+        code, _, err = self.unpack(self.root / "absent.zip", SHA)
+        self.assertEqual(code, 1)
+        self.assertIn("missing or unreadable", err)
+
+    def test_archive_without_top_level_executable_fails(self):
+        archive, sha = self.archive({"bin/rustfs": b"binary"})
+        code, _, err = self.unpack(archive, sha)
+        self.assertEqual(code, 1)
+        self.assertIn("exactly one top-level", err)
+
+    def test_empty_executable_fails(self):
+        archive, sha = self.archive({"rustfs": b""})
+        code, _, _ = self.unpack(archive, sha)
+        self.assertEqual(code, 1)
+
+    def test_non_zip_archive_fails(self):
+        archive = self.root / "rustfs.zip"
+        archive.write_bytes(b"not a zip")
+        code, _, err = self.unpack(archive, hashlib.sha256(b"not a zip").hexdigest())
+        self.assertEqual(code, 1)
+        self.assertIn("not a valid zip", err)
+
+    def test_extracts_only_the_executable(self):
+        archive, sha = self.archive({"rustfs": b"binary", "../escape": b"x", "README": b"r"})
+        code, out, err = self.unpack(archive, sha)
+        self.assertEqual(code, 0, err)
+        target = self.root / "out" / "rustfs"
+        self.assertEqual(out.strip(), f"path={target}")
+        self.assertEqual(target.read_bytes(), b"binary")
+        self.assertTrue(target.stat().st_mode & 0o111)
+        self.assertEqual(sorted(p.name for p in (self.root / "out").iterdir()), ["rustfs"])
+        self.assertFalse((self.root / "escape").exists())
+
+
+if __name__ == "__main__":
+    unittest.main()