ci: build linux-aarch64 releases on Ubicloud ARM (#8996)

- Resolve existing directories below base on Linux in a single openat2(BENEATH | NO_SYMLINKS) syscall in open_existing_directory_below_base_std instead of opening and closing every ancestor component.
- Reuse the pinned source directory guard across sibling metadata reads and control probes in validate_source_mutation, load_source, OrdinaryMultipartDeleteCheck::validate, and prepare_multipart_commit_staging.
- Read normal multipart source heads (<= 64 KiB) in a single pass in load_source while preserving constant-memory streaming observation for large prepared histories.
- Record root_missing during load_source, reuse loaded.original in restore_original, and coalesce finish source directory fsync into persist_with_source_sync.
- Reuse the opened commit directory in prepare_rename_commit_guard and LinuxCommitDirectories::validate when source_parent == destination_parent.

.agents/skills/adversarial-validation/references/performance.md

@@ -6,7 +6,17 @@
 - Route every new sync/flush through the durability-mode and bucket override
   gates; mode `none` must not pay the new fsync.
 - Keep blocking filesystem/CPU work off async runtime threads, but do not split
-  one small operation into many `spawn_blocking` round trips.
+  one small operation into many `spawn_blocking` round trips; coalesce `open`,
+  `read`/`write`, `fsync`, and `close` into one blocking closure per disk.
+- Check for per-disk syscall amplification (`disks × parts/files`): pin a
+  directory once (`openat2` on Linux) and reuse the handle across sibling reads
+  instead of re-walking ancestor components per file or re-opening identical
+  `source_parent == destination_parent` directories.
+- After `fstat`/`file.metadata()`, require `Vec::with_capacity(len + 1)` before
+  `read_to_end` so small metadata reads do not trigger 6–7 exponential-growth
+  `read()` syscalls.
+- Avoid redundant `store.load()` or `root_missing()` round-trips under the same
+  exclusive lock when the state or original bytes were just loaded or persisted.
 - Measure lock hold time across I/O and compare acquisition order for ABBA.
 - Keep cleanup, extra stat/rename, and diagnostics out of the PUT commit critical
   section when they need not be there.

.config/e2e-distributed-selection.txt

@@ -1,2 +1,2 @@
-sha256-linux=14becc458b9eb24add510d94ee6cd04ba1fb19907d598effe1524e5242400d10
-sha256-darwin=14becc458b9eb24add510d94ee6cd04ba1fb19907d598effe1524e5242400d10
+sha256-linux=6190506b9e4bdb60fd392544a7bc47be708077bb4fe7a2ef689d047455e1af86
+sha256-darwin=6190506b9e4bdb60fd392544a7bc47be708077bb4fe7a2ef689d047455e1af86

.config/make/tests.mak

@@ -52,6 +52,7 @@ script-tests: ## Run shell script tests
 	./scripts/check_embedded_secrets.sh --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/check_test_wiring.py --self-test
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_e2e_binary.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_btf_prototypes.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_build_workflow.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_release_glibc.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_upgrade_workflow.py

.github/actions/schedule-failure-issue/README.md

@@ -74,7 +74,6 @@ Notes:
 - `.github/workflows/e2e-s3tests.yml` (weekly full s3-tests sweep)
 - `.github/workflows/mint.yml` (weekly multi-SDK mint run)
 - `.github/workflows/fuzz.yml` (nightly fuzz corpus jobs)
-- `.github/workflows/performance-ab.yml` (nightly warp A/B gate)
 - **Pending:** the ci-7 e2e nightly-full workflow does not exist yet
   (backlog#1149 ci-7). When it lands, wire it up with the snippet above.
 

.github/scheduled-validations.json

@@ -18,7 +18,6 @@
     "never_ran_grace_until": "2026-09-08T00:00:00Z"
   },
   { "workflow": ".github/workflows/nightly-gnu.yml", "max_age_hours": 36 },
-  { "workflow": ".github/workflows/performance-ab.yml", "max_age_hours": 36 },
   {
     "workflow": ".github/workflows/runner-hygiene.yml",
     "max_age_hours": 792,

.github/workflows/README.md

@@ -7,7 +7,7 @@
 
 This repository is public. For a public repository, GitHub's standard hosted runners (`ubuntu-latest`, `windows-latest`, `macos-latest`) cost nothing and have no minute cap, whatever plan the organisation is on. GitHub bills only two things: larger hosted runners, which have custom labels and are not used here, and standard runners in private repositories.
 
-Every other label in this repository is capacity RustFS pays for by the hour whether or not a job is running well:
+Every other label in this repository is capacity RustFS pays for. ARC pods and lab machines accrue cost while the machine exists. A Ubicloud label is metered for the minutes a job holds its VM:
 
 | Label | Paid by | What it is | Use it for |
 |---|---|---|---|
@@ -15,7 +15,7 @@ Every other label in this repository is capacity RustFS pays for by the hour whe
 | `windows-latest`, `macos-latest` | nobody | GitHub-hosted standard runners | Windows and macOS targets |
 | `sm-standard-2` | RustFS | ARC pod, Ubuntu 24.04, minimal image: no `gh`, no C toolchain; Docker only on the `dind-` label | only a job that a hosted runner cannot run (see the placement rule) |
 | `sm-standard-4` | RustFS | ARC pod, Ubuntu 24.04, 14 CPUs and 28 GB (the label is not the core count), large disk | workspace-scale Cargo builds |
-| `sm-standard-4-arm` | RustFS | ARC pod, aarch64; reached through the `build.yml` matrix, so `actionlint.yaml` does not list it | native aarch64 release builds |
+| `ubicloud-standard-16-arm` | RustFS | Ubicloud VM, alias of `ubicloud-standard-16-arm-ubuntu-2404`: Ubuntu 24.04 arm64, 16 vCPU, 48 GB, 300 GB disk. Reached through the `build.yml` matrix, so `actionlint.yaml` does not list it | native aarch64 release builds (`linux-aarch64-musl`, `linux-aarch64-gnu`) |
 | `dind-sm-standard-2` | RustFS | ARC pod with Docker-in-Docker | image builds that need `buildx` and a registry push |
 | `smoke-testing`, `pf-testing` | RustFS | long-lived lab machines | the packaged functional chain and performance suites |
 
@@ -40,7 +40,7 @@ A hosted runner is also the only choice for a few things the pods cannot do: `mo
 
 - Every `shared-key` has exactly one writer. `cache-warm.yml` owns `ci-dev`, `ci-feat-rio`, `ci-feat-proto` and `ci-uring`; every other `./.github/actions/setup` call states `cache-save-if` explicitly, and `scripts/security/check_cache_save_if.sh` fails the build when one does not.
 - All readers of a key run on the same image as its writer. Keys that live on hosted runners say so in their name (`ci-uring` is warmed on `ubuntu-latest`; `ci-e2e-distributed-hosted`, `e2e-upgrade-server-hosted`, `e2e-upgrade-tests-hosted`).
-- Moving a job between pools therefore means renaming its key. The new key is empty until a run from `main` (a scheduled run, or a manual dispatch of that workflow from `main`) saves it; until then that lane does a cold build.
+- Moving a job between pools therefore means renaming its key. The new key is empty until a run from `main` (a scheduled run, or a manual dispatch of that workflow from `main`) saves it; until then that lane does a cold build. The `build.yml` aarch64 release legs append `-ubicloud-arm` for this reason. `cache-warm.yml` does not write those keys.
 - `CARGO_BUILD_JOBS` and every other `CARGO*`, `CC*`, `CFLAGS*`, `CXX*`, `CMAKE*` and `RUST*` variable is hashed into the key. `ci.yml` sets `CARGO_BUILD_JOBS` at step level for that reason; promoting it to job level rotates the key for every lane that shares it.
 
 ## Where self-hosted minutes leak
@@ -62,7 +62,7 @@ gh api "repos/rustfs/rustfs/actions/runs/<run_id>/jobs?per_page=100" \
   --jq '.jobs[] | "\(.conclusion)\t\(.labels | join(","))\t\(.started_at)\t\(.completed_at)\t\(.name)"'
 ```
 
-Only rows whose label starts with `sm-`, `dind-`, `smoke-` or `pf-` cost anything. At the October 2026 review, a full pull-request run of `ci.yml` spent about 150 self-hosted minutes, a third of it in jobs that never compiled; treat that as the scale of what a placement mistake costs, not as a current figure.
+Only rows whose label starts with `sm-`, `dind-`, `ubicloud-`, `smoke-` or `pf-` cost anything. At the October 2026 review, a full pull-request run of `ci.yml` spent about 150 self-hosted minutes, a third of it in jobs that never compiled; treat that as the scale of what a placement mistake costs, not as a current figure.
 
 ## Checklist for a new or moved job
 

.github/workflows/audit.yml

@@ -29,7 +29,6 @@ on:
       - 'scripts/test_package_service_scripts.sh'
       - 'scripts/test_package_versions.sh'
       - 'scripts/test_docker_workflow.py'
-      - 'scripts/security/check_performance_ab_workflow.sh'
       - 'scripts/security/check_preview_release_workflow.sh'
       - 'scripts/security/check_tier_artifact_workflow.sh'
       - 'scripts/security/check_workflow_pins.sh'
@@ -48,7 +47,6 @@ on:
       - 'scripts/test_package_service_scripts.sh'
       - 'scripts/test_package_versions.sh'
       - 'scripts/test_docker_workflow.py'
-      - 'scripts/security/check_performance_ab_workflow.sh'
       - 'scripts/security/check_preview_release_workflow.sh'
       - 'scripts/security/check_tier_artifact_workflow.sh'
       - 'scripts/security/check_workflow_pins.sh'
@@ -155,9 +153,6 @@ jobs:
       - name: Check preview release workflow policy
         run: ./scripts/security/check_preview_release_workflow.sh
 
-      - name: Check performance A/B workflow trust boundary
-        run: ./scripts/security/check_performance_ab_workflow.sh
-
       - name: Check tier evidence workflow isolation
         run: ./scripts/security/check_tier_artifact_workflow.sh
 

.github/workflows/btf-prototype-reference.yml

@@ -0,0 +1,70 @@
+name: BTF prototype reference
+
+on:
+  pull_request:
+    paths:
+      - "scripts/inspect_btf_prototypes.py"
+      - "scripts/test_inspect_btf_prototypes.py"
+      - "scripts/fixtures/btf_prototypes.c"
+      - ".github/workflows/btf-prototype-reference.yml"
+  workflow_dispatch:
+
+permissions:
+  contents: read
+
+concurrency:
+  group: btf-prototype-reference-${{ github.ref }}
+  cancel-in-progress: true
+
+jobs:
+  reference:
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    steps:
+      - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          persist-credentials: false
+      - uses: actions/setup-python@5fda3b95a4ea91299a34e894583c3862153e4b97 # v7.0.0
+        with:
+          python-version: "3.12"
+      - name: Check prototype and CLI contracts
+        run: python3 -B scripts/test_inspect_btf_prototypes.py
+      - name: Install only missing native reference tools
+        run: |
+          select_bpftool() {
+            local candidate
+            local candidates=()
+            if command -v bpftool >/dev/null; then
+              candidates+=("$(command -v bpftool)")
+            fi
+            for candidate in /usr/lib/linux-tools-*/bpftool /usr/lib/linux-tools/*/bpftool; do
+              if [[ -x "$candidate" ]]; then
+                candidates+=("$candidate")
+              fi
+            done
+            while IFS= read -r candidate; do
+              if [[ -x "$candidate" ]] && file -Lb "$candidate" | grep -q 'ELF' \
+                && "$candidate" version >/dev/null 2>&1; then
+                printf '%s\n' "$candidate"
+                return 0
+              fi
+            done < <(printf '%s\n' "${candidates[@]}" | sort -Vr)
+            return 1
+          }
+          packages=()
+          command -v clang >/dev/null || packages+=(clang)
+          command -v llvm-objcopy >/dev/null || packages+=(llvm)
+          select_bpftool >/dev/null || packages+=(linux-tools-generic)
+          if (( ${#packages[@]} )); then
+            sudo apt-get update
+            sudo apt-get install --no-install-recommends -y "${packages[@]}"
+          fi
+          # File parsing needs a working binary, not a host-kernel wrapper.
+          selected_bpftool="$(select_bpftool)"
+          "$selected_bpftool" version
+          printf 'BPFTOOL=%s\n' "$selected_bpftool" >> "$GITHUB_ENV"
+      - name: Compare compiled declarations with bpftool
+        run: |
+          clang -target bpf -O2 -g -c scripts/fixtures/btf_prototypes.c -o "$RUNNER_TEMP/prototypes.o"
+          llvm-objcopy --dump-section ".BTF=$RUNNER_TEMP/prototypes.btf" "$RUNNER_TEMP/prototypes.o"
+          python3 -B scripts/test_inspect_btf_prototypes.py --native-reference "$RUNNER_TEMP/prototypes.btf"

.github/workflows/build.yml

@@ -181,12 +181,18 @@ jobs:
             selected="all"
           fi
 
-
+          # linux-aarch64-* need a real aarch64 host (placement rule: hardware)
+          # so the packaged binary and console smoke execute the artifact.
+          # ubicloud-standard-16-arm is that host: Ubuntu 24.04 arm64, alias of
+          # ubicloud-standard-16-arm-ubuntu-2404 (16 vCPU, 48 GB). linux-x86_64-* stay on the
+          # ARC sm-standard-4 pool. GNU rows keep cross=true so cargo-zigbuild
+          # pins the glibc 2.31 floor on the same architecture; musl stays a
+          # native cargo build.
           all='{"include":[
             {"target_id":"linux-x86_64-musl","os":"sm-standard-4","target":"x86_64-unknown-linux-musl","cross":false,"platform":"linux","rustflags":"","features":"","glibc":""},
-            {"target_id":"linux-aarch64-musl","os":"sm-standard-4-arm","target":"aarch64-unknown-linux-musl","cross":false,"platform":"linux","rustflags":"","features":"","glibc":""},
+            {"target_id":"linux-aarch64-musl","os":"ubicloud-standard-16-arm","target":"aarch64-unknown-linux-musl","cross":false,"platform":"linux","rustflags":"","features":"","glibc":""},
             {"target_id":"linux-x86_64-gnu","os":"sm-standard-4","target":"x86_64-unknown-linux-gnu","cross":true,"platform":"linux","rustflags":"","features":"pyroscope","glibc":"2.31"},
-            {"target_id":"linux-aarch64-gnu","os":"sm-standard-4-arm","target":"aarch64-unknown-linux-gnu","cross":true,"platform":"linux","rustflags":"","features":"pyroscope","glibc":"2.31"},
+            {"target_id":"linux-aarch64-gnu","os":"ubicloud-standard-16-arm","target":"aarch64-unknown-linux-gnu","cross":true,"platform":"linux","rustflags":"","features":"pyroscope","glibc":"2.31"},
             {"target_id":"macos-aarch64","os":"macos-latest","target":"aarch64-apple-darwin","cross":false,"platform":"macos","rustflags":"","features":"pyroscope","glibc":""},
             {"target_id":"windows-x86_64","os":"windows-latest","target":"x86_64-pc-windows-msvc","cross":false,"platform":"windows","rustflags":"","features":"","glibc":""}
           ]}'
@@ -244,7 +250,11 @@ jobs:
         with:
           rust-version: stable
           target: ${{ matrix.target }}
-          cache-shared-key: build-${{ matrix.target }}${{ matrix.glibc && format('-glibc-{0}', matrix.glibc) || '' }}
+          # rust-cache keys include OS and architecture but not the runner image.
+          # The Ubicloud aarch64 legs therefore use their own suffix so they
+          # cannot restore a cache written on a different arm64 image.
+          # x86_64, macOS, and Windows keys stay unsuffixed.
+          cache-shared-key: build-${{ matrix.target }}${{ matrix.glibc && format('-glibc-{0}', matrix.glibc) || '' }}${{ matrix.os == 'ubicloud-standard-16-arm' && '-ubicloud-arm' || '' }}
           # main only. A cache saved on refs/tags/X is scoped to that tag: no
           # other tag, no main run and no PR can restore it, so every release
           # cycle wrote up to 12 entries of 1-2GB (preview tag plus final tag,
@@ -764,9 +774,9 @@ jobs:
             exit 0
           fi
 
-          # The self-hosted Linux runners do not ship the aws CLI (GitHub-hosted
-          # images did). Install the official CLI bundle on demand so R2 uploads
-          # survive a fresh runner without relying on a distro package.
+          # ARC pods (the x86_64 legs) do not ship the aws CLI. Ubicloud arm64
+          # images follow the GitHub runner image and often do. Install the
+          # official CLI bundle when it is missing so R2 uploads work on either.
           if ! command -v aws >/dev/null 2>&1; then
             echo "aws CLI not found on runner; installing..."
             if command -v apt-get >/dev/null 2>&1; then

.github/workflows/coverage.yml

@@ -45,7 +45,7 @@ on:
     # 07:00 UTC Sunday — staggered clear of the other Sunday crons: ci (00:00),
     # build (01:00), e2e-s3tests (02:00), audit (03:00), nix-flake-update
     # (05:00), mint (06:00), and the daily fuzz (02:00), minio-interop (03:17),
-    # e2e-replication-nightly (04:00) and performance-ab (06:00) lanes.
+    # and e2e-replication-nightly (04:00) lanes.
     - cron: "43 7 * * 0"
 
 concurrency:

.github/workflows/e2e-replication-nightly.yml

@@ -38,8 +38,8 @@ name: e2e-nightly
 on:
   workflow_dispatch:
   schedule:
-    # 04:00 UTC nightly — staggered clear of fuzz/e2e-s3tests (02:00),
-    # stale (01:30) and performance-ab (06:00).
+    # 04:00 UTC nightly — staggered clear of fuzz/e2e-s3tests (02:00)
+    # and stale (01:30).
     - cron: "29 4 * * *"
 
 # Only alert-on-failure needs more than read access; it declares its own

.github/workflows/performance-ab.yml

@@ -1,384 +0,0 @@
-# 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.
-
-# Warp A/B budget gate for the hotpath series (rustfs/backlog#935 HP-14, item 4).
-#
-# Two entry points, honestly scoped:
-#   * schedule (nightly, on main): post-merge detection — catches a regression
-#     within 24h of landing, not before merge.
-#   * workflow_dispatch: an explicitly selected trusted ref.
-# The dispatch input can run the gate with --allow-regression so a deliberate
-# correctness cost (e.g. the #4221 fsync durability fix) is recorded, not
-# blocked (rustfs/backlog#935 correction 1).
-
-name: Performance A/B
-
-on:
-  schedule:
-    - cron: "31 6 * * *" # 06:31 UTC nightly, against main
-  workflow_dispatch:
-    inputs:
-      duration:
-        description: "warp duration per round"
-        required: false
-        default: "12s"
-        type: string
-      allow_regression:
-        description: "Pass the gate despite a FAIL (deliberate tradeoff)"
-        required: false
-        default: false
-        type: boolean
-permissions:
-  actions: read
-  contents: read
-
-env:
-  CARGO_TERM_COLOR: always
-  RUST_BACKTRACE: 1
-
-jobs:
-  warp-ab:
-    if: github.repository == 'rustfs/rustfs'
-    name: Warp A/B budget gate
-    runs-on: sm-standard-2
-    # A normal nightly restores the last successful binary and builds only the
-    # candidate; daily access keeps that cache warm. A cache miss may build both
-    # and needs room for the A/B run plus artifact and cache publication.
-    timeout-minutes: 180
-    steps:
-      - name: Checkout repository
-        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-          fetch-depth: 0 # baseline may be an earlier successful scheduled head
-
-      - name: Setup Rust environment
-        uses: ./.github/actions/setup
-        with:
-          rust-version: stable
-          cache-shared-key: warp-ab-${{ hashFiles('**/Cargo.lock') }}
-          cache-save-if: ${{ github.ref == 'refs/heads/main' }}
-
-      - name: Install warp
-        run: |
-          set -euo pipefail
-          WARP_VERSION="1.0.0"
-          curl -fsSL "https://github.com/minio/warp/releases/download/v${WARP_VERSION}/warp_Linux_x86_64.tar.gz" \
-            | sudo tar -xz -C /usr/local/bin warp
-          warp --version
-
-      - name: Decide exemption
-        id: exempt
-        env:
-          INPUT_ALLOW_REGRESSION: ${{ github.event.inputs.allow_regression }}
-        run: |
-          allow="false"
-          if [[ "$INPUT_ALLOW_REGRESSION" == "true" ]]; then
-            allow="true"
-          fi
-          echo "allow_regression=$allow" >> "$GITHUB_OUTPUT"
-
-      # A failed regression run must keep comparing against the last known-good
-      # scheduled head. Otherwise the next nightly would absorb the regression
-      # into its baseline and turn green without a fix.
-      - name: Find last successful scheduled baseline
-        id: scheduled_baseline
-        if: github.event_name == 'schedule'
-        uses: actions/github-script@3a2844b7e9c422d3c10d287c895573f7108da1b3 # v9.0.0
-        with:
-          result-encoding: string
-          script: |
-            const { data } = await github.rest.actions.listWorkflowRuns({
-              owner: context.repo.owner,
-              repo: context.repo.repo,
-              workflow_id: "performance-ab.yml",
-              event: "schedule",
-              status: "success",
-              per_page: 1,
-            });
-            return data.workflow_runs[0]?.head_sha ?? "";
-
-      # Manual runs compare a selected ref with current main. Scheduled runs
-      # compare current main with the last successful scheduled head. With no
-      # history, the first run measures the candidate against itself and seeds
-      # that head only if the complete rig succeeds.
-      - name: Resolve baseline / candidate commits
-        id: commits
-        env:
-          SCHEDULED_BASELINE_SHA: ${{ steps.scheduled_baseline.outputs.result }}
-        run: |
-          set -euo pipefail
-          candidate_sha="$(git rev-parse HEAD)"
-          if [[ "${{ github.event_name }}" == "schedule" ]]; then
-            baseline_sha="${SCHEDULED_BASELINE_SHA:-$candidate_sha}"
-          else
-            baseline_sha="$(git rev-parse origin/main)"
-          fi
-          git cat-file -e "${baseline_sha}^{commit}"
-          if ! git merge-base --is-ancestor "$baseline_sha" "$candidate_sha"; then
-            echo "::error::baseline $baseline_sha is not an ancestor of candidate $candidate_sha; update the selected ref before comparing" >&2
-            exit 1
-          fi
-          echo "baseline_sha=$baseline_sha" >> "$GITHUB_OUTPUT"
-          echo "candidate_sha=$candidate_sha" >> "$GITHUB_OUTPUT"
-          echo "baseline commit: $baseline_sha"
-          echo "candidate commit: $candidate_sha"
-
-      # Exact-key restore of the candidate binary saved by its successful
-      # scheduled run. A miss leaves cache-hit unset and falls back to a source
-      # build of that known-good head.
-      - name: Restore cached baseline binary
-        id: baseline_cache
-        uses: actions/cache/restore@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
-        with:
-          path: baseline-bin/rustfs
-          key: rustfs-baseline-${{ steps.commits.outputs.baseline_sha }}
-
-      # Self-heal: on a nightly/dispatch run where the candidate commit IS the
-      # baseline commit, a cache miss would make the rig build the same commit
-      # twice (~65min per side with the post-#4806 LTO profile — no job budget
-      # fits that). Build it once here, reuse it for both phases, and save it
-      # back to the cache so the next run hits.
-      - name: Build baseline on cache miss (same-commit self-heal)
-        id: selfheal
-        if: >-
-          steps.baseline_cache.outputs.cache-hit != 'true' &&
-          steps.commits.outputs.baseline_sha == steps.commits.outputs.candidate_sha
-        run: |
-          set -euo pipefail
-          cargo build --release --bin rustfs
-          mkdir -p baseline-bin
-          cp target/release/rustfs baseline-bin/rustfs
-          echo "built=true" >> "$GITHUB_OUTPUT"
-
-      - name: Build baseline on cache miss (different candidate)
-        id: baseline_build
-        if: >-
-          steps.baseline_cache.outputs.cache-hit != 'true' &&
-          steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
-        run: |
-          set -euo pipefail
-          baseline_root="$RUNNER_TEMP/rustfs-baseline-${{ github.run_id }}"
-          baseline_target="$RUNNER_TEMP/rustfs-baseline-target-${{ github.run_id }}"
-          git worktree add --detach "$baseline_root" "${{ steps.commits.outputs.baseline_sha }}"
-          cargo build --release --manifest-path "$baseline_root/Cargo.toml" --bin rustfs --target-dir "$baseline_target"
-          mkdir -p baseline-bin
-          cp "$baseline_target/release/rustfs" baseline-bin/rustfs
-          git worktree remove --force "$baseline_root"
-          echo "built=true" >> "$GITHUB_OUTPUT"
-
-      - name: Build candidate binary
-        id: candidate_build
-        if: steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
-        run: |
-          set -euo pipefail
-          cargo build --release --bin rustfs
-          mkdir -p candidate-bin
-          cp target/release/rustfs candidate-bin/rustfs
-          echo "built=true" >> "$GITHUB_OUTPUT"
-
-      - name: Save self-healed baseline to cache
-        if: steps.selfheal.outputs.built == 'true' || steps.baseline_build.outputs.built == 'true'
-        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
-        with:
-          path: baseline-bin/rustfs
-          key: rustfs-baseline-${{ steps.commits.outputs.baseline_sha }}
-
-      - name: Run warp A/B and gate
-        id: ab
-        env:
-          INPUT_DURATION: ${{ github.event.inputs.duration }}
-        run: |
-          set -euo pipefail
-          # The formal runner executes A1 baseline -> B1 candidate -> B2 candidate
-          # -> A2 baseline for each workload and drive-sync cell. It requires three
-          # rounds per leg to emit tail latency and error-rate evidence.
-          # --health-timeout 180 outlasts the server's own 120s startup-readiness
-          # budget, which the rig's previous 60s health poll undershot (the first
-          # two nightly failures). perf-6 recalibrates these once the noise study
-          # lands.
-          duration="${INPUT_DURATION:-12s}"
-          baseline_sha="${{ steps.commits.outputs.baseline_sha }}"
-          candidate_sha="${{ steps.commits.outputs.candidate_sha }}"
-          baseline_hit="${{ steps.baseline_cache.outputs.cache-hit }}"
-          selfheal_built="${{ steps.selfheal.outputs.built }}"
-          baseline_built="${{ steps.baseline_build.outputs.built }}"
-          candidate_built="${{ steps.candidate_build.outputs.built }}"
-
-          args=(--duration "$duration" --rounds 3 --cooldown 5 --health-timeout 180 --baseline-revision "$baseline_sha" --candidate-revision "$candidate_sha")
-
-          if [[ "$baseline_hit" == "true" || "$selfheal_built" == "true" || "$baseline_built" == "true" ]]; then
-            chmod +x baseline-bin/rustfs
-            base_bin="$PWD/baseline-bin/rustfs"
-            args+=(--baseline-bin "$base_bin")
-            if [[ "$baseline_hit" == "true" ]]; then
-              base_src="actions-cache (rustfs-baseline-$baseline_sha)"
-            elif [[ "$selfheal_built" == "true" ]]; then
-              base_src="source build (cache self-heal, saved as rustfs-baseline-$baseline_sha)"
-            else
-              base_src="isolated baseline source build (saved as rustfs-baseline-$baseline_sha)"
-            fi
-            if [[ "$candidate_sha" == "$baseline_sha" ]]; then
-              # No commits landed since the last successful baseline, so reuse
-              # the one binary for both phases and measure only rig drift.
-              args+=(--candidate-bin "$base_bin")
-              cand_src="same binary as baseline (same commit)"
-            elif [[ "$candidate_built" == "true" ]]; then
-              chmod +x candidate-bin/rustfs
-              args+=(--candidate-bin "$PWD/candidate-bin/rustfs")
-              cand_src="source build of the checked-out ref"
-            else
-              echo "::error::candidate binary was not built" >&2
-              exit 2
-            fi
-          else
-            echo "::error::baseline binary was not restored or built" >&2
-            exit 2
-          fi
-
-          echo "baseline binary: $base_src"
-          echo "candidate binary: $cand_src"
-
-          if [[ "${{ steps.exempt.outputs.allow_regression }}" == "true" ]]; then
-            args+=(--allow-regression --exemption-reason "workflow dispatch override")
-          fi
-          # Do not let a gate FAIL abort the job here; capture status and surface
-          # it after the step summary is written.
-          set +e
-          bash scripts/run_hotpath_warp_abba.sh "${args[@]}"
-          echo "status=$?" >> "$GITHUB_OUTPUT"
-          set -e
-          # Locate the newest run dir + gate.md for the summary/comment/artifact
-          # steps. On a startup failure there is no gate.md, but the run dir still
-          # holds server-logs/ for diagnosis.
-          # Run dirs are UTC-timestamp names (no special chars); ls is safe here.
-          # shellcheck disable=SC2012
-          run_dir="$(ls -td target/hotpath-abba/*/ 2>/dev/null | head -n1 || true)"
-          echo "run_dir=${run_dir%/}" >> "$GITHUB_OUTPUT"
-          # shellcheck disable=SC2012
-          gate_md="$(ls -t target/hotpath-abba/*/candidate_gate.md 2>/dev/null | head -n1 || true)"
-          echo "gate_md=$gate_md" >> "$GITHUB_OUTPUT"
-
-      - name: Upload A/B results
-        if: always()
-        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
-        with:
-          name: hotpath-warp-ab-${{ github.run_number }}
-          # Includes per-cell median_summary.csv / baseline_compare.csv, both gates,
-          # and server-logs/ (rustfs.log + startup env per phase) so a failed run
-          # is diagnosable. Short retention: this is churny nightly debug data.
-          path: target/hotpath-abba/
-          if-no-files-found: warn
-          retention-days: 14
-
-      - name: Write gate summary
-        if: always()
-        run: |
-          set -euo pipefail
-          status="${{ steps.ab.outputs.status }}"
-          gate_md="${{ steps.ab.outputs.gate_md }}"
-          run_dir="${{ steps.ab.outputs.run_dir }}"
-          {
-            echo "## Hotpath warp A/B — run ${{ github.run_number }}"
-            echo
-            if [[ "$status" == "0" ]]; then
-              echo "Rig/gate exit: \`0\` (pass or warn)."
-            else
-              echo "Rig/gate exit: \`${status:-unknown}\` — **FAILED**."
-            fi
-            echo
-            if [[ -n "$gate_md" && -f "$gate_md" ]]; then
-              cat "$gate_md"
-            else
-              echo "No \`gate.md\` produced — the rig failed **before** the gate"
-              echo "(most likely server startup / health). Failing phase(s) below;"
-              echo "full logs in the \`hotpath-warp-ab-${{ github.run_number }}\` artifact."
-              if [[ -n "$run_dir" && -d "$run_dir/server-logs" ]]; then
-                for f in "$run_dir"/server-logs/*.log; do
-                  [[ -f "$f" ]] || continue
-                  echo
-                  echo "<details><summary>$(basename "$f")</summary>"
-                  echo
-                  echo '```'
-                  tail -n 30 "$f"
-                  echo '```'
-                  echo "</details>"
-                done
-              fi
-            fi
-          } >> "$GITHUB_STEP_SUMMARY"
-
-      - name: Stage successful candidate baseline
-        if: >-
-          steps.ab.outputs.status == '0' &&
-          steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
-        run: |
-          set -euo pipefail
-          cp candidate-bin/rustfs baseline-bin/rustfs
-
-      - name: Cache successful candidate baseline
-        if: >-
-          steps.ab.outputs.status == '0' &&
-          steps.commits.outputs.baseline_sha != steps.commits.outputs.candidate_sha
-        uses: actions/cache/save@55cc8345863c7cc4c66a329aec7e433d2d1c52a9 # v6
-        with:
-          path: baseline-bin/rustfs
-          key: rustfs-baseline-${{ steps.commits.outputs.candidate_sha }}
-
-      # Scheduled failure alerting is handled by the alert-on-failure job below
-      # (perf-2 consuming ci-8's schedule-failure-issue composite action).
-
-      - name: Enforce gate
-        if: always()
-        run: |
-          status="${{ steps.ab.outputs.status }}"
-          if [[ -z "$status" ]]; then
-            echo "::error::warp A/B setup failed before the rig ran. Check the first failed workflow step." >&2
-            exit 1
-          fi
-          if [[ "$status" != "0" ]]; then
-            echo "::error::warp A/B budget gate failed (exit $status). See the step summary / gate.md artifact." >&2
-            exit "$status"
-          fi
-          echo "warp A/B budget gate passed."
-
-  alert-on-failure:
-    name: Alert on scheduled failure
-    needs: [warp-ab]
-    # `always()` is required: without it this job is skipped when a needed
-    # job fails. Alerts only for scheduled (nightly) runs (backlog#1149
-    # ci-8); manual dispatch failures are already watched by a human.
-    # `cancelled` is included alongside `failure` on purpose: a job that hits
-    # timeout-minutes ends as `cancelled`, and the 2026-07-11..07-14 nightly
-    # timeouts went silent precisely because the guard was failure-only. The
-    # composite action already reports cancelled/timed-out jobs in the issue
-    # body.
-    if: >-
-      github.repository == 'rustfs/rustfs' && (
-        always() && github.event_name == 'schedule' &&
-        (contains(needs.*.result, 'failure') || contains(needs.*.result, 'cancelled'))
-      )
-    runs-on: ubuntu-latest
-    timeout-minutes: 10
-    permissions:
-      contents: read
-      issues: write
-    steps:
-      - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
-        with:
-          persist-credentials: false
-      - name: Open or update failure-tracking issue
-        uses: ./.github/actions/schedule-failure-issue
-        with:
-          github-token: ${{ secrets.BACKLOG_ISSUE_TOKEN }}

.github/workflows/schedule-failure-alert-drill.yml

@@ -16,8 +16,8 @@
 #
 # Forces a job failure and then runs .github/actions/schedule-failure-issue
 # through the exact `needs` + `always() && contains(needs.*.result,
-# 'failure')` wiring used by the real consumers (e2e-s3tests, mint, fuzz,
-# performance-ab). Dispatch it twice to verify both the issue-creation and
+# 'failure')` wiring used by the real consumers (e2e-s3tests, mint, fuzz).
+# Dispatch it twice to verify both the issue-creation and
 # the dedupe-comment paths, then close the resulting
 # "[scheduled-failure] Schedule Failure Alert Drill" issue in rustfs/backlog
 # (the default target repository). Requires the BACKLOG_ISSUE_TOKEN secret

.github/workflows/scheduled-validation-watchdog.yml

@@ -28,7 +28,6 @@ on:
       - "mint"
       - "minio-interop"
       - "Nightly GNU Build"
-      - "Performance A/B"
       - "Runner Hygiene"
     types: [completed]
 

crates/e2e_test/src/distributed/chaos_test.rs

@@ -13,18 +13,20 @@
 // limitations under the License.
 
 use super::harness::{
-    DistCluster, DistLayout, TestResult, assert_object_bytes, payload_for, put_object, retrying_get_equals, unique_bucket,
-    wait_for_ready, wait_until,
+    DistCluster, DistLayout, TestResult, assert_object_bytes, cluster_admin_ok, get_object_bytes, payload_for, put_object,
+    retrying_get_equals, retrying_put, s3_probe_client, s3_probe_error_code, unique_bucket, wait_for_ready, wait_until,
 };
 use crate::chaos::{census_object_version_on_disk, signed_admin_post};
 use crate::common::{build_test_s3_config, init_logging};
 use crate::fault_proxy::FaultMode;
 use aws_sdk_s3::Client;
+use aws_sdk_s3::error::SdkError;
+use aws_sdk_s3::operation::{get_object::GetObjectError, put_object::PutObjectError};
 use std::path::PathBuf;
 use std::sync::Arc;
 use std::time::Duration;
 use tokio::sync::{Barrier, mpsc};
-use tokio::time::timeout;
+use tokio::time::{Instant, timeout};
 
 #[tokio::test]
 async fn kill_and_restart_node_preserves_objects() -> TestResult {
@@ -44,6 +46,133 @@ async fn kill_and_restart_node_preserves_objects() -> TestResult {
     Ok(())
 }
 
+#[tokio::test]
+async fn ec4_partial_node_return_restores_read_write_quorum() -> TestResult {
+    init_logging();
+    let mut dist = DistCluster::start_with_env(
+        DistLayout::FourByFour,
+        &[
+            ("RUSTFS_ERASURE_SET_DRIVE_COUNT", "16"),
+            ("RUSTFS_STORAGE_CLASS_STANDARD", "EC:4"),
+        ],
+    )
+    .await?;
+    let storage = cluster_admin_ok(&dist.cluster, http::Method::GET, "/rustfs/admin/v3/storageinfo", None).await?;
+    let storage: serde_json::Value = serde_json::from_str(&storage)?;
+    let storage: rustfs_madmin::StorageInfo = serde_json::from_value(storage["info"].clone())?;
+    assert_eq!(storage.disks.len(), 16);
+    assert_eq!(storage.backend.total_sets, vec![1]);
+    assert_eq!(storage.backend.drives_per_set, vec![16]);
+    assert_eq!(storage.backend.standard_sc_parity, Some(4));
+
+    let bucket = unique_bucket("partial-ec4");
+    dist.create_bucket(&bucket).await?;
+    let client = s3_probe_client(&dist.client(0)?);
+    let baseline = payload_for("partial-recovery/baseline", 8 * 1024 * 1024);
+    put_object(&client, &bucket, "baseline.bin", baseline.clone()).await?;
+    wait_until(
+        Duration::from_secs(30),
+        || async {
+            for node in &dist.cluster.nodes {
+                for directory in &node.data_dirs {
+                    let census = census_object_version_on_disk(std::path::Path::new(directory), &bucket, "baseline.bin", None)?;
+                    if !census.is_complete() {
+                        return Ok(false);
+                    }
+                    assert_eq!(census.data_blocks, Some(12), "the object must require twelve data shards");
+                    assert_eq!(census.parity_blocks, Some(4), "the object must have four parity shards");
+                    assert!(!census.expected_part_numbers.is_empty(), "the baseline must use physical shards");
+                }
+            }
+            Ok(true)
+        },
+        "the baseline is complete on all sixteen EC:4 drives",
+    )
+    .await?;
+
+    dist.cluster.stop_node(2)?;
+    dist.cluster.stop_node(3)?;
+    let read_error = timeout(Duration::from_secs(60), get_object_bytes(&client, &bucket, "baseline.bin"))
+        .await?
+        .expect_err("eight unavailable drives must prevent the baseline GET");
+    assert!(
+        matches!(
+            s3_probe_error_code(read_error.as_ref()),
+            Some("SlowDown" | "SlowDownRead" | "ServiceUnavailable")
+        ),
+        "quorum loss must produce an availability error: {read_error:?}"
+    );
+    assert_eq!(
+        read_error
+            .downcast_ref::<SdkError<GetObjectError>>()
+            .and_then(SdkError::raw_response)
+            .map(|response| response.status().as_u16()),
+        Some(503),
+        "read quorum loss must remain a service availability failure"
+    );
+    let write_error = timeout(Duration::from_secs(60), put_object(&client, &bucket, "unavailable.bin", baseline.clone()))
+        .await?
+        .expect_err("eight unavailable drives must prevent a successful PUT");
+    assert!(
+        matches!(s3_probe_error_code(write_error.as_ref()), Some("SlowDown" | "ServiceUnavailable")),
+        "quorum loss must produce an availability error: {write_error:?}"
+    );
+    assert_eq!(
+        write_error
+            .downcast_ref::<SdkError<PutObjectError>>()
+            .and_then(SdkError::raw_response)
+            .map(|response| response.status().as_u16()),
+        Some(503),
+        "write quorum loss must remain a service availability failure"
+    );
+
+    // Recover only four drives. The remaining node must stay stopped while
+    // every live coordinator serves old data and acknowledges new writes.
+    dist.cluster.start_node_process(2).await?;
+    let recovery_deadline = Instant::now() + Duration::from_secs(90);
+    let mut recovered_objects = Vec::new();
+    for node in 0..3 {
+        assert!(dist.cluster.nodes[3].process.is_none(), "the fourth node must remain unavailable");
+        let client = s3_probe_client(&dist.client(node)?);
+        retrying_get_equals(
+            &client,
+            &bucket,
+            "baseline.bin",
+            &baseline,
+            recovery_deadline.saturating_duration_since(Instant::now()),
+        )
+        .await?;
+        let key = format!("partial-{node}.bin");
+        let body = payload_for(&key, 1024 * 1024);
+        retrying_put(
+            &client,
+            &bucket,
+            &key,
+            body.clone(),
+            recovery_deadline.saturating_duration_since(Instant::now()),
+        )
+        .await?;
+        recovered_objects.push((key, body));
+    }
+    for node in 0..3 {
+        assert!(dist.cluster.nodes[3].process.is_none(), "reads must complete before all drives return");
+        let client = s3_probe_client(&dist.client(node)?);
+        assert_object_bytes(&client, &bucket, "baseline.bin", &baseline).await?;
+        for (key, body) in &recovered_objects {
+            assert_object_bytes(&client, &bucket, key, body).await?;
+        }
+    }
+
+    dist.cluster.start_node(3).await?;
+    wait_for_ready(&dist.cluster).await?;
+    let client = s3_probe_client(&dist.client(3)?);
+    assert_object_bytes(&client, &bucket, "baseline.bin", &baseline).await?;
+    for (key, body) in &recovered_objects {
+        assert_object_bytes(&client, &bucket, key, body).await?;
+    }
+    Ok(())
+}
+
 #[tokio::test]
 async fn full_cluster_restart_preserves_objects() -> TestResult {
     init_logging();

crates/e2e_test/src/distributed/harness.rs

@@ -1181,11 +1181,16 @@ pub(crate) async fn retrying_get_equals(
                 )
                 .into()),
                 Err(error) => {
-                    if matches!(s3_probe_error_code(error.as_ref()), Some("SlowDown" | "ServiceUnavailable")) {
-                        Ok(false)
-                    } else {
-                        Err(error)
-                    }
+                    let retryable = match s3_probe_error_code(error.as_ref()) {
+                        Some("SlowDown" | "ServiceUnavailable") => true,
+                        // GET preserves this compatibility code for a lost data quorum.
+                        Some("SlowDownRead") => error
+                            .downcast_ref::<SdkError<GetObjectError>>()
+                            .and_then(SdkError::raw_response)
+                            .is_some_and(|response| response.status().as_u16() == 503),
+                        _ => false,
+                    };
+                    if retryable { Ok(false) } else { Err(error) }
                 }
             }
         },
@@ -1324,6 +1329,62 @@ mod retry_tests {
         Ok(())
     }
 
+    #[tokio::test(start_paused = true)]
+    async fn retrying_get_allows_read_quorum_recovery() -> TestResult {
+        use aws_sdk_s3::config::Region;
+        use aws_smithy_http_client::test_util::{ReplayEvent, StaticReplayClient};
+        use aws_smithy_types::body::SdkBody;
+
+        let endpoint = "http://read-quorum.example.invalid";
+        let event = |status: u16, body: &'static str| {
+            ReplayEvent::new(
+                http::Request::builder()
+                    .method(Method::GET)
+                    .uri(format!("{endpoint}/retry-probe/key?x-id=GetObject"))
+                    .body(SdkBody::empty())
+                    .expect("build replay request"),
+                http::Response::builder()
+                    .status(status)
+                    .header("content-type", "application/xml")
+                    .header("content-length", body.len())
+                    .body(SdkBody::from(body))
+                    .expect("build replay response"),
+            )
+        };
+        for status in [503, 403, 500] {
+            let transport = StaticReplayClient::new(vec![
+                event(status, "<Error><Code>SlowDownRead</Code></Error>"),
+                event(200, "expected bytes"),
+            ]);
+            let client = Client::from_conf(
+                aws_sdk_s3::Config::builder()
+                    .behavior_version_latest()
+                    .credentials_provider(Credentials::new(FAKE_ACCESS_KEY, FAKE_SECRET_KEY, None, None, "read-quorum-test"))
+                    .region(Region::new("us-east-1"))
+                    .endpoint_url(endpoint)
+                    .force_path_style(true)
+                    .retry_config(RetryConfig::disabled())
+                    .http_client(transport.clone())
+                    .build(),
+            );
+            let result = retrying_get_equals(&client, "retry-probe", "key", b"expected bytes", Duration::from_secs(5)).await;
+            if status == 503 {
+                result?;
+                assert_eq!(
+                    transport.actual_requests().count(),
+                    2,
+                    "the probe must retry the read-quorum error exactly once"
+                );
+                transport.assert_requests_match(&[]);
+            } else {
+                let error = result.expect_err("the compatibility code must not hide a non-503 failure");
+                assert_eq!(s3_probe_error_code(error.as_ref()), Some("SlowDownRead"));
+                assert_eq!(transport.actual_requests().count(), 1, "non-availability errors must fail immediately");
+            }
+        }
+        Ok(())
+    }
+
     #[tokio::test]
     async fn object_polls_distinguish_committed_and_replicated_keys() -> TestResult {
         let target = FakeS3Target::start().await?;

crates/e2e_test/src/distributed/multipart_fence_acceptance_test.rs

@@ -86,8 +86,7 @@ async fn fixture_from_binary(
         ("NO_PROXY", "127.0.0.1,::1,localhost"),
         ("HTTP_PROXY", ""),
         ("HTTPS_PROXY", ""),
-        ("RUSTFS_OBJECT_LOCK_DIAG_ENABLE", "true"),
-        ("RUST_LOG", "warn,rustfs_ecstore::object_lock_diag=trace"),
+        ("RUST_LOG", "warn"),
     ] {
         cluster.set_env(name, value);
     }
@@ -971,8 +970,7 @@ async fn phase_quorum_matrix(
                         .key(&key)
                         .body(ByteStream::from(previous.clone()))
                         .send()
-                        .await
-                        .map_err(|error| format!("setup PutObject phase={phase} key={key}: {error:?}"))?;
+                        .await?;
                     Some(previous)
                 } else {
                     None
@@ -1001,8 +999,7 @@ async fn phase_quorum_matrix(
                             .part_number(1)
                             .body(ByteStream::from(body.clone()))
                             .send()
-                            .await
-                            .map_err(|error| format!("setup UploadPart phase={phase} key={key}: {error:?}"))?,
+                            .await?,
                     )
                 } else {
                     None
@@ -1083,8 +1080,7 @@ async fn phase_quorum_matrix(
                             .upload_id(upload_id);
                         let result =
                             with_paused_peers(proxies, methods, &limits, &scope, async move { request.send().await }).await?;
-                        quorum_outcome(result, unavailable == unavailable_at_quorum, locks)
-                            .map_err(|error| format!("injected Abort phase={phase} key={key}: {error}"))?;
+                        quorum_outcome(result, unavailable == unavailable_at_quorum, locks)?;
                     }
                     _ => unreachable!(),
                 }
@@ -1642,6 +1638,191 @@ async fn late_complete_packets_cannot_publish_after_abort_and_bucket_recreation(
             "refused late packets must not recreate staging"
         );
     }
+    late_part_packets_preserve_retired_and_fresh_uploads().await?;
+    Ok(())
+}
+
+async fn late_part_packets_preserve_retired_and_fresh_uploads() -> TestResult {
+    for (method, settles) in [("PreparePartTransaction", false), ("SettlePartTransaction", true)] {
+        for recreate_bucket in [false, true] {
+            let (cluster, proxies, clients, bucket) = fixture(1, "EC:2").await?;
+            let key = "late-part-transaction";
+            let previous = payload_for("committed-before-late-part", 4096);
+            clients[0]
+                .put_object()
+                .bucket(&bucket)
+                .key(key)
+                .body(ByteStream::from(previous.clone()))
+                .send()
+                .await?;
+            let upload = clients[0].create_multipart_upload().bucket(&bucket).key(key).send().await?;
+            let upload_id = upload.upload_id().ok_or("missing original upload ID")?;
+            clients[0]
+                .upload_part()
+                .bucket(&bucket)
+                .key(key)
+                .upload_id(upload_id)
+                .part_number(1)
+                .body(ByteStream::from(vec![0x41; 9 * 1024 * 1024]))
+                .send()
+                .await?;
+            let path = upload_path_scope(&bucket, key, upload_id)?;
+            let hash = sha256_hex(format!("{bucket}/{key}").as_bytes());
+            let roots = cluster
+                .nodes
+                .iter()
+                .flat_map(|node| &node.data_dirs)
+                .map(|root| Path::new(root).join(RUSTFS_META_MULTIPART_BUCKET).join(&path))
+                .collect::<Vec<_>>();
+            assert_eq!(roots.len(), 4);
+            let original = multipart_sources(&cluster, Some(&hash))?;
+            assert_eq!(original.len(), 4);
+            assert!(
+                original.values().all(|source| source
+                    .files
+                    .keys()
+                    .any(|path| path.file_name().and_then(|name| name.to_str()) == Some("part.1"))),
+                "all selected peers originally own the acknowledged part"
+            );
+            let scope = upload_scope(upload_id)?;
+            // Retain only the original attempt; recovery of the same upload
+            // must be able to issue fresh settlement requests through the gate.
+            let pauses = proxies[1..3]
+                .iter()
+                .map(|proxy| proxy.pause_late(&[method], 1, &scope))
+                .collect::<Result<Vec<_>, _>>()?;
+            let replacement = clients[0]
+                .upload_part()
+                .bucket(&bucket)
+                .key(key)
+                .upload_id(upload_id)
+                .part_number(1)
+                .body(ByteStream::from(vec![0x62; 8 * 1024 * 1024]));
+            let caller = tokio::spawn(async move { replacement.send().await });
+            wait_for_faults(&pauses, &[method]).await?;
+            let failure = tokio::time::timeout(DEADLINE, caller)
+                .await??
+                .expect_err("two retained peers cannot acknowledge part replacement");
+            assert_unavailable(&failure, "storage_write_quorum", None);
+            assert!(pauses.iter().all(|pause| pause.hits() == 1 && pause.responses() == 0));
+            let expected_actions = if settles { (1, 0) } else { (0, 0) };
+            assert!(
+                pauses.iter().all(|pause| pause.retained_settle_actions() == expected_actions),
+                "settlement capture must be a real Commit, not Rollback"
+            );
+            abort_after_fault(&clients[0], &bucket, key, upload_id, false).await?;
+            assert!(
+                roots.iter().all(|root| !root.exists()),
+                "ordinary Abort physically retires all original roots"
+            );
+            assert!(multipart_sources(&cluster, Some(&hash))?.is_empty());
+            for client in &clients {
+                assert_object_bytes_after_fault(client, &bucket, key, &previous).await?;
+            }
+            let protected = if recreate_bucket {
+                clients[0].delete_object().bucket(&bucket).key(key).send().await?;
+                clients[0].delete_bucket().bucket(&bucket).send().await?;
+                clients[1].create_bucket().bucket(&bucket).send().await?;
+                let fresh = payload_for("recreated-before-late-part", 4096);
+                clients[1]
+                    .put_object()
+                    .bucket(&bucket)
+                    .key(key)
+                    .body(ByteStream::from(fresh.clone()))
+                    .send()
+                    .await?;
+                fresh
+            } else {
+                previous
+            };
+            let fresh = clients[2].create_multipart_upload().bucket(&bucket).key(key).send().await?;
+            let fresh_id = fresh.upload_id().ok_or("missing fresh upload ID")?;
+            let fresh_body = payload_for("fresh-active-part", 4096);
+            let fresh_part = clients[2]
+                .upload_part()
+                .bucket(&bucket)
+                .key(key)
+                .upload_id(fresh_id)
+                .part_number(1)
+                .body(ByteStream::from(fresh_body.clone()))
+                .send()
+                .await?;
+            let before = multipart_sources(&cluster, Some(&hash))?;
+            assert_eq!(before.len(), 4, "the new upload exists on every original participant");
+            for pause in &pauses {
+                pause.release();
+            }
+            wait_until(
+                DEADLINE,
+                || async {
+                    Ok(pauses.iter().all(|pause| {
+                        pause.responses() == pause.hits()
+                            && if settles {
+                                pause.successful_mutation_replies() == pause.hits() && pause.refused_mutations() == 0
+                            } else {
+                                pause.refused_mutations() == pause.hits() && pause.successful_mutation_replies() == 0
+                            }
+                    }))
+                },
+                "every original part packet returns its typed authenticated terminal outcome",
+            )
+            .await?;
+            assert!(roots.iter().all(|root| !root.exists()), "late part packets cannot recreate retired roots");
+            assert!(
+                multipart_sources(&cluster, Some(&hash))? == before,
+                "late packets cannot change a fresh upload's shard or control bytes"
+            );
+            for client in &clients {
+                assert_object_bytes_after_fault(client, &bucket, key, &protected).await?;
+                let listed = client
+                    .list_parts()
+                    .bucket(&bucket)
+                    .key(key)
+                    .upload_id(fresh_id)
+                    .send()
+                    .await?;
+                assert_eq!(listed.parts().len(), 1);
+                assert_eq!(listed.parts()[0].e_tag(), fresh_part.e_tag());
+                let missing = client
+                    .list_parts()
+                    .bucket(&bucket)
+                    .key(key)
+                    .upload_id(upload_id)
+                    .send()
+                    .await
+                    .expect_err("old upload remains retired");
+                assert_eq!(missing.raw_response().expect("old upload service response").status().as_u16(), 404);
+                assert_eq!(missing.as_service_error().and_then(|error| error.meta().code()), Some("NoSuchUpload"));
+            }
+            clients[2]
+                .complete_multipart_upload()
+                .bucket(&bucket)
+                .key(key)
+                .upload_id(fresh_id)
+                .multipart_upload(
+                    CompletedMultipartUpload::builder()
+                        .parts(
+                            CompletedPart::builder()
+                                .part_number(1)
+                                .e_tag(fresh_part.e_tag().ok_or("missing fresh part ETag")?)
+                                .build(),
+                        )
+                        .build(),
+                )
+                .send()
+                .await?;
+            for client in &clients {
+                assert_object_bytes_after_fault(client, &bucket, key, &fresh_body).await?;
+            }
+            assert!(multipart_sources(&cluster, Some(&hash))?.is_empty());
+            assert_no_uploads_after_fault(&clients, &bucket).await?;
+            eprintln!(
+                "MPU late part: method={method}, recreate_bucket={recreate_bucket}, retired_roots=4, delivered=2, refused={}, successful_replies={}, fresh_sources_unchanged=4, verified_readers=4",
+                pauses.iter().map(RpcPause::refused_mutations).sum::<usize>(),
+                pauses.iter().map(RpcPause::successful_mutation_replies).sum::<usize>()
+            );
+        }
+    }
     Ok(())
 }
 

crates/e2e_test/src/rpc_fault_proxy.rs

@@ -27,8 +27,8 @@ use hyper_util::{
 };
 use prost::Message as _;
 use rustfs_protos::proto_gen::node_service::{
-    BatchGenerallyLockRequest, DeleteRequest, GenerallyLockRequest, ReadVersionRequest, RecoverMultipartSourceRequest,
-    RenameDataResponse,
+    BatchGenerallyLockRequest, DeleteRequest, GenerallyLockRequest, PreparePartTransactionResponse, ReadVersionRequest,
+    RecoverMultipartSourceRequest, RenameDataResponse, SettlePartTransactionRequest, SettlePartTransactionResponse,
 };
 use std::{
     collections::HashSet,
@@ -56,6 +56,35 @@ enum Disposition {
     Discard,
 }
 
+#[derive(Clone, Copy)]
+enum MutationReply {
+    None,
+    Rename,
+    PreparePart,
+    SettlePart,
+}
+
+impl MutationReply {
+    fn outcome(self, bytes: &[u8]) -> io::Result<Option<(bool, bool)>> {
+        let result = match self {
+            Self::None => return Ok(None),
+            Self::Rename => {
+                let reply = RenameDataResponse::decode(unary_payload(bytes)?).map_err(io::Error::other)?;
+                (reply.success, reply.error.is_some())
+            }
+            Self::PreparePart => {
+                let reply = PreparePartTransactionResponse::decode(unary_payload(bytes)?).map_err(io::Error::other)?;
+                (reply.success, reply.error.is_some())
+            }
+            Self::SettlePart => {
+                let reply = SettlePartTransactionResponse::decode(unary_payload(bytes)?).map_err(io::Error::other)?;
+                (reply.success, reply.error.is_some())
+            }
+        };
+        Ok(Some(result))
+    }
+}
+
 struct PauseState {
     methods: Vec<&'static str>,
     scope: Option<Vec<u8>>,
@@ -64,6 +93,9 @@ struct PauseState {
     limit: usize,
     retain_after_disconnect: bool,
     refused_mutations: AtomicUsize,
+    successful_mutation_replies: AtomicUsize,
+    settle_commits: AtomicUsize,
+    settle_rollbacks: AtomicUsize,
     responses: AtomicUsize,
     matched: AtomicUsize,
     blocked: AtomicUsize,
@@ -122,6 +154,17 @@ impl RpcPause {
         self.state.refused_mutations.load(Ordering::SeqCst)
     }
 
+    pub(crate) fn successful_mutation_replies(&self) -> usize {
+        self.state.successful_mutation_replies.load(Ordering::SeqCst)
+    }
+
+    pub(crate) fn retained_settle_actions(&self) -> (usize, usize) {
+        (
+            self.state.settle_commits.load(Ordering::SeqCst),
+            self.state.settle_rollbacks.load(Ordering::SeqCst),
+        )
+    }
+
     pub(crate) fn unsupported_source_inspection_routes(&self) -> HashSet<String> {
         self.state
             .unsupported_inspection_routes
@@ -394,6 +437,9 @@ impl RpcFaultProxy {
             limit,
             retain_after_disconnect,
             refused_mutations: AtomicUsize::new(0),
+            successful_mutation_replies: AtomicUsize::new(0),
+            settle_commits: AtomicUsize::new(0),
+            settle_rollbacks: AtomicUsize::new(0),
             responses: AtomicUsize::new(0),
             matched: AtomicUsize::new(0),
             blocked: AtomicUsize::new(0),
@@ -509,6 +555,7 @@ async fn forward_request(
         if let Some(method) = method.filter(|name| pause.methods.contains(name)) {
             let mut route = None;
             let mut lock_owners = Vec::new();
+            let mut settle_action = None;
             let matches = if let Some(scope) = pause.scope.as_deref() {
                 // Inspect bounded control routing while preserving the exact
                 // signed bytes and trailers forwarded to the backend.
@@ -522,12 +569,19 @@ async fn forward_request(
                     inspection = Some((pause.clone(), disk));
                 }
                 let mut matches = scope_matches(method, &bytes, scope)?;
+                if matches && method == "SettlePartTransaction" {
+                    settle_action = Some(
+                        SettlePartTransactionRequest::decode(unary_payload(&bytes)?)
+                            .map_err(io::Error::other)?
+                            .rollback,
+                    );
+                }
                 if matches && (pause.record_lock_owners || pause.lock_owner.is_some()) {
                     lock_owners = scoped_lock_owners(method, &bytes, scope)?;
                     if let Some(owner) = &pause.lock_owner {
                         matches = lock_owners.iter().any(|candidate| candidate == owner);
                         if !matches && pause.record_lock_owners {
-                            observed = Some((pause.clone(), None, std::mem::take(&mut lock_owners), false));
+                            observed = Some((pause.clone(), None, std::mem::take(&mut lock_owners), MutationReply::None));
                         }
                     }
                 }
@@ -549,6 +603,14 @@ async fn forward_request(
                 true
             };
             if matches && pause.matched.fetch_add(1, Ordering::SeqCst) < pause.limit {
+                if let Some(rollback) = settle_action {
+                    let counter = if rollback {
+                        &pause.settle_rollbacks
+                    } else {
+                        &pause.settle_commits
+                    };
+                    counter.fetch_add(1, Ordering::SeqCst);
+                }
                 let mut released = pause.release.subscribe();
                 pause.blocked.fetch_add(1, Ordering::SeqCst);
                 pause.reached.notify_one();
@@ -568,10 +630,14 @@ async fn forward_request(
                     pause,
                     route,
                     lock_owners,
-                    matches!(
-                        method,
-                        "RenameData" | "RenameMultipartData" | "RenameDataAtIncarnation" | "RenameDataWithQuotaFence"
-                    ),
+                    match method {
+                        "RenameData" | "RenameMultipartData" | "RenameDataAtIncarnation" | "RenameDataWithQuotaFence" => {
+                            MutationReply::Rename
+                        }
+                        "PreparePartTransaction" => MutationReply::PreparePart,
+                        "SettlePartTransaction" => MutationReply::SettlePart,
+                        _ => MutationReply::None,
+                    },
                 ));
             }
         }
@@ -604,7 +670,7 @@ async fn forward_request(
             .expect("inspection response routes")
             .insert(disk);
     }
-    if let Some((pause, route, lock_owners, rename)) = observed {
+    if let Some((pause, route, lock_owners, mutation_reply)) = observed {
         pause.responses.fetch_add(1, Ordering::SeqCst);
         if response.status().is_success() && response.headers().get("grpc-status").is_none_or(|status| status == "0") {
             pause.forwarded.fetch_add(1, Ordering::SeqCst);
@@ -626,10 +692,14 @@ async fn forward_request(
                 .headers
                 .get("grpc-status")
                 .or_else(|| trailers.as_ref()?.get("grpc-status"));
-            if rename && parts.status.is_success() && status.is_none_or(|status| status == "0") {
-                let result = RenameDataResponse::decode(unary_payload(&bytes)?).map_err(io::Error::other)?;
-                if !result.success && result.error.is_some() {
+            if parts.status.is_success()
+                && status.is_some_and(|status| status == "0")
+                && let Some((success, has_error)) = mutation_reply.outcome(&bytes)?
+            {
+                if !success && has_error {
                     pause.refused_mutations.fetch_add(1, Ordering::SeqCst);
+                } else if success && !has_error {
+                    pause.successful_mutation_replies.fetch_add(1, Ordering::SeqCst);
                 }
             }
             let body = Full::new(bytes)
@@ -676,6 +746,10 @@ mod tests {
     }
 
     async fn backend_with_status(status: Option<&'static str>) -> io::Result<Backend> {
+        backend_with_reply(status, None).await
+    }
+
+    async fn backend_with_reply(status: Option<&'static str>, reply: Option<Bytes>) -> io::Result<Backend> {
         let listener = TcpListener::bind("127.0.0.1:0").await?;
         let address = listener.local_addr()?;
         let received = Arc::new(AtomicUsize::new(0));
@@ -687,9 +761,11 @@ mod tests {
                     accepted = listener.accept() => {
                         let Ok((socket, _)) = accepted else { break };
                         let received = received_task.clone();
+                        let reply = reply.clone();
                         children.spawn(async move {
                             let service = service_fn(move |request: Request<Incoming>| {
                                 let received = received.clone();
+                                let reply = reply.clone();
                                 async move {
                                 received.fetch_add(1, Ordering::SeqCst);
                                 let authority = request.headers().get(http::header::HOST).cloned().or_else(|| {
@@ -698,6 +774,7 @@ mod tests {
                                     })
                                 });
                                 let body = request.into_body().collect().await?.to_bytes();
+                                let body = reply.unwrap_or(body);
                                 let mut trailers = http::HeaderMap::new();
                                 trailers.insert("grpc-status", http::HeaderValue::from_static(status.unwrap_or("0")));
                                 trailers.insert("x-proof", http::HeaderValue::from_static("unchanged-proof"));
@@ -1434,6 +1511,93 @@ mod tests {
         .await
         .expect("scoped discard deadline");
         assert_late_packet_ownership().await;
+        assert_late_part_reply_refusal().await;
+    }
+
+    async fn assert_late_part_reply_refusal() {
+        use rustfs_protos::proto_gen::node_service::{Error, PreparePartTransactionRequest};
+        tokio::time::timeout(Duration::from_secs(10), async {
+            // method, success, error, gRPC status, rollback, expected successes/refusals
+            for (method, success, has_error, status, rollback, expected) in [
+                ("PreparePartTransaction", false, true, None, false, (0, 1)),
+                ("PreparePartTransaction", true, false, None, false, (1, 0)),
+                ("PreparePartTransaction", false, false, None, false, (0, 0)),
+                ("SettlePartTransaction", true, false, None, false, (1, 0)),
+                ("SettlePartTransaction", false, true, None, true, (0, 1)),
+                ("SettlePartTransaction", true, true, None, false, (0, 0)),
+                ("SettlePartTransaction", false, true, Some("16"), false, (0, 0)),
+                ("PreparePartTransaction", false, true, Some("16"), false, (0, 0)),
+                ("RenameMultipartData", false, true, None, false, (0, 1)),
+            ] {
+                let error = has_error.then(|| Error {
+                    code: 1,
+                    error_info: "retired source".to_owned(),
+                });
+                let reply = match method {
+                    "PreparePartTransaction" => unary_frame(&PreparePartTransactionResponse { success, error }),
+                    "SettlePartTransaction" => unary_frame(&SettlePartTransactionResponse { success, error }),
+                    _ => unary_frame(&RenameDataResponse {
+                        success,
+                        error,
+                        ..Default::default()
+                    }),
+                };
+                // An unknown field must survive typed inspection byte-for-byte.
+                let mut framed = reply.to_vec();
+                framed.extend_from_slice(&[0xfa, 0x03, 0x03, b'x', b'y', b'z']);
+                let payload_len = u32::try_from(framed.len() - 5).expect("small reply");
+                framed[1..5].copy_from_slice(&payload_len.to_be_bytes());
+                let reply = Bytes::from(framed);
+                let backend = backend_with_reply(status, Some(reply.clone()))
+                    .await
+                    .expect("typed reply backend");
+                let proxy = RpcFaultProxy::start_on("127.0.0.1:0".parse().expect("loopback"), backend.address)
+                    .await
+                    .expect("proxy");
+                let pause = proxy.pause_late(&[method], 1, "late-upload").expect("part gate");
+                let request = match method {
+                    "PreparePartTransaction" => unary_frame(&PreparePartTransactionRequest {
+                        disk: "disk".to_owned(),
+                        dst_path: "late-upload/part.1".to_owned(),
+                        ..Default::default()
+                    }),
+                    "SettlePartTransaction" => unary_frame(&SettlePartTransactionRequest {
+                        disk: "disk".to_owned(),
+                        path: "late-upload/part.1".to_owned(),
+                        rollback,
+                        ..Default::default()
+                    }),
+                    _ => Bytes::from_static(b"late-upload"),
+                };
+                let address = proxy.local_addr();
+                let caller = tokio::spawn(async move {
+                    let client: ProxyClient = Client::builder(TokioExecutor::new()).http2_only(true).build_http();
+                    let response = call_body(&client, address, method, request).await.expect("retained reply");
+                    response.into_body().collect().await.expect("typed response frames")
+                });
+                pause.wait_for_hits(1).await.expect("actual retained request");
+                pause.release();
+                let response = caller.await.expect("caller task");
+                let trailers = response.trailers().expect("reply trailers");
+                assert_eq!(trailers["x-proof"], "unchanged-proof");
+                assert_eq!(trailers["grpc-status"], status.unwrap_or("0"));
+                assert_eq!(response.to_bytes(), reply);
+                assert_eq!(pause.responses(), 1);
+                assert_eq!(
+                    (pause.successful_mutation_replies(), pause.refused_mutations()),
+                    expected,
+                    "typed {method} outcome"
+                );
+                let actions = if method == "SettlePartTransaction" {
+                    if rollback { (0, 1) } else { (1, 0) }
+                } else {
+                    (0, 0)
+                };
+                assert_eq!(pause.retained_settle_actions(), actions);
+            }
+        })
+        .await
+        .expect("part reply deadline");
     }
     async fn assert_late_packet_ownership() {
         tokio::time::timeout(Duration::from_secs(10), async {

crates/ecstore/AGENTS.md

@@ -12,6 +12,9 @@ Applies to `crates/ecstore/`.
 
 - Be careful with allocations and locking in hot paths.
 - Keep network and disk operations async-friendly; avoid introducing unnecessary blocking.
+- Coalesce related `open`, `read`/`write`, `fsync`, and `close` filesystem steps into a single `blocking_queue` or `run_blocking_namespace_operation` closure per disk rather than chaining multiple `tokio::fs` or `spawn_blocking` dispatches.
+- Pin a parent directory once (`openat2(BENEATH | NO_SYMLINKS)` on Linux) and reuse the pinned guard across sibling metadata reads (`read_optional_in_pinned`, `part.N.meta` loops) instead of re-walking ancestor components per file or re-opening when `source_parent == destination_parent`.
+- After `file.metadata()`, pre-size read buffers (`Vec::with_capacity(len + 1)`) before `read_to_end` to avoid `std::io::read_to_end`'s 32-byte exponential growth (`6–7` `read()` syscalls per small metadata file).
 - Benchmark-sensitive changes should include measurable rationale.
 
 ## Allocation Discipline in Hot Paths

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

@@ -6,6 +6,7 @@ use super::{
     LocalDisk,
     multipart_source_state::{
         LoadedSource, SourceHead, SourceState, SourceStore, decode, encode, load_source, raw_sha, read_optional,
+        read_optional_in_pinned,
     },
 };
 use crate::disk::{
@@ -127,20 +128,6 @@ pub(super) fn validate_retired_head(
 }
 
 impl SourceStore {
-    pub async fn root_missing(&self) -> Result<bool> {
-        let source = self.source.clone();
-        let base = self.base.clone();
-        let root = self.publication_root.clone();
-        os::run_blocking_namespace_operation(self.lease.clone(), move || {
-            match os::pin_multipart_source_directory(&source, &base, &root) {
-                Ok(_) => Ok(false),
-                Err(error) if error.kind() == ErrorKind::NotFound => Ok(true),
-                Err(error) => Err(error),
-            }
-        })
-        .await
-        .map_err(DiskError::from)
-    }
     fn verify_abort_cohort(&self, certificate: &MultipartAbortCertificate) -> Result<()> {
         certificate.verify_with_authority(&self.authority)?;
         if certificate.disk_ids != self.disk_ids
@@ -265,7 +252,7 @@ impl SourceStore {
         let certificate = self.read_abort_certificate(&loaded).await?;
         Ok(MultipartSourceInspection {
             disk_id: self.disk_id,
-            root_missing: self.root_missing().await?,
+            root_missing: loaded.root_missing,
             abort_round: loaded
                 .state
                 .as_ref()
@@ -620,12 +607,17 @@ pub(super) struct OrdinaryMultipartDeleteCheck {
 }
 impl OrdinaryMultipartDeleteCheck {
     pub(super) fn validate(&self) -> Result<()> {
-        if let Some(raw) = read_optional(&self.source.join(STATE), &self.base, &self.root)? {
+        let pinned = match os::pin_multipart_source_directory(&self.source, &self.base, &self.root) {
+            Ok(pinned) => pinned,
+            Err(error) if error.kind() == ErrorKind::NotFound => return Ok(()),
+            Err(error) => return Err(error.into()),
+        };
+        if let Some(raw) = read_optional_in_pinned(&pinned, &self.source, STATE, &self.base, &self.root)? {
             let state: SourceState = decode(&raw)?;
             if state.abort.is_some() || state.phase == MultipartSourcePhase::Aborting {
                 return Err(DiskError::ConditionalDeleteConflict);
             }
-        } else if os::multipart_source_has_native_controls(&self.source, &self.base, &self.root)? {
+        } else if os::multipart_source_has_native_controls_in_pinned(&self.source, &pinned)? {
             return Err(DiskError::ConditionalDeleteConflict);
         }
         Ok(())

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

@@ -170,7 +170,7 @@ impl LocalDisk {
             recovery: loaded.state.as_ref().and_then(|state| state.recovery.clone()),
             negative_certificate,
             ready_certificate,
-            root_missing: store.root_missing().await?,
+            root_missing: loaded.root_missing,
             abort_round: loaded
                 .state
                 .as_ref()

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

@@ -106,6 +106,7 @@ impl SourceHead {
 }
 
 pub(super) struct LoadedSource {
+    pub root_missing: bool,
     pub state: Option<SourceState>,
     pub head: Option<SourceHead>,
     pub original: Option<Bytes>,
@@ -224,11 +225,31 @@ pub(super) fn read_optional(path: &Path, base: &Path, root: &os::PublicationRoot
         Err(error) => Err(error.into()),
     }
 }
+pub(super) fn read_optional_in_pinned(
+    pinned: &os::ExistingBaseDirectoryGuard,
+    parent: &Path,
+    name: &str,
+    base: &Path,
+    root: &os::PublicationRoot,
+) -> Result<Option<Bytes>> {
+    match os::read_multipart_source_metadata_in_pinned(pinned, parent, std::ffi::OsStr::new(name), base, root) {
+        Ok(bytes) => Ok(Some(Bytes::from(bytes))),
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => Ok(None),
+        Err(error) => Err(error.into()),
+    }
+}
 /// Canonical source mutations are admitted under SourceRW at the syscall,
 /// not on the earlier async caller that may have timed out.
 pub(super) fn validate_source_mutation(source: &Path, base: &Path, root: &os::PublicationRoot, initial: bool) -> Result<()> {
-    let state = read_optional(&source.join(STATE), base, root)?;
-    let head = read_optional(&source.join(STORAGE_FORMAT_FILE), base, root)?;
+    let pinned = match os::pin_multipart_source_directory(source, base, root) {
+        Ok(pinned) => pinned,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return if initial { Ok(()) } else { Err(DiskError::FileNotFound) };
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let state = read_optional_in_pinned(&pinned, source, STATE, base, root)?;
+    let head = read_optional_in_pinned(&pinned, source, STORAGE_FORMAT_FILE, base, root)?;
     let Some(head) = head else {
         return if initial && state.is_none() {
             Ok(())
@@ -348,6 +369,9 @@ impl SourceStore {
         .map_err(DiskError::from)?
     }
     pub async fn persist(&self, state: &SourceState) -> Result<()> {
+        self.persist_with_source_sync(state, false).await
+    }
+    async fn persist_with_source_sync(&self, state: &SourceState, sync_source_before: bool) -> Result<()> {
         state.validate(self.disk_id, &self.path)?;
         let raw = encode(state)?;
         let source = self.source.clone();
@@ -365,6 +389,9 @@ impl SourceStore {
         });
         os::run_blocking_namespace_operation(self.lease.clone(), move || {
             let _pin = os::pin_multipart_source_directory(&source, &base, &root)?;
+            if sync_source_before {
+                os::fsync_dir_std(&source)?;
+            }
             if let Some(path) = terminal {
                 os::replace_multipart_source_metadata(&path, &raw, &base, &root)?;
             }
@@ -373,6 +400,7 @@ impl SourceStore {
         .await
         .map_err(DiskError::from)
     }
+    #[cfg(test)]
     pub async fn original(&self, state: &SourceState) -> Result<Bytes> {
         let path = original_path(&self.source, state.backup_owner);
         let base = self.base.clone();
@@ -490,7 +518,7 @@ impl SourceStore {
         if encode(actual)? != encode(state)? {
             return Err(DiskError::ConditionalDeleteConflict);
         }
-        let raw = self.original(state).await?;
+        let raw = loaded.original.as_ref().ok_or(DiskError::FileCorrupt)?;
         let allowed = match loaded.head.as_ref() {
             None => {
                 state.phase == MultipartSourcePhase::Recovering
@@ -551,7 +579,7 @@ impl SourceStore {
         if !allowed {
             return Err(DiskError::ConditionalDeleteConflict);
         }
-        let mut info = file_info(&raw, RUSTFS_META_MULTIPART_BUCKET, &self.path, state.lineage.original_version_id)?;
+        let mut info = file_info(raw, RUSTFS_META_MULTIPART_BUCKET, &self.path, state.lineage.original_version_id)?;
         strip_native_controls(&mut info.metadata);
         mark_native_source(&mut info);
         set_source_status(
@@ -772,13 +800,32 @@ pub(super) fn load_source(
     physical_slot: usize,
     disk_ids: &[Uuid],
 ) -> Result<LoadedSource> {
-    let state_raw = read_optional(&source.join(STATE), base, root)?;
+    let pinned = match os::pin_multipart_source_directory(source, base, root) {
+        Ok(pinned) => pinned,
+        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
+            return Ok(LoadedSource {
+                root_missing: true,
+                state: None,
+                head: None,
+                original: None,
+                info: None,
+                snapshot: None,
+                head_matches_state: false,
+                head_matches_activation: false,
+            });
+        }
+        Err(error) => return Err(error.into()),
+    };
+    let state_raw = read_optional_in_pinned(&pinned, source, STATE, base, root)?;
     let state: Option<SourceState> = state_raw.as_deref().map(decode).transpose()?;
-    if state.is_none() && os::multipart_source_has_native_controls(source, base, root)? {
+    if state.is_none() && os::multipart_source_has_native_controls_in_pinned(source, &pinned)? {
         return Err(DiskError::ConditionalDeleteConflict);
     }
-    let head_path = source.join(STORAGE_FORMAT_FILE);
-    let head = if let Some(state) = state.as_ref() {
+    let head = if let Some(state) = state
+        .as_ref()
+        .filter(|state| state.prepared_length.is_some_and(|len| len > 64 * 1024))
+    {
+        let head_path = source.join(STORAGE_FORMAT_FILE);
         let maximum = state
             .prepared_length
             .unwrap_or(0)
@@ -789,7 +836,8 @@ pub(super) fn load_source(
                 let raw = if is_proxy {
                     None
                 } else {
-                    let raw = read_optional(&head_path, base, root)?.ok_or(DiskError::FileCorrupt)?;
+                    let raw = read_optional_in_pinned(&pinned, source, STORAGE_FORMAT_FILE, base, root)?
+                        .ok_or(DiskError::FileCorrupt)?;
                     if raw_sha(&raw) != sha256 {
                         return Err(DiskError::FileCorrupt);
                     }
@@ -801,10 +849,17 @@ pub(super) fn load_source(
             Err(error) => return Err(error.into()),
         }
     } else {
-        read_optional(&head_path, base, root)?.map(|raw| SourceHead {
-            sha256: raw_sha(&raw),
-            length: raw.len() as u64,
-            raw: Some(raw),
+        read_optional_in_pinned(&pinned, source, STORAGE_FORMAT_FILE, base, root)?.map(|raw| {
+            let sha256 = raw_sha(&raw);
+            let length = raw.len() as u64;
+            let is_proxy = state
+                .as_ref()
+                .is_some_and(|state| Some(sha256) == state.prepared_sha256 && Some(length) == state.prepared_length);
+            SourceHead {
+                sha256,
+                length,
+                raw: (!is_proxy).then_some(raw),
+            }
         })
     };
     let mut original = None;
@@ -950,6 +1005,7 @@ pub(super) fn load_source(
         head_matches_state = true;
     }
     Ok(LoadedSource {
+        root_missing: false,
         state,
         head,
         original,
@@ -1587,10 +1643,8 @@ impl NativeMultipartPublication {
             }
         }
         self.store.sync_target_with_lease(&self.publication, target_lease).await?;
-        let source = self.store.source.clone();
-        os::run_blocking_namespace_operation(self.store.lease.clone(), move || os::fsync_dir_std(&source)).await?;
         state.advance(MultipartSourcePhase::ConsumedUnsealed)?;
-        self.store.persist(&state).await?;
+        self.store.persist_with_source_sync(&state, true).await?;
         *self.state.lock() = state;
         let loaded = self.store.load().await?;
         let snapshot = loaded.snapshot.ok_or(DiskError::FileCorrupt)?;

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

@@ -411,15 +411,23 @@ impl LocalDisk {
             }
             let _attempt = os::mkdir_all_below_existing_base_std(&private_object, &base, &publication_root)?;
             if incoming.data.is_none() && !incoming.is_remote() {
+                let source_data_dir = source_object.join(source_dir.to_string());
+                let pinned_source_data = (!incoming.parts.is_empty())
+                    .then(|| os::pin_multipart_source_directory(&source_data_dir, &base, &publication_root))
+                    .transpose()?;
                 let mut names = Vec::with_capacity(incoming.parts.len().saturating_mul(3));
                 for part in &incoming.parts {
                     if cancelled() {
                         return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "multipart staging cancelled"));
                     }
-                    let path = source_object
-                        .join(source_dir.to_string())
-                        .join(format!("part.{}.meta", part.number));
-                    let bytes = os::read_multipart_source_metadata(&path, &base, &publication_root)?;
+                    let meta_name = format!("part.{}.meta", part.number);
+                    let bytes = os::read_multipart_source_metadata_in_pinned(
+                        pinned_source_data.as_ref().expect("pinned source data directory"),
+                        &source_data_dir,
+                        std::ffi::OsStr::new(&meta_name),
+                        &base,
+                        &publication_root,
+                    )?;
                     let actual = rustfs_filemeta::ObjectPartInfo::unmarshal(&bytes).map_err(std::io::Error::other)?;
                     let normalized = i64::try_from(actual.size)
                         .map_err(std::io::Error::other)?
@@ -441,13 +449,13 @@ impl LocalDisk {
                     // Preserve the native completed-MPU directory format. This
                     // immutable, validated generation record is distinct from
                     // the upload's mutable top-level xl.meta.
-                    names.push(format!("part.{}.meta", part.number));
+                    names.push(meta_name);
                     if let Some(integrity) = &part.integrity {
                         names.push(integrity.file_name());
                     }
                 }
                 os::hardlink_multipart_source_files(
-                    &source_object.join(source_dir.to_string()),
+                    &source_data_dir,
                     &private_object.join(final_dir.to_string()),
                     &names,
                     &base,

crates/ecstore/src/disk/os.rs

@@ -3372,6 +3372,29 @@ fn open_existing_directory_below_base_std(
     use rustix::fs::{Mode, OFlags, openat};
     let relative = existing_base_directory_suffix(directory, base)?;
     #[cfg(target_os = "linux")]
+    {
+        use rustix::fs::{ResolveFlags, openat2};
+        use std::os::fd::AsFd;
+        let flags = OFlags::RDONLY | OFlags::DIRECTORY | OFlags::NOFOLLOW | OFlags::CLOEXEC;
+        let owned_base;
+        let base_fd = if base == publication_root.io_path.as_path() || base == publication_root.path.as_path() {
+            publication_root.directory.as_fd()
+        } else {
+            owned_base = std::os::fd::OwnedFd::from(publication_root.open_directory(base)?);
+            owned_base.as_fd()
+        };
+        let target = if relative.as_os_str().is_empty() {
+            Path::new(".")
+        } else {
+            relative
+        };
+        match openat2(base_fd, target, flags, Mode::empty(), ResolveFlags::BENEATH | ResolveFlags::NO_SYMLINKS) {
+            Ok(leaf) => return Ok(vec![leaf]),
+            Err(rustix::io::Errno::NOSYS) => {}
+            Err(err) => return Err(err.into()),
+        }
+    }
+    #[cfg(target_os = "linux")]
     let base = publication_root.open_directory(base)?.into();
     #[cfg(not(target_os = "linux"))]
     let base = {
@@ -4925,12 +4948,16 @@ impl LinuxCommitDirectories {
         use std::os::unix::fs::MetadataExt;
         // A pinned directory may have been unlinked or replaced. Do not report a
         // successful publication into a detached object tree.
-        for (path, directory) in [
-            (&self.source_path, &self.source),
-            (&self.destination_path, self.prepare_destination()?),
-        ] {
-            let current = self.root.open_directory(path)?.metadata()?;
-            if (current.dev(), current.ino()) != (directory.device, directory.inode) {
+        let current = self.root.open_directory(&self.source_path)?.metadata()?;
+        if (current.dev(), current.ino()) != (self.source.device, self.source.inode) {
+            return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed"));
+        }
+        let destination = self.prepare_destination()?;
+        if self.source_path != self.destination_path
+            || (self.source.device, self.source.inode) != (destination.device, destination.inode)
+        {
+            let current = self.root.open_directory(&self.destination_path)?.metadata()?;
+            if (current.dev(), current.ino()) != (destination.device, destination.inode) {
                 return Err(io::Error::new(io::ErrorKind::InvalidData, "commit directory identity changed"));
             }
         }
@@ -5223,12 +5250,20 @@ pub(crate) fn prepare_rename_commit_guard(
         existing_base_directory_suffix(destination_parent, destination_base)?;
         let source = LinuxCommitDirectory::new(publication_root.open_directory(source_parent)?)?;
         let destination = std::sync::OnceLock::new();
-        match publication_root.open_directory(destination_parent) {
-            Ok(directory) => {
-                let _ = destination.set(LinuxCommitDirectory::new(directory)?);
+        if source_parent == destination_parent {
+            let _ = destination.set(LinuxCommitDirectory {
+                file: Arc::clone(&source.file),
+                device: source.device,
+                inode: source.inode,
+            });
+        } else {
+            match publication_root.open_directory(destination_parent) {
+                Ok(directory) => {
+                    let _ = destination.set(LinuxCommitDirectory::new(directory)?);
+                }
+                Err(err) if err.kind() == io::ErrorKind::NotFound => {}
+                Err(err) => return Err(err),
             }
-            Err(err) if err.kind() == io::ErrorKind::NotFound => {}
-            Err(err) => return Err(err),
         }
         Ok(RenameCommitGuard {
             directories: Arc::new(LinuxCommitDirectories {
@@ -6088,12 +6123,20 @@ fn existing_base_directory_suffix<'a>(dir_path: &'a Path, base_dir: &Path) -> io
 
 /// Missing state never downgrades surviving native control records to legacy.
 /// Inspect only this pinned upload root; parts and proof trees are not traversed.
+#[cfg(test)]
 pub(crate) fn multipart_source_has_native_controls(source: &Path, base: &Path, root: &PublicationRoot) -> io::Result<bool> {
-    let _pin = match pin_multipart_source_directory(source, base, root) {
+    let pin = match pin_multipart_source_directory(source, base, root) {
         Ok(pin) => pin,
         Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(false),
         Err(error) => return Err(error),
     };
+    multipart_source_has_native_controls_in_pinned(source, &pin)
+}
+
+pub(crate) fn multipart_source_has_native_controls_in_pinned(
+    source: &Path,
+    _pin: &ExistingBaseDirectoryGuard,
+) -> io::Result<bool> {
     let mut entries = std::fs::read_dir(source)?;
     for _ in 0..READ_DIR_PROBE_RAW_LIMIT {
         let Some(entry) = entries.next() else {
@@ -6397,15 +6440,10 @@ pub(crate) fn sync_multipart_owned_record_parents(
         if !directory.starts_with(source) {
             return Err(io::Error::new(io::ErrorKind::InvalidInput, "native record barrier escaped source"));
         }
-        let pinned = pin_multipart_source_directory(directory, base, publication_root)?;
+        let mut pinned = pin_multipart_source_directory(directory, base, publication_root)?;
         #[cfg(unix)]
         {
-            let opened = std::fs::File::from(
-                pinned
-                    .last()
-                    .ok_or_else(|| io::Error::other("empty native directory guard"))?
-                    .try_clone()?,
-            );
+            let opened = std::fs::File::from(pinned.pop().ok_or_else(|| io::Error::other("empty native directory guard"))?);
             fsync_directory_handle_std(directory, Some(&opened))?;
         }
         #[cfg(not(unix))]
@@ -6517,6 +6555,16 @@ pub(crate) fn read_multipart_source_metadata(
     read_multipart_metadata(path, base, publication_root, Some(16 * 1024 * 1024))
 }
 
+pub(crate) fn read_multipart_source_metadata_in_pinned(
+    pinned: &ExistingBaseDirectoryGuard,
+    parent: &Path,
+    name: &std::ffi::OsStr,
+    base: &Path,
+    publication_root: &PublicationRoot,
+) -> io::Result<Vec<u8>> {
+    read_multipart_metadata_in_pinned(pinned, parent, name, base, publication_root, Some(16 * 1024 * 1024))
+}
+
 /// Target FI histories have no source-control size limit. Keep the same
 /// pinned-parent and no-follow checks when inspecting every referenced DD.
 pub(crate) fn read_multipart_reclaim_target_metadata(
@@ -6558,8 +6606,10 @@ fn read_multipart_metadata_bytes(
     Ok(bytes)
 }
 
-fn read_multipart_metadata(
-    path: &Path,
+fn read_multipart_metadata_in_pinned(
+    pinned: &ExistingBaseDirectoryGuard,
+    parent: &Path,
+    name: &std::ffi::OsStr,
     base: &Path,
     publication_root: &PublicationRoot,
     maximum: Option<u64>,
@@ -6567,16 +6617,10 @@ fn read_multipart_metadata(
     #[cfg(unix)]
     {
         use rustix::fs::{Mode, OFlags, openat};
-        let parent = path
-            .parent()
-            .ok_or_else(|| io::Error::other("multipart metadata has no parent"))?;
-        let guards = open_existing_directory_below_base_std(parent, base, publication_root)?;
-        let directory = guards
+        let _ = (parent, base, publication_root);
+        let directory = pinned
             .last()
             .ok_or_else(|| io::Error::other("missing multipart source directory"))?;
-        let name = path
-            .file_name()
-            .ok_or_else(|| io::Error::other("multipart metadata has no name"))?;
         let mut file = std::fs::File::from(openat(
             directory,
             name,
@@ -6589,6 +6633,30 @@ fn read_multipart_metadata(
         }
         read_multipart_metadata_bytes(&mut file, Some(metadata.len()), maximum)
     }
+    #[cfg(not(unix))]
+    {
+        let _ = pinned;
+        read_multipart_metadata(&parent.join(name), base, publication_root, maximum)
+    }
+}
+
+fn read_multipart_metadata(
+    path: &Path,
+    base: &Path,
+    publication_root: &PublicationRoot,
+    maximum: Option<u64>,
+) -> io::Result<Vec<u8>> {
+    #[cfg(unix)]
+    {
+        let parent = path
+            .parent()
+            .ok_or_else(|| io::Error::other("multipart metadata has no parent"))?;
+        let name = path
+            .file_name()
+            .ok_or_else(|| io::Error::other("multipart metadata has no name"))?;
+        let guards = open_existing_directory_below_base_std(parent, base, publication_root)?;
+        read_multipart_metadata_in_pinned(&guards, parent, name, base, publication_root, maximum)
+    }
     #[cfg(windows)]
     {
         use windows_sys::{

crates/ecstore/src/set_disk/mod.rs

@@ -129,7 +129,7 @@ use rustfs_filemeta::{
 };
 use rustfs_heal_contracts::heal_channel::{
     DriveState, HealAdmissionResult, HealChannelPriority, HealItemType, HealOpts, HealRequestSource, HealScanMode,
-    send_heal_replacement_disk, send_heal_request_with_admission,
+    send_heal_request_with_admission,
 };
 use rustfs_io_metrics::{
     record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration,

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

@@ -21,19 +21,22 @@
 
 use super::super::{
     Arc, DiskError, DiskInfo, DiskInfoOptions, DiskOption, DiskStore, Endpoint, Error, FormatV3, HealChannelPriority, LockResult,
-    NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks, StorageError, debug, disk, info, load_format_erasure,
-    send_heal_replacement_disk, warn,
+    NamespaceLock, NamespaceLockWrapper, ObjectKey, Result, SetDisks, StorageError, debug, disk, load_format_erasure, warn,
 };
 use crate::disk::health_state::DriveMembershipSnapshot;
 use crate::disk::{DiskAPI, new_disk};
 use crate::layout::disks_layout::MAX_ERASURE_SET_DRIVE_COUNT;
 use crate::runtime::sources as runtime_sources;
-use futures::{StreamExt, stream};
+use futures::{StreamExt, stream, stream::FuturesUnordered};
 use rand::prelude::SliceRandom;
+use rustfs_heal_contracts::heal_channel::{create_auto_replacement_disks_request, send_heal_request};
 use rustfs_utils::path::object_name_disk_identity;
 #[cfg(test)]
 use uuid::Uuid;
 
+// Bound batching when another reconnect is still pending.
+const REPLACEMENT_BATCH_WAIT: std::time::Duration = std::time::Duration::from_secs(1);
+
 #[async_trait::async_trait]
 impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
     type Error = Error;
@@ -329,33 +332,82 @@ impl SetDisks {
 
     async fn connect_disks_with_active_monitoring(&self, active_monitoring: bool) {
         let disks = self.get_disks_internal().await;
-        // A slow failed endpoint must not hold back other returning drives.
-        // Keep the fanout within one set and release the inventory lock before I/O.
-        stream::iter(disks.into_iter().enumerate())
-            .for_each_concurrent(MAX_ERASURE_SET_DRIVE_COUNT, |(i, opdisk)| async move {
+        // Reconnect drives concurrently without holding the inventory lock.
+        let mut reconnects = stream::iter(disks.into_iter().enumerate())
+            .map(|(i, opdisk)| async move {
                 if let Some(disk) = opdisk {
                     if !active_monitoring
                         && disk.get_disk_location().set_idx.is_some()
                         && let disk::Disk::Local(local) = disk.as_ref()
                         && local.can_skip_background_probe().await
                     {
-                        return;
+                        return None;
                     }
                     if disk.is_online().await && disk.get_disk_location().set_idx.is_some() {
-                        return;
+                        return None;
                     }
 
                     let _ = disk.close().await;
                 }
 
-                if let Some(endpoint) = self.set_endpoints.get(i) {
-                    self.renew_disk(endpoint).await;
-                }
+                let endpoint = self.set_endpoints.get(i)?;
+                self.reconnect_disk(endpoint).await.then(|| endpoint.to_string())
             })
-            .await;
+            .buffer_unordered(MAX_ERASURE_SET_DRIVE_COUNT);
+        let mut submissions = FuturesUnordered::new();
+        let mut batch = Vec::new();
+        let mut deadline = None;
+        let mut reconnects_done = false;
+        loop {
+            if reconnects_done && submissions.is_empty() {
+                break;
+            }
+            tokio::select! {
+                result = reconnects.next(), if !reconnects_done => match result {
+                    Some(Some(endpoint)) => {
+                        batch.push(endpoint);
+                        deadline.get_or_insert_with(|| tokio::time::Instant::now() + REPLACEMENT_BATCH_WAIT);
+                    }
+                    Some(None) => {}
+                    None => {
+                        reconnects_done = true;
+                        if !batch.is_empty() {
+                            submissions.push(self.submit_replacement_heal(std::mem::take(&mut batch)));
+                        }
+                        deadline = None;
+                    }
+                },
+                _ = tokio::time::sleep_until(deadline.unwrap_or_else(tokio::time::Instant::now)), if deadline.is_some() => {
+                    submissions.push(self.submit_replacement_heal(std::mem::take(&mut batch)));
+                    deadline = None;
+                }
+                _ = submissions.next(), if !submissions.is_empty() => {}
+            }
+        }
+    }
+
+    async fn submit_replacement_heal(&self, replacements: Vec<String>) {
+        if replacements.is_empty() {
+            return;
+        }
+        // Queue admission deduplicates by set, so publish ready targets together.
+        let request = create_auto_replacement_disks_request(
+            self.pool_index,
+            self.set_index,
+            replacements,
+            Some(HealChannelPriority::Normal),
+        );
+        let _ = send_heal_request(request).await;
     }
 
     pub async fn renew_disk(&self, ep: &Endpoint) {
+        if self.reconnect_disk(ep).await {
+            self.submit_replacement_heal(vec![ep.to_string()]).await;
+        }
+    }
+
+    /// Return whether this reconnect attached a verified unformatted target.
+    async fn reconnect_disk(&self, ep: &Endpoint) -> bool {
         debug!("renew_disk: start {:?}", ep);
 
         let previous_disk = {
@@ -375,28 +427,17 @@ impl SetDisks {
             Err(e) => {
                 warn!("renew_disk: connect_endpoint err {:?}", &e);
                 if !matches!(e, DiskError::UnformattedDisk | DiskError::Io(_)) {
-                    return;
+                    return false;
                 }
 
                 let attached = match self.attach_unformatted_replacement_disk(ep).await {
                     Ok(attached) => attached,
                     Err(err) => {
                         warn!(endpoint = %ep, error = ?err, "renew_disk: unformatted replacement probe failed");
-                        return;
+                        return false;
                     }
                 };
-                if !attached {
-                    return;
-                }
-
-                info!("renew_disk attached unformatted replacement and will trigger heal_disk, {:?}", ep);
-                let (Ok(pool_index), Ok(set_index)) = (usize::try_from(ep.pool_idx), usize::try_from(ep.set_idx)) else {
-                    warn!("renew_disk: replacement target has invalid pool or set index, {:?}", ep);
-                    return;
-                };
-                let _ =
-                    send_heal_replacement_disk(pool_index, set_index, ep.to_string(), Some(HealChannelPriority::Normal)).await;
-                return;
+                return attached;
             }
         };
 
@@ -404,7 +445,7 @@ impl SetDisks {
             Ok(res) => res,
             Err(e) => {
                 warn!("renew_disk: find_disk_index err {:?}", e);
-                return;
+                return false;
             }
         };
 
@@ -422,7 +463,7 @@ impl SetDisks {
                 expected_set_index = self.set_index,
                 "renew_disk rejected a drive whose endpoint and format do not identify the same topology slot"
             );
-            return;
+            return false;
         }
 
         let _ = new_disk.set_disk_id(Some(fm.erasure.this)).await;
@@ -442,7 +483,7 @@ impl SetDisks {
                 new_disk.enable_health_check();
                 *slot = Some(new_disk);
             }
-            return;
+            return false;
         }
 
         new_disk.enable_health_check();
@@ -464,6 +505,7 @@ impl SetDisks {
 
         let mut disk_lock = self.disks.write().await;
         disk_lock[disk_idx] = Some(new_disk);
+        false
     }
 
     /// Attach a replacement target only after proving that the exact local slot
@@ -1102,6 +1144,162 @@ mod tests {
         drop(temp_dirs);
     }
 
+    #[tokio::test]
+    async fn reconnect_sweep_collects_all_blank_targets_and_excludes_original_disks() {
+        use rustfs_heal_contracts::heal_channel::{HealAdmissionResult, HealChannelCommand, init_heal_channel};
+
+        const CHILD: &str = "RUSTFS_TEST_GROUPED_RECONNECT_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            // The heal sender is process-global; keep this receiver isolated.
+            let output = std::process::Command::new(std::env::current_exe().expect("test executable"))
+                .args([
+                    "--exact",
+                    "set_disk::ops::locking::tests::reconnect_sweep_collects_all_blank_targets_and_excludes_original_disks",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .output()
+                .expect("isolated reconnect test should run");
+            let stdout = String::from_utf8_lossy(&output.stdout);
+            let stderr = String::from_utf8_lossy(&output.stderr);
+            assert!(output.status.success(), "isolated reconnect test failed: {stdout} {stderr}");
+            assert!(stdout.contains("1 passed"), "child must execute the reconnect test: {stdout}");
+            return;
+        }
+        let mut heal_rx = init_heal_channel().expect("isolated heal receiver");
+        let disk_count = 4;
+        let format = FormatV3::new(1, disk_count);
+        let mut directories = Vec::new();
+        let mut endpoints = Vec::new();
+        let mut disks = Vec::new();
+        for disk_idx in 0..2 {
+            let (directory, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
+            directories.push(directory);
+            endpoints.push(endpoint);
+            disks.push(Some(disk));
+        }
+        for disk_idx in 2..disk_count {
+            let directory = tempfile::tempdir().expect("replacement directory");
+            let mut endpoint =
+                Endpoint::try_from(directory.path().to_str().expect("UTF-8 replacement path")).expect("replacement endpoint");
+            endpoint.set_pool_index(0);
+            endpoint.set_set_index(0);
+            endpoint.set_disk_index(disk_idx);
+            directories.push(directory);
+            endpoints.push(endpoint);
+            disks.push(None);
+        }
+        let mut ready_targets = endpoints[2..].iter().map(ToString::to_string).collect::<Vec<_>>();
+        ready_targets.sort_unstable();
+        // Disk 1 returns with its format and data; disks 2 and 3 are blank.
+        disks[1]
+            .take()
+            .expect("original disk handle")
+            .close()
+            .await
+            .expect("close original disk");
+        let retained_data = directories[1].path().join("retained-data");
+        tokio::fs::write(&retained_data, b"original data")
+            .await
+            .expect("write original data");
+        let set = Arc::new(
+            SetDisks::new(
+                "reconnect-sweep".to_string(),
+                Arc::new(RwLock::new(disks)),
+                disk_count,
+                2,
+                0,
+                0,
+                endpoints.clone(),
+                format,
+                Vec::new(),
+            )
+            .await,
+        );
+
+        // Block disk 0's format check, then make it a late blank replacement.
+        let slow = set.get_disks_internal().await[0].clone().expect("original disk");
+        assert!(slow.is_online().await, "initialize original disk identity");
+        let disk::Disk::Local(wrapper) = slow.as_ref() else {
+            panic!("fixture must use a local disk");
+        };
+        let slow_local = wrapper.get_disk();
+        let mut slow_format_guard = slow_local.format_info.write().await;
+        slow_format_guard.last_check = None;
+        std::fs::remove_file(
+            directories[0]
+                .path()
+                .join(disk::RUSTFS_META_BUCKET)
+                .join(disk::FORMAT_CONFIG_FILE),
+        )
+        .expect("make the blocked disk a blank replacement");
+        let sweep = tokio::spawn({
+            let set = set.clone();
+            async move { set.connect_disks_with_active_monitoring(true).await }
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            loop {
+                let attached = set.get_disks_internal().await;
+                if attached[1..].iter().all(Option::is_some) {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("other drives must reconnect while the first drive is blocked");
+        let command = tokio::time::timeout(std::time::Duration::from_secs(10), heal_rx.recv())
+            .await
+            .expect("ready targets must be submitted while a drive is blocked")
+            .expect("heal receiver should remain open");
+        let HealChannelCommand::Start { request, response_tx } = command else {
+            panic!("expected a replacement heal request");
+        };
+        assert_eq!(request.heal_endpoints, ready_targets, "ready blank disks must share one request");
+        assert!(!sweep.is_finished(), "the blocked drive must still be pending");
+        drop(slow_format_guard);
+        // Do not acknowledge the first request yet: a delayed admission
+        // response must not stop the late drive reconnecting and submitting.
+        let late_command = tokio::time::timeout(std::time::Duration::from_secs(10), heal_rx.recv())
+            .await
+            .expect("late target must be submitted while the first response is pending")
+            .expect("heal receiver should remain open");
+        let HealChannelCommand::Start {
+            request: late_request,
+            response_tx: late_response,
+        } = late_command
+        else {
+            panic!("expected a late replacement heal request");
+        };
+        assert_eq!(late_request.heal_endpoints, [endpoints[0].to_string()]);
+        response_tx
+            .send(Ok(HealAdmissionResult::Accepted))
+            .expect("acknowledge ready targets");
+        late_response
+            .send(Ok(HealAdmissionResult::Accepted))
+            .expect("acknowledge late target");
+        sweep.await.expect("reconnect sweep should finish");
+        assert!(heal_rx.try_recv().is_err(), "no duplicate request should be sent");
+        let attached = set.get_disks_internal().await;
+        for (index, disk) in attached.iter().enumerate() {
+            let disk = disk.as_ref().expect("each verified endpoint should be attached");
+            let format = load_format_erasure(disk, false).await;
+            if index == 1 {
+                assert!(format.is_ok(), "original disk must retain its format");
+            } else {
+                assert_eq!(format.expect_err("replacement must remain unformatted"), DiskError::UnformattedDisk);
+                assert!(!disk.health_check_enabled_for_test(), "format must precede replacement health checks");
+            }
+            disk.close().await.expect("close test disk");
+        }
+        assert_eq!(tokio::fs::read(&retained_data).await.expect("read original data"), b"original data");
+        let local_map = runtime_sources::local_disk_map_handle();
+        let mut local_map = local_map.write().await;
+        for endpoint in endpoints {
+            local_map.remove(&endpoint.to_string());
+        }
+    }
+
     #[tokio::test]
     async fn renew_disk_attaches_only_a_verified_local_unformatted_replacement() {
         let disk_count = 4;

crates/heal-contracts/src/heal_channel.rs

@@ -683,14 +683,17 @@ fn create_auto_heal_disk_request(set_disk_id: String, priority: Option<HealChann
     }
 }
 
-fn create_auto_replacement_disk_request(
+/// Build one automatic replacement request for the verified targets of a set.
+pub fn create_auto_replacement_disks_request(
     pool_index: usize,
     set_index: usize,
-    replacement_endpoint: String,
+    mut replacement_endpoints: Vec<String>,
     priority: Option<HealChannelPriority>,
 ) -> HealChannelRequest {
+    replacement_endpoints.sort_unstable();
+    replacement_endpoints.dedup();
     let mut request = create_auto_heal_disk_request(format!("pool_{pool_index}_set_{set_index}"), priority);
-    request.heal_endpoints = vec![replacement_endpoint];
+    request.heal_endpoints = replacement_endpoints;
     request.pool_index = Some(pool_index);
     request.set_index = Some(set_index);
     request
@@ -711,10 +714,10 @@ pub async fn send_heal_replacement_disk(
     replacement_endpoint: String,
     priority: Option<HealChannelPriority>,
 ) -> Result<(), String> {
-    send_heal_request(create_auto_replacement_disk_request(
+    send_heal_request(create_auto_replacement_disks_request(
         pool_index,
         set_index,
-        replacement_endpoint,
+        vec![replacement_endpoint],
         priority,
     ))
     .await
@@ -726,8 +729,12 @@ mod auto_heal_disk_request_tests {
 
     #[test]
     fn replacement_disk_request_carries_its_exact_endpoint() {
-        let request =
-            create_auto_replacement_disk_request(2, 3, "http://node2:9000/drive3".to_string(), Some(HealChannelPriority::Normal));
+        let request = create_auto_replacement_disks_request(
+            2,
+            3,
+            vec!["http://node2:9000/drive3".to_string()],
+            Some(HealChannelPriority::Normal),
+        );
 
         assert_eq!(request.disk.as_deref(), Some("pool_2_set_3"));
         assert_eq!(request.heal_endpoints, ["http://node2:9000/drive3"]);
@@ -736,6 +743,23 @@ mod auto_heal_disk_request_tests {
         assert_eq!(request.source, HealRequestSource::AutoHeal);
     }
 
+    #[test]
+    fn replacement_disk_request_keeps_every_distinct_target() {
+        let request = create_auto_replacement_disks_request(
+            2,
+            3,
+            vec!["drive-b".to_string(), "drive-a".to_string(), "drive-b".to_string()],
+            Some(HealChannelPriority::Normal),
+        );
+
+        assert_eq!(request.disk.as_deref(), Some("pool_2_set_3"));
+        assert_eq!(request.heal_endpoints, ["drive-a", "drive-b"]);
+        assert_eq!(request.pool_index, Some(2));
+        assert_eq!(request.set_index, Some(3));
+        assert_eq!(request.priority, HealChannelPriority::Normal);
+        assert_eq!(request.source, HealRequestSource::AutoHeal);
+    }
+
     #[test]
     fn legacy_auto_heal_disk_request_has_no_replacement_endpoint() {
         let request = create_auto_heal_disk_request("pool_2_set_3".to_string(), None);

crates/heal/src/heal/manager.rs

@@ -825,6 +825,21 @@ fn admin_heal_options_compatible(left: &HealOptions, right: &HealOptions) -> boo
         && left.no_lock == right.no_lock
 }
 
+fn replacement_request_compatible(
+    request: &HealRequest,
+    owner_type: &HealType,
+    owner_options: &HealOptions,
+    owner_source: HealRequestSource,
+) -> bool {
+    request.source == HealRequestSource::AutoHeal
+        && owner_source == HealRequestSource::AutoHeal
+        && matches!(request.heal_type, HealType::ErasureSet { .. })
+        && request.heal_type == *owner_type
+        && request.options.pool_index == owner_options.pool_index
+        && request.options.set_index == owner_options.set_index
+        && admin_heal_options_compatible(&request.options, owner_options)
+}
+
 fn admin_overlap_rejection(
     request: &HealRequest,
     owner_type: &HealType,
@@ -1276,7 +1291,9 @@ impl HealManager {
         request: HealRequest,
         config: &HealConfig,
         context: &'static str,
+        replacement_targets_frozen: bool,
     ) -> HealAdmissionDecision {
+        let replacement_targets_frozen = replacement_targets_frozen || request.retry_attempts != 0;
         let queue_len = queue.len();
         publish_heal_queue_length(queue);
         let queue_capacity = config.queue_size;
@@ -1284,6 +1301,25 @@ impl HealManager {
         let best_effort_reserve = best_effort_queue_reserve(queue_capacity);
         let best_effort_source_reserves = best_effort_source_reserves(queue_capacity);
 
+        // Scanner and retry admission bypass submit_heal_request's duplicate
+        // checks. Keep their target ownership checks ahead of queue capacity.
+        if !request.force_start
+            && request.source == HealRequestSource::AutoHeal
+            && !request.heal_endpoints.is_empty()
+            && let Some(owner_id) = queue
+                .queued_request_id_for_dedup_key(&PriorityHealQueue::make_dedup_key(&request))
+                .map(ToOwned::to_owned)
+        {
+            let mut admission = Self::duplicate_admission_for_request(&request, config);
+            if admission == HealAdmissionResult::Merged
+                && !queue.merge_replacement_targets(&owner_id, &request, true, replacement_targets_frozen)
+            {
+                admission = HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning);
+            }
+            Self::record_admission_metric(request.source, admission, context);
+            return HealAdmissionDecision::new(admission);
+        }
+
         if queue_len >= queue_capacity && !request.force_start {
             if Self::can_displace_queued_work(&request)
                 && queue.can_displace_lower_priority_where(request.priority, Self::queued_request_can_be_displaced)
@@ -1443,7 +1479,7 @@ impl HealManager {
         let priority = request.priority;
         let source = request.source;
 
-        match queue.push(request) {
+        match queue.push_with_replacement_scope(request, replacement_targets_frozen) {
             QueuePushOutcome::Accepted => {
                 publish_heal_queue_length(queue);
                 Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
@@ -2260,7 +2296,7 @@ impl HealManager {
             // HS-06: under the minio_error overlap policy an exact duplicate
             // admin start reports the typed AlreadyRunning rejection instead
             // of the silent merge (MinIO's ErrHealAlreadyRunning).
-            let admission = if (request.source == HealRequestSource::Admin
+            let mut admission = if (request.source == HealRequestSource::Admin
                 && config.overlap_policy == HealOverlapPolicy::MinioError)
                 || (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable()))
             {
@@ -2270,6 +2306,44 @@ impl HealManager {
             } else {
                 Self::duplicate_admission_for_request(&request, &config)
             };
+            if matches!(admission, HealAdmissionResult::Merged)
+                && request.source == HealRequestSource::AutoHeal
+                && !request.heal_endpoints.is_empty()
+            {
+                let (owner_recovered, incoming_recovered) = {
+                    // Match queue -> recovery-anchor ordering in the scanner.
+                    let anchors = self
+                        .replacement_recovery_anchors
+                        .lock()
+                        .unwrap_or_else(|poisoned| poisoned.into_inner());
+                    (anchors.contains_key(&merged_task_id), anchors.contains_key(&request.id))
+                };
+                let incoming_frozen = incoming_recovered || request.retry_attempts != 0;
+                let covered = if incoming_frozen && request.id != merged_task_id {
+                    false
+                } else if duplicate_state == "queued" {
+                    queue.merge_replacement_targets(&merged_task_id, &request, !owner_recovered, incoming_frozen)
+                } else {
+                    let owner = active_heals
+                        .get(&merged_task_id)
+                        .map(|task| (&task.heal_type, &task.options, task.source, &task.heal_endpoints))
+                        .or_else(|| {
+                            retrying_heals.get(&merged_task_id).map(|retry| {
+                                let owner = &retry.request;
+                                (&owner.heal_type, &owner.options, owner.source, &owner.heal_endpoints)
+                            })
+                        });
+                    owner.is_some_and(|(heal_type, options, source, endpoints)| {
+                        replacement_request_compatible(&request, heal_type, options, source)
+                            && request.heal_endpoints.iter().all(|endpoint| endpoints.contains(endpoint))
+                    })
+                };
+                if !covered {
+                    // A frozen generation cannot certify newly arriving disks.
+                    // Leave them discoverable by the scanner for the next pass.
+                    admission = HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning);
+                }
+            }
             if matches!(admission, HealAdmissionResult::Merged)
                 && let Some(target) = mrf_notice_target
             {
@@ -2426,7 +2500,14 @@ impl HealManager {
         let request_id = request.id.clone();
         let request_heal_type = request.heal_type.clone();
         let request_source = request.source;
-        let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit");
+        let recovered_replacement = request.source == HealRequestSource::AutoHeal
+            && !request.heal_endpoints.is_empty()
+            && self
+                .replacement_recovery_anchors
+                .lock()
+                .unwrap_or_else(|poisoned| poisoned.into_inner())
+                .contains_key(&request.id);
+        let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit", recovered_replacement);
         let admission = admission_decision.result;
         if durable_handoff_required && !admission.is_admitted() {
             self.root_recovery

crates/heal/src/heal/manager/auto_scan.rs

@@ -682,6 +682,7 @@ impl HealManager {
                                 HealPriority::Low,
                             );
                             let recovery_anchor = recovery.as_ref().map(|(_, _, _, anchor)| anchor.clone());
+                            let recovered_replacement = recovery.is_some();
                             if let Some((task_id, _, _, _)) = recovery {
                                 req.id = task_id;
                             }
@@ -691,7 +692,8 @@ impl HealManager {
                             let endpoint_count = req.heal_endpoints.len();
                             let config = config.read().await;
                             let mut queue = heal_queue.lock().await;
-                            let admission_decision = Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan");
+                            let admission_decision =
+                                Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan", recovered_replacement);
                             let admission = admission_decision.result;
                             let should_notify =
                                 matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;

crates/heal/src/heal/manager/queue.rs

@@ -70,6 +70,8 @@ pub(super) struct PriorityQueueItem {
     pub(super) sequence: u64,
     pub(super) dedup_key: String,
     pub(super) request: HealRequest,
+    /// Execution or durable recovery fixed these targets; grace waits can have zero retries.
+    pub(super) replacement_targets_frozen: bool,
 }
 
 impl Eq for PriorityQueueItem {}
@@ -316,6 +318,10 @@ impl PriorityHealQueue {
     }
 
     pub(super) fn push(&mut self, request: HealRequest) -> QueuePushOutcome {
+        self.push_with_replacement_scope(request, false)
+    }
+
+    pub(super) fn push_with_replacement_scope(&mut self, request: HealRequest, frozen: bool) -> QueuePushOutcome {
         let key = Self::make_dedup_key(&request);
 
         // Check for duplicates unless the caller explicitly forces admission.
@@ -340,6 +346,7 @@ impl PriorityHealQueue {
             sequence: self.sequence,
             dedup_key: key,
             request,
+            replacement_targets_frozen: frozen,
         });
         if is_best_effort_source(source) {
             self.best_effort_queued += 1;
@@ -680,6 +687,51 @@ impl PriorityHealQueue {
         self.heap.iter().map(|item| &item.request)
     }
 
+    pub(super) fn merge_replacement_targets(
+        &mut self,
+        owner_id: &str,
+        request: &HealRequest,
+        allow_expand: bool,
+        incoming_frozen: bool,
+    ) -> bool {
+        let Some(item) = self.heap.iter().find(|item| item.request.id == owner_id) else {
+            return false;
+        };
+        let owner = &item.request;
+        let same_id = request.id == owner.id;
+        let eligible = replacement_request_compatible(request, &owner.heal_type, &owner.options, owner.source)
+            && (!incoming_frozen || same_id);
+        let covered = eligible
+            && request
+                .heal_endpoints
+                .iter()
+                .all(|endpoint| owner.heal_endpoints.contains(endpoint));
+        let mark_frozen = incoming_frozen && same_id && !item.replacement_targets_frozen;
+        let can_expand = eligible
+            && allow_expand
+            && !item.replacement_targets_frozen
+            && !incoming_frozen
+            && owner.retry_attempts == 0
+            && request.retry_attempts == 0
+            && !owner.heal_endpoints.is_empty();
+        if !mark_frozen && (covered || !can_expand) {
+            return covered;
+        }
+
+        // Neither targets nor their frozen flag affect heap ordering.
+        let mut items = std::mem::take(&mut self.heap).into_vec();
+        if let Some(item) = items.iter_mut().find(|item| item.request.id == owner_id) {
+            item.replacement_targets_frozen |= mark_frozen;
+            if !covered && can_expand {
+                item.request.heal_endpoints.extend(request.heal_endpoints.iter().cloned());
+                item.request.heal_endpoints.sort_unstable();
+                item.request.heal_endpoints.dedup();
+            }
+        }
+        self.heap = BinaryHeap::from(items);
+        covered || can_expand
+    }
+
     #[cfg(test)]
     pub(super) fn contains_request_id(&self, request_id: &str) -> bool {
         self.heap.iter().any(|item| item.request.id == request_id)

crates/heal/src/heal/manager/scheduler.rs

@@ -582,7 +582,7 @@ impl HealManager {
                                     return;
                                 }
                                 let admission_decision =
-                                    Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry");
+                                    Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry", true);
                                 let admission = admission_decision.result;
                                 let should_notify = matches!(admission, HealAdmissionResult::Accepted)
                                     && retry_config.event_driven_scheduler_enable;

crates/heal/src/heal/manager/tests.rs

@@ -2505,6 +2505,301 @@ fn test_heal_request_and_task_metric_labels_match() {
     assert_eq!(task.metric_set_label(), "pool_0_set_1");
 }
 
+#[tokio::test]
+async fn automatic_replacement_queued_targets_merge_without_changing_fifo_or_token() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    let owner = automatic_replacement_request(&["replacement-a"]);
+    let owner_id = owner.id.clone();
+    let created_at = owner.created_at;
+    let first = manager
+        .submit_heal_request_with_receipt(owner)
+        .await
+        .expect("queue first target");
+    assert_eq!(first.result, HealAdmissionResult::Accepted);
+
+    let other = HealRequest::object("bucket".to_string(), "unrelated".to_string(), None);
+    let other_id = other.id.clone();
+    manager.submit_heal_request(other).await.expect("queue unrelated work");
+    let merged = manager
+        .submit_heal_request_with_receipt(automatic_replacement_request(&["replacement-b", "replacement-a"]))
+        .await
+        .expect("merge all fresh targets");
+    assert_eq!(merged.result, HealAdmissionResult::Merged);
+    assert_eq!(merged.task_id, owner_id);
+
+    let mut queue = manager.heal_queue.lock().await;
+    assert_eq!(queue.len(), 2);
+    let owner = queue.pop_next().expect("original task remains first");
+    assert_eq!(owner.id, owner_id);
+    assert_eq!(owner.created_at, created_at);
+    assert_eq!(owner.heal_endpoints, ["replacement-a", "replacement-b"]);
+    assert_eq!(queue.pop_next().expect("unrelated task remains queued").id, other_id);
+}
+
+fn automatic_replacement_request(targets: &[&str]) -> HealRequest {
+    let mut request = HealRequest::new(
+        HealType::ErasureSet {
+            buckets: Vec::new(),
+            set_disk_id: "pool_0_set_1".to_string(),
+        },
+        HealOptions {
+            pool_index: Some(0),
+            set_index: Some(1),
+            ..Default::default()
+        },
+        HealPriority::Normal,
+    );
+    request.source = HealRequestSource::AutoHeal;
+    request.heal_endpoints = targets.iter().map(|target| (*target).to_string()).collect();
+    request
+}
+
+#[tokio::test]
+async fn automatic_replacement_frozen_targets_reject_late_disks_but_cover_duplicates() {
+    for state in ["active", "retrying", "grace", "recovered", "recovery_anchor"] {
+        let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+        let manager = HealManager::new_without_root_recovery_for_test(storage.clone(), None);
+        let mut owner = automatic_replacement_request(&["replacement-a"]);
+        let owner_id = owner.id.clone();
+        if state == "grace" {
+            let now = SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .expect("test clock")
+                .as_secs();
+            let task = HealTask::from_request(owner, storage.clone());
+            owner = retry_request_for_result_with_budget(
+                &task,
+                &Err(Error::DanglingDeleteDeferred {
+                    retry_not_before: now + 60,
+                }),
+            )
+            .await
+            .expect("grace wait keeps the generation replayable")
+            .0;
+            assert_eq!(owner.retry_attempts, 0, "grace must not consume a failure retry");
+        }
+        match state {
+            "active" => {
+                manager
+                    .active_heals
+                    .lock()
+                    .await
+                    .insert(owner_id.clone(), Arc::new(HealTask::from_request(owner, storage.clone())));
+            }
+            "retrying" => {
+                manager.retrying_heals.lock().await.insert(
+                    owner_id.clone(),
+                    RetryingHeal {
+                        request: owner,
+                        error: "target unavailable".to_string(),
+                        cancel_token: CancellationToken::new(),
+                    },
+                );
+            }
+            "grace" | "recovered" => {
+                let config = manager.config.read().await.clone();
+                let mut queue = manager.heal_queue.lock().await;
+                let context = if state == "grace" { "retry" } else { "auto_scan" };
+                assert_eq!(
+                    HealManager::admit_request_to_queue(&mut queue, owner, &config, context, true).result,
+                    HealAdmissionResult::Accepted
+                );
+                assert!(
+                    queue
+                        .heap
+                        .iter()
+                        .any(|item| item.request.id == owner_id && item.replacement_targets_frozen),
+                    "zero retries cannot mean the scope is fresh"
+                );
+            }
+            _ => {
+                manager.submit_heal_request(owner).await.expect("queue frozen generation");
+                if state == "recovery_anchor" {
+                    manager
+                        .replacement_recovery_anchors
+                        .lock()
+                        .expect("recovery anchors")
+                        .insert(owner_id.clone(), "survivor".to_string());
+                }
+            }
+        }
+
+        let late = manager
+            .submit_heal_request(automatic_replacement_request(&["replacement-b"]))
+            .await
+            .expect("classify late target");
+        assert_eq!(late, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning), "{state}");
+        let covered = manager
+            .submit_heal_request_with_receipt(automatic_replacement_request(&["replacement-a"]))
+            .await
+            .expect("covered target remains owned");
+        assert_eq!(covered.result, HealAdmissionResult::Merged, "{state}");
+        assert_eq!(covered.task_id, owner_id, "{state}");
+        let targets = match state {
+            "active" => manager.active_heals.lock().await[&owner_id].heal_endpoints.clone(),
+            "retrying" => manager.retrying_heals.lock().await[&owner_id].request.heal_endpoints.clone(),
+            _ => {
+                manager
+                    .heal_queue
+                    .lock()
+                    .await
+                    .pop_next()
+                    .expect("retained generation")
+                    .heal_endpoints
+            }
+        };
+        assert_eq!(targets, ["replacement-a"], "{state}: frozen scope must not grow");
+    }
+}
+
+#[tokio::test]
+async fn automatic_replacement_merge_preserves_options_policy_and_recovery_identity() {
+    let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
+    manager
+        .submit_heal_request(automatic_replacement_request(&["replacement-a"]))
+        .await
+        .expect("queue fresh generation");
+    for variant in [
+        "dry_run",
+        "bucket_scope",
+        "pool_scope",
+        "retried_incoming",
+        "recovery_anchor",
+        "merge_disabled",
+    ] {
+        let mut incoming = automatic_replacement_request(&["replacement-b"]);
+        match variant {
+            "dry_run" => incoming.options.dry_run = true,
+            "bucket_scope" => {
+                if let HealType::ErasureSet { buckets, .. } = &mut incoming.heal_type {
+                    buckets.push("other-bucket".to_string());
+                }
+            }
+            "pool_scope" => incoming.options.pool_index = Some(2),
+            "retried_incoming" => {
+                incoming.heal_endpoints = vec!["replacement-a".to_string()];
+                incoming.retry_attempts = 1;
+            }
+            "recovery_anchor" => {
+                incoming.heal_endpoints = vec!["replacement-a".to_string()];
+                manager
+                    .replacement_recovery_anchors
+                    .lock()
+                    .expect("recovery anchors")
+                    .insert(incoming.id.clone(), "survivor".to_string());
+            }
+            "merge_disabled" => {
+                manager.config.write().await.low_priority_merge_enable = false;
+                incoming.priority = HealPriority::Low;
+            }
+            _ => unreachable!("test variant"),
+        }
+        let result = manager
+            .submit_heal_request(incoming)
+            .await
+            .expect("classify incompatible generation");
+        let reason = if variant == "merge_disabled" {
+            HealAdmissionDropReason::PolicyDropped
+        } else {
+            HealAdmissionDropReason::AlreadyRunning
+        };
+        assert_eq!(result, HealAdmissionResult::Dropped(reason), "{variant}");
+        let queue = manager.heal_queue.lock().await;
+        assert_eq!(queue.len(), 1, "{variant}");
+        assert_eq!(
+            queue.requests().next().expect("original generation").heal_endpoints,
+            ["replacement-a"],
+            "{variant}"
+        );
+    }
+}
+
+#[test]
+fn automatic_replacement_direct_admission_checks_frozen_targets_before_capacity() {
+    let config = HealConfig {
+        queue_size: 1,
+        ..HealConfig::default()
+    };
+    let mut queue = PriorityHealQueue::new();
+    let owner = automatic_replacement_request(&["replacement-a"]);
+    let owner_id = owner.id.clone();
+    assert_eq!(queue.push(owner), QueuePushOutcome::Accepted);
+    let admit = |queue: &mut PriorityHealQueue, request, frozen| {
+        HealManager::admit_request_to_queue(queue, request, &config, "auto_scan", frozen).result
+    };
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-b"]), false),
+        HealAdmissionResult::Merged
+    );
+    let mut replay = automatic_replacement_request(&["replacement-a"]);
+    replay.id = owner_id.clone();
+    assert_eq!(admit(&mut queue, replay, true), HealAdmissionResult::Merged);
+    assert!(
+        queue
+            .heap
+            .iter()
+            .any(|item| item.request.id == owner_id && item.replacement_targets_frozen)
+    );
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-c"]), false),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-a"]), false),
+        HealAdmissionResult::Merged
+    );
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-a"]), true),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+    assert_eq!(queue.len(), 1);
+    let owner = queue.pop_next().expect("frozen owner retained");
+    assert_eq!(owner.id, owner_id);
+    assert_eq!(owner.heal_endpoints, ["replacement-a", "replacement-b"]);
+
+    let owner = automatic_replacement_request(&["replacement-a"]);
+    let owner_id = owner.id.clone();
+    assert_eq!(queue.push(owner), QueuePushOutcome::Accepted);
+    let mut conflicting_replay = automatic_replacement_request(&["replacement-b"]);
+    conflicting_replay.id = owner_id.clone();
+    assert_eq!(
+        admit(&mut queue, conflicting_replay, true),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+    assert!(
+        queue
+            .heap
+            .iter()
+            .any(|item| item.request.id == owner_id && item.replacement_targets_frozen)
+    );
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-c"]), false),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+
+    let mut queue = PriorityHealQueue::new();
+    let owner = automatic_replacement_request(&["replacement-a"]);
+    let owner_id = owner.id.clone();
+    assert_eq!(queue.push(owner), QueuePushOutcome::Accepted);
+    let mut incompatible_replay = automatic_replacement_request(&["replacement-a"]);
+    incompatible_replay.id = owner_id.clone();
+    incompatible_replay.options.dry_run = true;
+    assert_eq!(
+        admit(&mut queue, incompatible_replay, true),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+    assert!(
+        queue
+            .heap
+            .iter()
+            .any(|item| item.request.id == owner_id && item.replacement_targets_frozen)
+    );
+    assert_eq!(
+        admit(&mut queue, automatic_replacement_request(&["replacement-b"]), false),
+        HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
+    );
+}
+
 #[tokio::test]
 async fn test_submit_heal_request_returns_merged_for_duplicate() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);

crates/heal/src/heal/manager/tests/auto_scan.rs

@@ -258,6 +258,16 @@ impl ScannerFixture {
         assert_eq!(request.id, self.task_id, "scanner must replay the durable UUID");
         assert_eq!(request.heal_endpoints, vec![self.local_targets[0].clone()]);
         assert_eq!(request.source, HealRequestSource::AutoHeal);
+        assert!(
+            self.manager
+                .heal_queue
+                .lock()
+                .await
+                .heap
+                .iter()
+                .any(|item| item.request.id == request.id && item.replacement_targets_frozen),
+            "replayed durable targets cannot expand"
+        );
         assert!(matches!(
             request.heal_type,
             HealType::ErasureSet { set_disk_id, buckets }

crates/heal/tests/heal_request_api_compat.rs

@@ -0,0 +1,29 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use rustfs_heal::{HealOptions, HealPriority, HealRequest, HealType};
+use rustfs_heal_contracts::heal_channel::HealRequestSource;
+use std::time::SystemTime;
+
+#[test]
+fn existing_heal_request_literal_remains_constructible() {
+    let now = SystemTime::now();
+    let request = HealRequest {
+        id: "request".to_string(),
+        heal_type: HealType::ErasureSet {
+            set_disk_id: "pool_0_set_0".to_string(),
+            buckets: Vec::new(),
+        },
+        bucket_incarnation_id: None,
+        expected_mrf_bucket_incarnation_id: None,
+        options: HealOptions::default(),
+        priority: HealPriority::Normal,
+        source: HealRequestSource::AutoHeal,
+        force_start: false,
+        retry_attempts: 0,
+        heal_endpoints: vec!["drive-a".to_string()],
+        created_at: now,
+        enqueued_at: now,
+    };
+    assert_eq!(request.heal_endpoints, ["drive-a"]);
+}

crates/scanner/src/scanner_folder/tests/checkpoint_fixture.rs

@@ -22,6 +22,7 @@ use tokio::io::AsyncReadExt;
 
 mod completed_compaction;
 mod deep_compacted;
+mod memory_budget;
 mod periodic_snapshot;
 mod segment_observation;
 

crates/scanner/src/scanner_folder/tests/checkpoint_fixture/memory_budget.rs

@@ -0,0 +1,167 @@
+// Copyright 2026 RustFS Team
+// Licensed under the Apache License, Version 2.0.
+
+use super::*;
+
+const OBJECTS_PER_PREFIX: usize = 16;
+const FROZEN_FRONTIER: &str = "bucket/prefix-0000/0000";
+
+fn cache_shape(cache: &DataUsageCache, wire_bytes: usize) -> serde_json::Value {
+    let mut key_capacity_bytes = 0;
+    let mut child_links = 0;
+    let mut child_set_capacity = 0;
+    let mut child_key_capacity_bytes = 0;
+    for (key, entry) in &cache.cache {
+        key_capacity_bytes += key.capacity();
+        child_links += entry.children.len();
+        child_set_capacity += entry.children.capacity();
+        child_key_capacity_bytes += entry.children.iter().map(String::capacity).sum::<usize>();
+    }
+    // The canonical histograms own eleven and seven u64 counters. Map control
+    // bytes, allocator overhead, optional statistics and transient work are excluded.
+    let histogram_payload_lower_bound_bytes = cache.cache.len() * (11 + 7) * std::mem::size_of::<u64>();
+    let map_slot_size = std::mem::size_of::<(String, DataUsageEntry)>();
+    let retained_lower_bound_bytes = cache.cache.capacity() * map_slot_size
+        + key_capacity_bytes
+        + child_set_capacity * std::mem::size_of::<String>()
+        + child_key_capacity_bytes
+        + histogram_payload_lower_bound_bytes;
+    serde_json::json!({
+        "entries": cache.cache.len(),
+        "entry_capacity": cache.cache.capacity(),
+        "entry_size_bytes": std::mem::size_of::<DataUsageEntry>(),
+        "map_slot_size_bytes": map_slot_size,
+        "map_key_capacity_bytes": key_capacity_bytes,
+        "child_links": child_links,
+        "child_set_capacity": child_set_capacity,
+        "child_key_capacity_bytes": child_key_capacity_bytes,
+        "histogram_payload_lower_bound_bytes": histogram_payload_lower_bound_bytes,
+        "retained_lower_bound_bytes": retained_lower_bound_bytes,
+        "wire_bytes": wire_bytes,
+    })
+}
+
+async fn characterize_memory_budget(prefixes: usize, coverage_gap: bool) {
+    let (mut scanner, root) = build_test_scanner().await;
+    let _cleanup = TestGuard {
+        temp_dir: Some(root.clone()),
+    };
+    for prefix in 0..prefixes {
+        for object in 0..OBJECTS_PER_PREFIX {
+            write_checkpoint_object(&root, &format!("prefix-{prefix:04}/{object:04}"), &[(None, 1)]).await;
+        }
+    }
+    let identity = crate::DataUsageScanIdentity {
+        tier_registry_generation: crate::runtime_tier_registry_for_cycle(11, 7).await.generation,
+        ..bound_checkpoint().1
+    };
+    scanner
+        .old_cache
+        .prepare_bucket_checkpoint("bucket", 11, 7, SOURCE, PLAN, identity);
+    scanner.new_cache.info = scanner.old_cache.info.clone();
+    scanner.is_erasure_mode = true;
+    scanner.skip_heal.store(true, Ordering::SeqCst);
+    scanner.sleeper = DynamicSleeper::new(rustfs_config::ScannerSpeed::Fastest);
+    // Isolate the already-established gap state from filesystem fault injection.
+    // Every object below is healthy; this control does not reproduce the fault
+    // or durable debt that caused a real scan's coverage gap.
+    scanner.coverage_gap = coverage_gap;
+    if coverage_gap {
+        scanner.coverage_frontier = Some(FROZEN_FRONTIER.to_string());
+    }
+    scanner
+        .scan_folder(
+            CancellationToken::new(),
+            CachedFolder {
+                name: "bucket".to_string(),
+                parent: None,
+                object_heal_prob_div: 1,
+            },
+            &mut DataUsageEntry::default(),
+        )
+        .await
+        .expect("walk the real wide-root metadata fixture");
+
+    let expected_objects = prefixes * OBJECTS_PER_PREFIX;
+    let totals = scanner
+        .new_cache
+        .checked_flatten_complete_scope("bucket")
+        .expect("the measured cache must be connected and contain no orphan entries");
+    assert_eq!((totals.objects, totals.size, totals.versions), (expected_objects, expected_objects, 0));
+    assert_eq!(totals.failed_objects, 0, "the gap control must not fabricate object failure debt");
+    assert!(
+        scanner.usage_updates.is_none(),
+        "production-style scans must not retain a progress mirror"
+    );
+    let bucket = scanner.new_cache.find("bucket").expect("retained root inventory");
+    assert!(!bucket.compacted, "these root fanouts preserve prefix statistics");
+    assert_eq!(bucket.children.len(), prefixes);
+    if coverage_gap {
+        assert_eq!(scanner.coverage_frontier.as_deref(), Some(FROZEN_FRONTIER));
+        assert!(
+            scanner.new_cache.find(FROZEN_FRONTIER).is_some(),
+            "the frozen endpoint must remain addressable"
+        );
+    } else {
+        for prefix in 0..prefixes {
+            let path = format!("bucket/prefix-{prefix:04}");
+            let entry = scanner.new_cache.find(&path).expect("completed root prefix");
+            assert!(entry.compacted);
+            assert!(entry.children.is_empty());
+            assert_eq!((entry.objects, entry.size), (OBJECTS_PER_PREFIX, OBJECTS_PER_PREFIX));
+        }
+        assert_eq!(scanner.new_cache.cache.len(), prefixes + 1, "healthy completed leaves must be released");
+    }
+
+    scanner
+        .new_cache
+        .seal_scan_frontier(scanner.coverage_frontier.as_deref())
+        .expect("seal the measured coverage endpoint");
+    assert!(scanner.new_cache.validated_scan_frontier().is_some());
+    let encoded = scanner
+        .new_cache
+        .marshal_msg()
+        .expect("encode the measured cache with the production codec");
+    let restored = DataUsageCache::unmarshal(&encoded).expect("reload the measured coverage graph");
+    assert_eq!(restored.validated_scan_frontier(), scanner.new_cache.validated_scan_frontier());
+    assert_eq!(restored.info.scan_coverage_receipt, scanner.new_cache.info.scan_coverage_receipt);
+    let restored_totals = restored
+        .checked_flatten_complete_scope("bucket")
+        .expect("reloaded accounting graph");
+    assert_eq!((restored_totals.objects, restored_totals.size), (expected_objects, expected_objects));
+    if coverage_gap {
+        assert_eq!(restored.validated_scan_frontier(), Some(FROZEN_FRONTIER));
+    }
+
+    println!(
+        "SCANNER_MEMORY_BUDGET {}",
+        serde_json::json!({
+            "schema": 1,
+            "measurement_point": "folder_walk_return_before_bucket_finalization",
+            "scenario": if coverage_gap { "seeded_coverage_gap" } else { "healthy_forward_sweep" },
+            "synthetic_gap_control": coverage_gap,
+            "prefixes": prefixes,
+            "objects_per_prefix": OBJECTS_PER_PREFIX,
+            "objects": expected_objects,
+            "object_bytes": expected_objects,
+            "validated_frontier": restored.validated_scan_frontier(),
+            "cache": cache_shape(&scanner.new_cache, encoded.len()),
+        })
+    );
+}
+
+#[tokio::test]
+#[serial]
+async fn memory_budget_characterizes_wide_root_and_seeded_coverage_gap() {
+    let prefixes: &[usize] = match std::env::var("RUSTFS_SCANNER_MEMORY_BUDGET_PROFILE") {
+        Err(std::env::VarError::NotPresent) => &[64],
+        Ok(value) if value == "1" => &[64, 128, 256],
+        _ => panic!("RUSTFS_SCANNER_MEMORY_BUDGET_PROFILE must be absent or 1"),
+    };
+    for &prefixes in prefixes {
+        assert!(prefixes * (OBJECTS_PER_PREFIX + 1) < DATA_SCANNER_COMPACT_AT_CHILDREN);
+        for coverage_gap in [false, true] {
+            characterize_memory_budget(prefixes, coverage_gap).await;
+        }
+    }
+}

docs/README.md

@@ -9,6 +9,7 @@ collection:
 ## Operations
 
 - [Retained MRF responsibilities](operations/mrf-responsibilities.md) — unknown bucket identities, bounded probes, and unreadable checkpoint recovery.
+- [Read-only BTF prototype inspection](operations/btf-prototype-inspection.md) — bounded declarations, explicit void handling, and independent native reference tests.
 
 - [Internode HTTP streaming transport](operations/internode-http-transport.md) — protocol selection, pooled connection isolation, and distributed read validation.
 

docs/architecture/README.md

@@ -46,6 +46,7 @@ Required headings and strings in these files are asserted by `scripts/check_arch
 | [background-controller-contract.md](background-controller-contract.md) | adding a status snapshot or reconcile surface for a background service, or being tempted to fold several services into a generic controller |
 | [background-services-inventory.md](background-services-inventory.md) | you need one audited background service's desired source, current-status inputs, status surface, and declared side effects |
 | [scanner-usage-publication.md](scanner-usage-publication.md) | changing scanner data-usage cache publication, quota-visible usage snapshots, scanner cycle recovery, or the persisted scanner usage artifacts |
+| [scanner-dirty-journal-source-contract.md](scanner-dirty-journal-source-contract.md) | designing storage-verified provenance, completion receipts, repair responsibility or replay for the node-owned scanner dirty-producer journal (proposed; not runtime implemented) |
 | [scanner-usage-authority-decision.md](scanner-usage-authority-decision.md) | deciding whether quota admission depends on scanner data usage, removing scanner publication layers, or designing a scanner storage boundary |
 | [scanner-per-bucket-convergence.md](scanner-per-bucket-convergence.md) | changing what makes a scanner cycle `Complete`, `Superseded`, or converged, adding or re-classifying a dirty-usage producer, deciding whether a background writeback may invalidate a cycle, changing scanner worker concurrency or throttling, or activating scoped dirty acknowledgements |
 | [config-model-boundary-adr.md](config-model-boundary-adr.md) | touching the server-config model (`Config`, `KV`, `KVS`) or its persistence, or asking which crate owns which part of server configuration |

docs/architecture/scanner-dirty-journal-source-contract.md

@@ -0,0 +1,233 @@
+# Scanner dirty-journal metadata source contract
+
+**Use this when:** adding storage-verified provenance to the node-owned scanner dirty-producer journal, its write completion, repair responsibilities, or replay.
+
+**Status:** proposed implementation contract. The authority, mutation tickets and
+receipts below are not implemented by the current journal. This document does
+not enable a new writer, change a persisted format, migrate data, or authorize
+repair. The existing reporting and quota decisions remain in
+[scanner usage publication](scanner-usage-publication.md) and
+[scanner usage authority](scanner-usage-authority-decision.md).
+
+## Scope and current boundary
+
+The scope is only the node-owned dirty-producer journal currently implemented in
+`rustfs/src/storage/scanner_dirty_journal.rs`. Its configured endpoint selects an
+owner hash and the owned JSON object. Mutation callbacks coalesce into bounded
+state; the worker captures a state/revision under the mutex and releases it
+before storage I/O. `confirm_flush` cannot confirm a newer in-memory revision.
+Replay validates the stored owner and producer/generation fields. This is an
+asynchronous restart hint, not a durability receipt for each S3 mutation.
+
+The endpoint hash is a routing key, not a storage-verified namespace or writer
+generation. A successful configuration save is not, by itself, proof that all
+physical tail tasks completed. These facts must not be promoted into authority
+for the old source-unknown MRF entries associated with internal JSON writes.
+
+This contract introduces no generic internal-metadata authority. It does not
+cover usage publication, bucket metadata, bucket-incarnation markers or
+`pool.bin`. Pool metadata's transaction owner is not a dirty-journal owner.
+Storage primitives may be reused only if they enforce this scope's own fences.
+
+## Required identities
+
+The names below describe storage roles, not exported Rust APIs.
+
+| Identity | Required evidence | Meaning |
+|---|---|---|
+| Owner key | Domain-separated digest of the configured RPC authority | Selects the owned journal; never proves historical provenance |
+| Namespace bootstrap | Immutable committed bytes naming this class, owner key and a non-nil namespace UUID, with an independent native transaction commit ID and digest | Establishes a new namespace only after storage verification; never labels legacy bytes |
+| Writer epoch | A positive epoch activated by fenced CAS of the journal's active head | Excludes a replaced writer without treating process identity or restart as proof |
+| Mutation ticket | Exact namespace/epoch/bootstrap binding, responsibility and mutation UUIDs, target pool/set/bucket/key, producer/trigger, payload digest and retained payload bytes or an immutable verified payload reference | Retains one new write's responsibility before dispatch |
+| Completion receipt | The exact ticket, committed target revision/digest, terminal physical-tail evidence and read-back binding | Can settle only that new responsibility |
+
+Namespace, bootstrap-commit, responsibility and mutation UUIDs must be non-nil.
+The native target revision is an explicit observed-absent or observed-existing
+CAS token, not an empty/default "current" value. Read failure, unsupported data
+and proven absence are distinct; only proven absence permits conditional create.
+
+The namespace UUID is generated only for an explicit new bootstrap transaction
+and becomes usable only after that transaction is fully verified. Missing,
+corrupt, conflicting, unsupported or unverified bootstrap data means unavailable
+authority. It must not become a new current identity through a default value.
+Unknown legacy entries retain their original identity, object/scope, status,
+enqueue/held times and operator audit.
+
+## Bootstrap and epoch activation
+
+Bootstrap requires a storage-native conditional-create transaction with its own
+durable recovery owner. It must retain indeterminate creation work independently
+of the namespace being created: bootstrap cannot prove its own success by
+decoding its JSON or by borrowing a current endpoint, process or pool owner.
+
+The transaction creates immutable bootstrap bytes, drains native work, validates
+the durability result and reads back the exact committed bytes. Only this
+storage-verified result may construct the opaque bootstrap binding. Arbitrary
+decoded records cannot construct it. Concurrent creation must select the same
+committed record or fail closed; it cannot overwrite a competing namespace.
+
+The active checkpoint is the single CAS surface for both the writer epoch and
+the journal publication. An epoch transition cannot be a separate unguarded
+object update followed by an ordinary PUT. Every physical mutation must carry a
+storage-native parent/epoch condition effective at the final commit boundary.
+The current generic configuration precondition callback is not assumed to
+provide this capability. An implementation lacking the primitive cannot enable
+source-bearing writes.
+
+Epoch activation also retains all old in-flight work and debt. Every admitted
+write needs a native terminal record; missing or indeterminate records remain
+unresolved. A new process, expired lease or current registry observation cannot
+stand in for proof that the old physical operations ended.
+
+## Producer, binding and dispatch
+
+Only the actual dirty-journal worker may use this class. Its target must be the
+owner-selected journal object in the storage metadata namespace, with the exact
+pool/set routing identity checked at admission and commit. The bootstrap object
+and unrelated metadata objects are outside the ticket's repair scope.
+
+User metadata, a decoded wire record or an authenticated RPC claim cannot
+construct a verified binding. A receiving storage peer must validate the
+bootstrap/ticket, target and native fence again before mutation; transport
+authentication alone is not this scope's repair authorization.
+
+The new head key is
+`scanner/dirty-journal-source/v1/<owner>/checkpoint.bin`, where `<owner>` is the
+canonical hash produced by `durable_dirty_usage_owner`. Immutable bootstrap
+records live under that owner's distinct `bootstrap/` keyspace. This does not
+reuse the shared or owner-only legacy JSON key. Mutation tickets and receipts
+belong to the bounded, source-aware responsibility/checkpoint store; it must not
+create an unbounded new object per callback or flush.
+
+The producer obtains a verified bootstrap and active epoch outside the journal
+state mutex. It captures the exact payload, revision and producer causes under
+that mutex, then releases it. The dirty-state/namespace/storage locks must not
+be hidden beneath the journal mutex while binding or writing.
+
+Authority I/O belongs to bootstrap and the journal worker. S3 callbacks remain
+bounded in-memory coalescing; unavailable authority retains dirty/unverified
+state and does not turn this restart hint into an S3 admission veto.
+
+Before dispatch, it durably admits the bounded mutation ticket and retains the
+exact input required to retry that mutation after restart. Ticket admission
+failure prevents dispatch and leaves the pending state unconfirmed. Digest-only
+evidence is insufficient when the exact payload cannot otherwise be recovered.
+
+After namespace locks are acquired, storage rechecks the bootstrap, active epoch,
+target scope and parent revision. A changed identity rejects the candidate; it
+does not rebind the old ticket to the current identity. The producer/trigger
+distinguishes initial dispatch, minority/tail failure, cancellation and native
+indeterminate completion. Missing trigger history remains unknown.
+The metadata producer is the journal worker; the original object-mutation
+producer labels remain payload hints. They do not confer metadata authority.
+
+## Commit, tail and publication
+
+A write-quorum ACK is not a completion receipt. Physical disk tasks retain the
+mutation identity and their owner until terminal completion, including when an
+async wrapper times out or its caller is cancelled.
+
+Completion must honor the selected durability policy, including its file,
+rename, parent and required-ancestor persistence boundaries. A mode that cannot
+supply durable evidence cannot produce a durable source receipt. It retains
+unverified state; this protocol neither silently forces extra fsync work in that
+mode nor weakens the configured durability policy to obtain success.
+
+System-metadata repair remains with this scoped owner. A user-bucket MRF proof
+cannot authorize it. `TailDrainedWithOwnerRepair` is an available completion
+boundary, not an authority token: a repair-required result must retain the
+ticket and retry the exact mutation or remain unresolved. It cannot be converted
+to success by a later ordinary read or a successful unrelated mutation.
+
+Publication requires the current epoch/parent fence, the completed native
+transaction and a read-back of the exact intended payload. The committed
+checkpoint binds the bootstrap, ticket, target revision, payload digest and
+native completion result. A torn checkpoint, stale epoch, conflicting bytes or
+unknown tail status prevents confirmation and responsibility settlement.
+
+`confirm_flush` additionally compares the captured revision with the current
+in-memory revision. Completion of an older snapshot must not acknowledge newer
+callbacks. Producer coalescing does not prove that a newer payload supersedes a
+different pending ticket; automatic cross-mutation settlement is out of scope.
+
+An empty verified state is a fenced tombstone publication, not an unguarded
+physical delete of the active head. It cannot remove another owner's record or
+any legacy/shared object. This keeps the epoch and completion binding available
+for replay.
+
+## Checkpoint, replay and exact settlement
+
+Source-bearing readers are introduced before a source-bearing writer. They
+validate the complete schema/version and byte limits, the immutable bootstrap
+and native transaction evidence, the active checkpoint, and every ticket's exact
+target and payload. A companion must bind the checkpoint owner, sequence, full
+length and digest; absence or a mismatch cannot be read as current authority.
+
+Bootstrap, ticket, checkpoint and receipt persistence retain their independent
+storage-transaction recovery owners. Publishing a receipt cannot use the same
+responsibility it settles as its sole evidence. A persistence failure leaves
+the original responsibility and native recovery work unresolved; it does not
+start a recursive chain of self-authorizing source claims.
+
+Replay may resume an exact source-bearing ticket only with the same verified
+namespace and mutation evidence. Reuse of an endpoint after restart does not
+change the historical source. A new epoch retains old debt until the required
+native terminal evidence and exact receipt exist.
+
+Only a durably published receipt matching the complete responsibility identity
+may settle it. Matching an object key, current owner, version, payload digest or
+successful newer write alone is insufficient. Source-unknown entries, another
+scope, another mutation, and legacy records are never automatically accepted,
+rechecked, retired or physically cleaned by this protocol.
+
+## Compatibility and bounded resources
+
+The legacy shared JSON and current owner-only JSON remain retained as legacy
+sources. They cannot be imported into a new authority namespace or overwritten
+to manufacture provenance. A reader-first rollout needs distinct source-bearing
+storage and an explicit minimum-reader/capability contract. An unsupported peer
+must reject source-bearing admission before mutation, rather than silently
+falling back to an owner-only write. Rollback cannot treat unknown fields as
+source authority or automatically remove source-bearing material.
+
+These are restrictions on the new protocol. A legacy process may still update
+its own old-format key; that activity is not provenance for the new namespace
+and must not be mistaken for byte-stable historical evidence.
+
+This protocol does not turn scanner reports or restart hints into quota
+admission authority. Existing full-scan, dirty-generation, peer-instance and
+movement fences remain required. The existing user-bucket MRF journal and old
+unknown responsibilities retain their current semantics.
+
+The current journal's 64 KiB record, entry/scope limits and queue byte limits
+remain constraints. Bootstrap bytes, tickets and retained payloads, companions,
+checkpoints, native transaction records and receipts are all charged explicitly.
+An implementation must not borrow an unrelated bucket's spare budget, omit
+fields, evict identities, silently increase limits or accumulate unbounded
+immutable objects. Exhaustion retains dirty state and blocks new admission.
+No epoch/ticket garbage collection is authorized without a separate exact
+ownership and durable-retention proof. This contract authorizes no cleanup of
+the original residual data.
+
+## Required conformance cases
+
+These are implementation requirements, not tests claimed to have passed.
+
+| Boundary | Required observation |
+|---|---|
+| Bootstrap absent, racing create, corrupt bytes or unknown native result | No authority token; retained creation/recovery evidence; legacy bytes unchanged |
+| Endpoint reuse, writer replacement, movement or stale parent | Native fence rejects the stale mutation; old in-flight work/debt is not relabelled |
+| Admission failure or crash before dispatch | No unrecorded write; exact retained ticket/payload can be read after restart |
+| Quorum ACK while a native tail is blocked | No completion receipt, checkpoint settlement or premature owner release |
+| Minority/tail failure, caller timeout/cancel or fence loss | Same ticket remains unresolved; no generic user-bucket repair authorization |
+| Crash before/after native commit and checkpoint publication | Exact replay outcome or explicit indeterminate state, never a guessed success |
+| Callback or dirty acknowledgement concurrent with I/O | Only the captured snapshot can complete; newer state remains pending |
+| Partial/torn/misbound companion, payload or receipt | Fail closed without adopting a current namespace or deleting retained evidence |
+| Exact success versus same-key/different-source/mutation success | Settle only the exact new ticket; all other and legacy unknown entries remain |
+| Empty state, capacity exhaustion and unsupported reader | Fenced tombstone or retained failure; no broad delete, capacity relaxation or implicit downgrade |
+
+The current owner-only journal, its tests and the pool metadata contract are
+useful foundations, but none alone satisfies this source contract. Runtime
+implementation and conformance acceptance are separate from this design. If the
+contract enters a performance comparison, both arms must first use the same
+accepted implementation and be fixed and registered again.

docs/operations/btf-prototype-inspection.md

@@ -0,0 +1,58 @@
+# Read-only BTF prototype inspection
+
+Use `scripts/inspect_btf_prototypes.py` to inspect declared function types in a
+raw, little-endian BTF v1 regular file. Symlinks, devices and pipes are rejected.
+The command reads the named file and writes one
+JSON document to stdout. It does not load BPF, attach probes, read kernel memory,
+change a service, or write to the source file.
+
+```bash
+python3 scripts/inspect_btf_prototypes.py \
+  --btf /sys/kernel/btf/vmlinux \
+  --function bpf_ringbuf_output
+```
+
+Repeat `--function` for up to 16 unique, exact names. The result preserves request
+order, the observed source size/SHA-256, function/prototype type IDs, original
+parameter type IDs, qualifier chains, and resolved declarations. A missing or
+ambiguous function is an error, rather than a substitute selected by proximity.
+The schema version is `1`.
+
+Selected type records expose their common BTF headers. Kind-specific payloads
+and recursive layouts are not decoded by this command; use an independent BTF
+dump when those records are required. A reported declaration is not a complete
+C type, layout proof or runtime calling convention.
+
+The reader supports BTF kinds 1 through 19, indexes at most 500,000 records,
+reads at most 32 MiB plus one overflow-detection byte, limits names to 4,096
+bytes, follows at most 32 qualifier records, and accepts at most 16 parameters
+per selected prototype. Reports above 1 MiB are rejected before any JSON prefix
+is emitted. Invalid/truncated/unsupported source, missing resolved type IDs,
+malformed pointer/qualifier headers, and ambiguous selectors exit nonzero with
+empty stdout. Success exits zero. Input I/O and inspection errors exit `2`.
+
+Under the [Linux BTF format](https://docs.kernel.org/bpf/btf.html), ID zero denotes
+void without a stored type record. The tool permits it at explicit return and
+argument-pointee boundaries, including qualified void. A void descriptor has no
+invented size or offset. The tool does not infer a pointer width, dereference
+runtime pointers, or certify that a declared type matches the currently executing
+machine code. It is a prototype inspector, not a complete BTF semantic verifier,
+recursive layout validator, caller profiler, or runtime ABI acceptance gate.
+
+Run the local contract tests with:
+
+```bash
+python3 -B scripts/test_inspect_btf_prototypes.py
+```
+
+The `BTF prototype reference` workflow additionally compiles a small C fixture,
+extracts its raw `.BTF` section, and independently compares both selected
+declarations and resolved types against `bpftool` output. The object is never
+loaded. The workflow selects a concrete `bpftool` executable by availability and
+version before comparing data; it does not depend on the hosted kernel's tools
+wrapper or switch tools after a comparison failure. The native
+[file-dump mode](https://github.com/libbpf/bpftool/blob/main/docs/bpftool-btf.rst)
+accepts ELF or raw BTF files. That comparison validates
+the tool's declared-format scope; it does not
+qualify an original test node's kernel generation, complete observer budget, or
+storage/business performance experiment.

docs/operations/hotpath-warp-ab-runbook.md

@@ -1,7 +1,7 @@
 # Hotpath warp runbook (A/B gate and ABBA evidence)
 
-**Use this when:** you need S3-face (PutObject / GetObject / mixed) performance evidence for a code change: the nightly relative-budget gate, a quick local A/B, or a formal ABBA run whose numbers will be quoted in a PR.
-**Source of truth:** `scripts/run_hotpath_warp_ab.sh` (quick A/B rig), `scripts/run_hotpath_warp_abba.sh` (ABBA runner; also what CI executes), `scripts/hotpath_warp_ab_gate.sh` (relative-budget gate), `scripts/run_object_batch_bench_enhanced.sh` (warp driver, medians, `baseline_compare.csv`), `.github/workflows/performance-ab.yml` (CI gate).
+**Use this when:** you need S3-face (PutObject / GetObject / mixed) performance evidence for a code change: a quick local A/B, or a formal ABBA run whose numbers will be quoted in a PR.
+**Source of truth:** `scripts/run_hotpath_warp_ab.sh` (quick A/B rig), `scripts/run_hotpath_warp_abba.sh` (ABBA runner), `scripts/hotpath_warp_ab_gate.sh` (relative-budget gate), `scripts/run_object_batch_bench_enhanced.sh` (warp driver, medians, `baseline_compare.csv`).
 
 This runbook compares two binaries. To sweep one `RUSTFS_*` runtime knob at a time against a fixed binary, use [object-io-tuning-ab-matrix.md](object-io-tuning-ab-matrix.md) instead.
 
@@ -10,7 +10,7 @@ Two entry points share one workload matrix, one gate, and one deploy-hook shape:
 | Entry | Script | Legs per cell | Use |
 | --- | --- | --- | --- |
 | Quick A/B | `scripts/run_hotpath_warp_ab.sh` | baseline → candidate | local smoke, fast triage of a suspected regression |
-| ABBA | `scripts/run_hotpath_warp_abba.sh` | A1 baseline → B1 candidate → B2 candidate → A2 baseline | formal evidence; the CI gate |
+| ABBA | `scripts/run_hotpath_warp_abba.sh` | A1 baseline → B1 candidate → B2 candidate → A2 baseline | formal evidence |
 
 This rig is the only entry point for S3-face performance coverage. There is deliberately no in-process criterion benchmark for those operations: a criterion harness that stands up an embedded server measures the harness, not the S3 path. Micro-benchmarks stay at function level (EC encode, `xl.meta` parse, `rename_data`). Knob-level sweeps against a fixed binary are a different question; see [object-io-tuning-ab-matrix.md](object-io-tuning-ab-matrix.md).
 
@@ -163,16 +163,6 @@ After each window confirm the `.json.gz` profile and its `.syms.json` sidecar ar
 
 Instrumented builds (features in `rustfs/Cargo.toml`): `--features hotpath-alloc` for allocation attribution, `--features hotpath-cpu` for CPU hotpath sections. Compare instrumented binaries only with other builds of the same mode; never use them for throughput acceptance, because the instrumentation changes what is measured.
 
-## CI gate (`performance-ab.yml`)
+## Performance gate status
 
-| Aspect | Value |
-| --- | --- |
-| Triggers | `schedule` (nightly cron `31 6 * * *` UTC against `main`) and `workflow_dispatch` (inputs `duration`, default `12s`; `allow_regression`, boolean). No `pull_request` trigger and no label gating. |
-| Jobs | `warp-ab` (runner `sm-standard-2`, `timeout-minutes: 180`); `alert-on-failure` (opens the scheduled-failure issue; scheduled runs only). |
-| Baseline commit | scheduled: head of the last successful scheduled run (falls back to the candidate itself when there is none); dispatch: `origin/main`. The baseline must be an ancestor of the candidate. |
-| Binary cache | `actions/cache` inside `warp-ab`, key `rustfs-baseline-<baseline_sha>`. Miss: source build (same-commit runs build once and reuse the binary for both phases); a run whose gate passes saves the candidate binary under `rustfs-baseline-<candidate_sha>` for the next night. |
-| Command | `scripts/run_hotpath_warp_abba.sh --duration <input> --rounds 3 --cooldown 5 --health-timeout 180 --baseline-revision <sha> --candidate-revision <sha> --baseline-bin ... --candidate-bin ...` |
-| Exemption | `allow_regression=true` on dispatch adds `--allow-regression --exemption-reason "workflow dispatch override"`. |
-| Artifacts | `hotpath-warp-ab-<run_number>` containing `target/hotpath-abba/` (14-day retention); the step summary renders `candidate_gate.md`, or the server-log tails when the rig failed before the gate. The `Enforce gate` step fails the job on a non-zero rig exit. |
-
-The nightly detects a regression within a day of landing; it does not block a merge. For pre-merge evidence run the ABBA procedure above and attach the outputs to the PR.
+There is no scheduled CI performance gate. The ABBA procedure above is the evidence path for performance comparisons; attach its outputs to the PR that claims a performance effect. A scheduled regression gate requires calibrated budgets: on the previous gate's runner, measured drift between identical binaries at the default 12-second rounds exceeded the former 10% fail budget, so the scheduled entry never seeded a baseline (the workflow only ever passed on its pre-removal push trigger). Restore a scheduled gate only after a noise study grounds the budgets on the target runner.

docs/operations/scanner-benchmark-runbook.md

@@ -41,7 +41,7 @@ The `--abba` mode runs five independent scenario cells: `cold-hot`, `fresh-hot`,
 three A1/B1/B2/A2 groups for both baseline/candidate with background work on,
 and candidate-only background off/on. A measured release leg lasts at least 7200
 seconds; the minimum matrix contains 120 legs (240 hours before setup/oracles).
-The existing `performance-ab.yml` supplies the pattern for immutable build
+The hotpath warp ABBA rig (`scripts/run_hotpath_warp_abba.sh`) supplies the pattern for immutable build
 provenance and failure propagation, but its short Warp workload is not this
 scanner gate. No scheduled workflow starts this matrix automatically.
 

docs/operations/scanner-runtime-controls.md

@@ -727,6 +727,8 @@ Stale-upload cleanup differs in one crash-recovery detail. RustFS's stale multip
 
 ## Replacement Recovery Completion
 
+One reconnect sweep groups the newly attached blank disks in a set. If another reconnect is still pending, the ready targets are submitted after a one-second window rather than waiting indefinitely; later targets are submitted separately. Compatible automatic requests can add targets while the owner is still queued and has never executed or recovered a durable generation. Once execution or recovery fixes the target scope, later disks cannot join that generation or receive a merged receipt from it. Blank disks outside the frozen scope remain discoverable for the next automatic scan after the owner finishes.
+
 `POST /v3/background-heal/status` is an execution-queue view. `state=idle`, zero queue and active counts, an online disk, a readable object, or acceptance of an admin deep-heal request do not independently prove that a replacement disk contains every erasure shard.
 
 Treat replacement recovery as verified only after the repair task has completed for the exact replacement instance and an operator has confirmed the target disk contains the expected `xl.meta` and data parts for every relevant object version. A replacement that is not mounted, is unsafe to format, loses its marker, or returns a partial target outcome must be treated as deferred or incomplete. Do not automate destructive replacement actions from an `idle` observation alone. A new node must not infer replacement completion from an old or unavailable peer; regard that information as unknown or degraded until every required peer can report the same replacement instance and verified completion.

docs/testing/README.md

@@ -28,6 +28,8 @@ The [scanner checkpoint fixture](scanner-checkpoint-fixture.md) diagnoses retain
 
 The [scanner cache cost profile](scanner-cache-cost.md) separates clone, subtree copy, encoding, and counted save costs without changing production cache behavior.
 
+The [scanner memory acceptance runner](scanner-memory-acceptance.md) verifies owned S3 datasets, scanner restart progress, and Linux process/cgroup memory evidence under a fixed container limit.
+
 The [Pool layout compatibility reference](pool-layout-compatibility.md) defines the topology and EC regression matrix for single-drive, single-node multi-drive, and multi-node expansion pools.
 
 The [Scanner/Heal topology protection gates](scanner-heal-topology-gates.md) map G-EXP, G-REB and G-DEC and their fault windows to the `scanner_heal_topology_` cases, their recorded inventory, and the basis of their timeouts.

docs/testing/ci-gates.md

@@ -99,7 +99,6 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 | `fuzz.yml` (nightly) | `nightly-fuzz-corpus` per target | gate; corpus and crash artifacts | yes | `MAX_TOTAL_TIME=<seconds> ./scripts/fuzz/run.sh` |
 | `minio-interop.yml` (nightly) | `minio-interop` | EC + SSE read-parity gate | yes, with `never_ran_grace_until` | pinned Docker fixture steps in the workflow |
 | `on-demand-migration-interop.yml` (nightly) | `minio-source`, `cloud-source` (`aws`, `r2`, `gcs`) | report-only provider interop; one JSON report per provider naming cases, timings and source request counts, plus JUnit and MinIO logs. A cloud provider whose `ODM_INTEROP_*` secrets are absent is skipped with a summary note, not failed | no | start the pinned MinIO container as in the job, export the `RUSTFS_ODM_INTEROP_*` variables, run `python3 scripts/e2e_binary.py build --bins`, then `python3 scripts/e2e_binary.py run -- cargo nextest run --profile e2e-odm-interop -p e2e_test` |
-| `performance-ab.yml` (nightly) | `warp-ab` | regression-budget gate; A/B summaries and server logs | yes | `bash scripts/run_hotpath_warp_abba.sh --help` |
 | `nightly-gnu.yml` (nightly) | `build`, `kms-vault-lane`, `kms-vault-ha-failover` | build, live Vault, and HA failover gates | yes | commands and pinned Vault images in the workflow |
 | `audit.yml` (nightly) | `cargo-deny`, `workflow-pin-report` | dependency and workflow-pin gates | yes | `cargo deny check`; `scripts/security/check_workflow_pins.sh` |
 | `mint.yml` (weekly) | `mint` | report-only by design; per-suite PASS/FAIL/NA and raw `log.json` | yes | pinned Docker sequence in the workflow |
@@ -111,7 +110,7 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 | `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 | no | dispatch, optionally with `source_ref` |
 | `scheduled-validation-freshness.yml` (nightly) | `check-freshness` | fails on missing or stale attempts or completed successes | n/a | dispatch |
 
-Manual `workflow_dispatch` runs are debugging evidence and do not open scheduled-failure issues. A manual performance run may explicitly allow a known regression; that override is not a passing baseline.
+Manual `workflow_dispatch` runs are debugging evidence and do not open scheduled-failure issues.
 
 ## Packaged functional acceptance
 

docs/testing/distributed-e2e.md

@@ -33,7 +33,7 @@ The expansion fixture is an all-current-binary fleet, so it initializes pool met
 - Pool expand, decommission, rebalance, checksum integrity, versioned and multipart data, and S3 during active movement
 - Bidirectional site-replication convergence plus enabled/synchronized peer state on both sites
 - A 24-worker mixed PUT/HEAD/GET/COPY/DELETE workload; concurrent PUT during active decommission
-- Node kill/restart, full process restart, node-facing TCP blackhole/recovery, in-flight streaming GETs across a peer kill, and fresh-drive replacement verified by physical `xl.meta`/part-shard census
+- Node kill/restart, full process restart, node-facing TCP blackhole/recovery, in-flight streaming GETs across a peer kill, and fresh-drive replacement verified by physical `xl.meta`/part-shard census. The EC:4 partial-return case pins one 16-drive set and actual 12+4 object metadata, stops two nodes (eight drives), then restores only one. All three live coordinators must read existing objects and write/read new objects while the fourth node remains stopped. This is a process-fault acceptance test, not physical disk-unmount evidence.
 - Multipart, cross-node listing, list-buckets agreement
 - Direct and rolling upgrade from the pinned previous release: historical objects, versioned history, and IAM user AK/SK still work afterwards
 - The `scanner_heal_topology_` protection matrix: versioned, locked and multipart inventory verified through every node across expansion, rebalance stop/restart and queued decommission, with kill and commit-failure injection at proven movement windows; gate map and timeouts in [scanner-heal-topology-gates.md](scanner-heal-topology-gates.md)
@@ -160,3 +160,11 @@ python3 scripts/e2e_binary.py run --features e2e-test-hooks -- \
 Match the S3 response request ID to the retained `http-request` span. The `object_namespace_lock` acquired and guard-dropped events carry `lock_attempt_id` and the logical guard `lock_id`. Each `object_namespace_lock_peer` event maps that guard to an actual granted `peer_lock_id`; acquisition retries can give peers an ID different from the logical guard. Join `peer_lock_id` to the `lock_id` in `namespace_lock_release`. Its local `state=mapped` record identifies the process-local `physical_guard_id`; the separate `state=released` record is emitted only after the underlying manager actually releases that guard and unregisters it. A stale or disabled guard cannot produce that physical completion record. Remote `state=acknowledged` records require a successful unlock response without an error; they describe the remote response, while physical completion still requires the peer's local record. Batch acknowledgements retain each actual lock ID and peer. Missing, false or errored release responses do not produce an acknowledgement. Combine `process_id` and physical guard IDs with the retained node process/lifecycle provenance; guard IDs are unique only within one process.
 
 `guard_dropped` still means that logical guard ownership ended; asynchronous peer cleanup can follow it. The local mapping is logged after dropping the guard so logging does not delay that physical release; join records by identity rather than arrival order. A lock-acquisition failure or a fast local guard has no acquired distributed lock ID. Transport request IDs may differ from the S3 request ID; use the actual lock identity rather than assuming equality. These events describe ordinary namespace cleanup; forced release and lease reaping require separate evidence. They do not establish completion of native filesystem callbacks, snapshot leases or detached mutation tails. Preserve their separate physical-ownership evidence and the final scoped lock-table snapshots. This diagnostic configuration increases logging and is not a performance-measurement configuration.
+
+### Late part transaction packets
+
+The registered late-Complete case also retains original signed `PreparePartTransaction` and `SettlePartTransaction(Commit)` packets from an 8 MiB replacement of an acknowledged 9 MiB part. Two peers hold one original packet each until the single-attempt S3 replacement fails below quorum. The one-packet limit permits fresh recovery requests for that same upload. Ordinary Abort must reclaim every original source before delivery. Each method runs with and without same-name bucket recreation.
+
+Before releasing the old packets, create and acknowledge a new upload at the same key. Preserve every fresh source/control/shard byte, then require the original packets to reach their peers with successful gRPC transport and their typed application result. Prepare must refuse. Settlement after its transaction directory was removed may return idempotent success; success alone does not establish that no write occurred. The case separately checks absent old roots, exact fresh-source snapshots, the prior committed object through all four clients, old-upload NoSuchUpload, and completion of the fresh upload with its full acknowledged body. No expired packet is discarded or re-signed to obtain this delivery evidence.
+
+Proxy regressions retain unknown protobuf fields and gRPC trailers, distinguish success/refusal/ambiguous results, inspect the captured settlement action, and exclude gRPC authentication failure from application outcomes. The local `native_multipart_late_part_mutators_cannot_recreate_removed_upload` regression additionally supplies an existing TMP shard while the source is missing, so missing temporary input cannot substitute for source-admission protection. These EC:2 non-inline cases do not qualify late rollback packets, empty-part transactions, other layouts or actual server-task cancellation.

docs/testing/scanner-memory-acceptance.md

@@ -0,0 +1,106 @@
+# Scanner memory acceptance
+
+**Use this when:** validating scanner memory with large S3 namespaces, a memory-limited Linux process, interrupted scans, or a separate scanner-disabled request burst.
+**Source of truth:** `scripts/scanner_memory_acceptance.py`; scanner status and usage contracts in `rustfs/src/admin/handlers/scanner.rs`, `rustfs/src/admin/handlers/account_info.rs`, and `crates/scanner-metrics/src/metrics.rs`.
+
+This is an operator-run acceptance harness, not a CI gate or a throughput benchmark. It creates real S3 objects and preserves raw observations in a private run directory. It never starts, restarts, stops, or removes a service, container, volume, bucket, or base fixture object. Service lifecycle belongs to the operator. The only deletes are copies created under the current run's uniquely named burst prefix.
+
+## Evidence and limits
+
+Run wide and nested layouts as separate manifests. The defaults are 2,000,000 objects of 50 KiB, with 170,000 direct source prefixes for the wide layout. Nested keys put the same sources below `photos/xx/yy/`. Payloads are deterministic pseudorandom bytes, shared by all fixture objects; they are not a variable-size photo distribution. Each bucket also contains one small ownership marker. Exact listing and scanner usage checks include this marker separately, so expected scanner totals are **N + 1 objects**, with the marker's serialized bytes added to logical payload size.
+
+Every live observation verifies the running executable digest, process start identity, cgroup v2 memory limit and S3 endpoint binding. Docker mode additionally verifies the container's run label, bound data source and, for named volumes, the volume's run label. The source revision is declared build provenance; a digest match proves which binary is running, not that the declared source was independently rebuilt by this script. Keep the exact source archive and build receipt beside the evidence.
+
+Samples distinguish process RSS, anonymous RSS, `smaps_rollup` anonymous memory, cgroup current/peak usage, anonymous/file/slab/kernel accounting, swap and OOM events. A missing reading remains unknown. Optional Prometheus collection accepts live allocator requested bytes only when `rustfs_memory_allocator_malloc_requested_bytes_available` equals one. Reserved/committed bytes and allocation totals are not substitutes for active heap.
+
+`summary.json` reports phase measurements and restart evidence, but deliberately does not automatically certify overall acceptance. A finite run cannot prove a universal memory bound, and RSS retention alone is not a memory-leak diagnosis. Inspect the trend across completed cycles, anonymous memory versus file/slab accounting, live heap when available, and the burst idle window. Record the actual object scale, duration, resource limit and unavailable measurements in the conclusion. A small pilot does not satisfy the 2M-object case.
+
+## Prerequisites and ownership
+
+Use Python 3.11+ with `boto3`, a disposable local RustFS instance, and disposable credentials supplied as `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`. The runner bypasses HTTP proxies explicitly for localhost traffic. It accepts only literal `127.0.0.1` or `::1` endpoints with explicit ports; ambiguous `localhost` names are rejected. TCP bindings must match the endpoint address and IP family, or an explicitly observed wildcard in that same family. Credentials and raw SDK error messages are never written to evidence.
+
+Docker containers need `sh`, `cat`, `stat`, `readlink`, `sha256sum`, `df`, and `tail`, a readable `/proc`, and cgroup v2. Bind the tested binary read-only. Use a four-GiB memory limit and, for an exact no-swap limit, matching `--memory-swap=4g`. The runner records and verifies the actual kernel limit rather than trusting a launch command. Native Linux `--pid PID` mode requires the current user to own the process, a dedicated cgroup containing only that PID, the endpoint listening socket to belong to it, and the same mount namespace and filesystem root as the observer.
+
+`prepare` creates a new mode-0700 run directory and immutable manifest. It prints a generated run ID, bucket name and required label. The bucket is never selected by a user-supplied name. On every writing batch, the runner verifies its exact ownership marker and requires versioning to remain disabled. Resume markers bind to the manifest digest. Reusing an existing unmarked bucket, a foreign container, or a foreign volume fails closed.
+
+Before any workload write and on every live sample, the runner reads authenticated `/rustfs/admin/v3/storageinfo` from the serving process. Credentials need permission for this route as well as scanner status. The inventory must be fresh, complete against `DrivesPerSet`/`TotalSets`, healthy and local to one node; missing, cached, remote or incomplete observations fail closed. Every effective disk path must resolve inside the manifest's bind/named-volume mount or native `data_root`, on that same filesystem. Symlink escapes, nested mounts and a different process mount namespace/root are rejected. The inventory reflects expanded server configuration, including positional CLI arguments that override `RUSTFS_VOLUMES`; an unused owned mount cannot authorize a foreign write directory. Storage bindings are hashed into phase identity. Resolved disk paths and raw storage-info responses stay out of samples.
+
+The storage reserve is one GiB plus 1.2 times remaining payload and 16 KiB per-object metadata allowance. Bind/native modes check host free space at preparation. Named-volume mode defers the capacity check until the labelled container is running, and checks `df` inside the mounted volume before the first PUT. Batches recheck the sampled free-space allowance. This is a guard against an undersized test volume, not a promise about every filesystem's physical overhead.
+
+## Prepare and run a small pilot
+
+Use a new absolute output path. Values below are placeholders; source and binary digests must be taken from the exact build being tested. Keep binary/build artifacts outside the repository.
+
+```sh
+python3 scripts/scanner_memory_acceptance.py prepare /tmp/scanner-memory-wide-pilot \
+  --endpoint http://127.0.0.1:19000 \
+  --source-sha SOURCE_COMMIT_SHA \
+  --binary-sha256 BINARY_SHA256 \
+  --layout wide --objects 4096 --root-prefixes 256 --object-bytes 51200 \
+  --data-volume scanner-memory-wide-pilot
+```
+
+Create that **new** named volume and the disposable service container with the printed `rustfs.scanner-memory.run-id=UUID` label. Start with `RUSTFS_SCANNER_ENABLED=false`, healing disabled, the chosen binary and the manifest's four-GiB limit. Publish the S3 port directly to the supplied loopback endpoint. Alternatively use `--data-root /absolute/owned/data` at preparation and bind that exact directory.
+
+```sh
+python3 scripts/scanner_memory_acceptance.py load /tmp/scanner-memory-wide-pilot \
+  --container scanner-memory-wide-pilot --workers 48 --seconds 3600
+python3 scripts/scanner_memory_acceptance.py verify /tmp/scanner-memory-wide-pilot
+```
+
+Only one action may hold a run directory at a time. Load checkpoints advance after a complete bounded batch. An interrupted batch can be safely repeated because fixture keys and bodies are deterministic and the owned bucket is unversioned. A timeout or failed request leaves progress for the next invocation and exits nonzero; it never produces passing load evidence.
+
+For the full-scale matrix, create separate new manifests with the defaults, once with `--layout wide` and once with `--layout nested`. Do not change a pilot manifest's object count or repurpose its bucket. Allocate adequate disk and wall time; two million S3 PUTs are real work.
+
+## Scanner cycles and external restart
+
+Recreate or restart the owned service externally with scanner enabled, preserving the same volume and binary. Record the chosen runtime knobs, including `RUSTFS_SCANNER_CYCLE`, `RUSTFS_SCANNER_CYCLE_MAX_OBJECTS`, `RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS`, and the disk I/O budget. Runtime configuration samples capture those values. A change in binary requires a separate manifest.
+
+```sh
+python3 scripts/scanner_memory_acceptance.py probe /tmp/scanner-memory-wide-pilot \
+  --container scanner-memory-wide-pilot --label before_restart \
+  --seconds 600 --stop-after-progress 128
+# Externally stop/restart the owned process while preserving its data.
+python3 scripts/scanner_memory_acceptance.py probe /tmp/scanner-memory-wide-pilot \
+  --container scanner-memory-wide-pilot --label resumed_cycles \
+  --seconds 1800 --min-completed-cycles 2
+```
+
+`--stop-after-progress` ends observation after scanner status reports an active cycle with the requested amount of work. It does **not** cancel the scanner itself. Act on the external process promptly and retain the lifecycle event evidence; a completed cycle before the restart is not an interrupted-scan test. Each invocation has a unique phase ID and binds to one process identity through its final inventory and usage verification. It rejects a restart at any point in that phase. `--seconds` is the total deadline, including verification; a probe exits once its requested progress or new completed-cycle count is observed. A later successful probe on a different identity is recorded as a restart window, after exact S3 inventory and converged admin usage verification. Completion counts include only newly observed successful cycle-end timestamps, not a previous completion present when the probe began.
+
+A budgeted partial sweep is a separate state from process cancellation. Pair real observations with the scanner checkpoint fixture described in [scanner-checkpoint-fixture.md](scanner-checkpoint-fixture.md); do not infer process-crash durability from an in-process test.
+
+`restart_window_observed` alone does not prove checkpoint reuse: the new process could have rescanned the entire namespace. Checkpoint-reuse acceptance additionally needs persisted checkpoint identity/frontier evidence and metadata-visit evidence showing that completed prefixes were retained rather than reread. Likewise, a successful cycle can reuse cached usage; its success does not prove a full namespace walk. Keep the recorded object-visit counts and distinguish full-walk, cached-cycle and restart-reuse evidence in the conclusion.
+
+## Scanner-disabled burst and idle window
+
+Restart externally with scanner disabled, then run the separate request burst:
+
+```sh
+python3 scripts/scanner_memory_acceptance.py burst /tmp/scanner-memory-wide-pilot \
+  --container scanner-memory-wide-pilot --workers 48 \
+  --burst-objects 4096 --seconds 3600 --idle-seconds 1800
+python3 scripts/scanner_memory_acceptance.py summary /tmp/scanner-memory-wide-pilot
+```
+
+Each operation copies a base fixture object to an owned scratch key, HEADs its length, and deletes that copy. Completed batches are resumable. Results distinguish the burst ID, initial/final cursors, newly completed objects and cumulative progress. Three logical requests per newly completed object do not count SDK retry attempts; observed HTTP request count remains unknown. An idle-only continuation reports zero new logical requests. `--seconds` includes the workload, inventory verification and the requested idle interval. The full inventory must return to the original exact keys and byte count before the idle window begins. No `delete_objects`, bucket deletion or broad prefix cleanup is performed. An aborted burst can leave owned scratch objects; resume the same burst to reconcile them rather than deleting unrelated keys.
+
+Supply `--allocator-metrics-url http://127.0.0.1:PORT/metrics` to live phases only when that endpoint exports this process's allocator metrics. Without an authenticated binding between that metrics endpoint and the observed process, treat allocator attribution as operator-supplied context. The runner cannot fabricate live-heap evidence from process RSS.
+
+## Artifacts and validation
+
+- `manifest.json`: immutable workload, source declaration, binary digest, storage source and memory limit.
+- `state.json`: manifest-bound load and burst resume positions.
+- `observations.jsonl`: phase starts/results/errors, raw selected scanner status, process/cgroup samples and inventory oracles.
+- `summary.json`: measurements grouped by invocation, process identity and workload/verification/idle stage, sampled-current and kernel-reported cgroup peaks, first/last RSS, allocator availability, operation results and restart-window status.
+
+Raw artifacts may contain local paths, binary fingerprints and configuration. Keep them private; publish only a redacted conclusion and the measurements needed to understand it. There is no automatic GitHub write or cleanup step.
+
+Offline harness tests run without Docker, networking or credentials:
+
+```sh
+python3 scripts/test_scanner_memory_acceptance.py
+python3 -m py_compile scripts/scanner_memory_acceptance.py scripts/test_scanner_memory_acceptance.py
+```
+
+The structural characterization in `crates/scanner/src/scanner_folder/tests/checkpoint_fixture/memory_budget.rs` complements these measurements. Its `SCANNER_MEMORY_BUDGET` rows describe cache entries/capacities, a heap lower bound and wire size after a real small filesystem walk. The seeded coverage-gap case is synthetic and retains a frozen frontier; it is not a real fault injection, process RSS measurement, or 2M-object reproduction. Use it to explain retained structure, then use this harness to measure the running service.

scripts/README.md

@@ -58,6 +58,8 @@ their issue closes.
 | `run.ps1` | dev-tool | Windows counterpart of `run.sh` | — |
 | `probe.sh` | dev-tool | Probe-style e2e run | `make probe-e2e` |
 | `run_scanner_validation_harness.sh` | dev-tool | Scanner validation harness | `docs/operations/scanner-benchmark-runbook.md` |
+| `scanner_memory_acceptance.py` | dev-tool | Owned S3 scanner-memory and scanner-off burst acceptance phases | [Memory acceptance](../docs/testing/scanner-memory-acceptance.md) |
+| `test_scanner_memory_acceptance.py` | dev-tool | Dataset, ownership, telemetry and acceptance-oracle tests | Python unittest; same guide |
 | `check_test_wiring.py` | dev-tool | Validates test wiring and assembles one or more measured Scanner/Heal release descriptors into a single release evidence bundle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --self-test` |
 | `run_scanner_heal_evidence_case.sh` | dev-tool | Runs one Scanner/Heal release-evidence registry case and checks the produced receipt/oracle | `.config/scanner-heal-required-tests.json`; `check_test_wiring.py --check-scanner-heal` |
 | `run_scanner_heal_authority_evidence.py` | dev-tool | Assembles measured Scanner/Heal G01 root/quota authority release descriptors from operator-collected authority artifacts | `.config/scanner-heal-required-tests.json`; `test_scanner_heal_authority_evidence.sh` |
@@ -100,7 +102,7 @@ their issue closes.
 | Entry | Status | Purpose | Wiring / docs |
 |---|---|---|---|
 | `run_hotpath_warp_ab.sh` | dev-tool | Linux warp A/B rig for the hotpath series (quick local A/B) | `docs/operations/hotpath-warp-ab-runbook.md` |
-| `run_hotpath_warp_abba.sh` | ci-gate | Formal ABBA warp runner (baseline/candidate interleaved, `--rounds >= 3`); the runner performance-ab.yml executes | performance-ab.yml (scheduled); `docs/operations/hotpath-warp-ab-runbook.md` |
+| `run_hotpath_warp_abba.sh` | dev-tool | Formal ABBA warp runner (baseline/candidate interleaved, `--rounds >= 3`) | `docs/operations/hotpath-warp-ab-runbook.md` |
 | `hotpath_warp_ab_gate.sh` | dev-tool | Relative-budget gate evaluated over the warp A/B results | used by `run_hotpath_warp_ab.sh`; hotpath runbook |
 | `run_internode_grpc_ab_bench.sh` | dev-tool | One-click A/B driver for the internode gRPC optimization stages | `docs/operations/internode-grpc-benchmark-runbook.md` |
 | `run_internode_transport_baseline.sh` | dev-tool | Internode transport baseline runner | internode runbook; `crates/io-metrics/README.md` |

scripts/check_test_wiring.py

@@ -520,7 +520,6 @@ SCHEDULED_ALERT_CHECKOUT_OVERRIDES = {
     ".github/workflows/ci.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
     ".github/workflows/fuzz.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
     ".github/workflows/nightly-gnu.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
-    ".github/workflows/performance-ab.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
     ".github/workflows/runner-hygiene.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
     ".github/workflows/scheduled-validation-watchdog.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
     ".github/workflows/scheduled-validation-freshness.yml": "actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a",
@@ -5953,7 +5952,6 @@ class SelfTests(unittest.TestCase):
             ".github/workflows/ci.yml",
             ".github/workflows/fuzz.yml",
             ".github/workflows/nightly-gnu.yml",
-            ".github/workflows/performance-ab.yml",
             ".github/workflows/runner-hygiene.yml",
         )
         with tempfile.TemporaryDirectory() as tmp:

scripts/fixtures/btf_prototypes.c

@@ -0,0 +1,16 @@
+// Copyright 2026 RustFS Team
+// SPDX-License-Identifier: Apache-2.0
+// Compiled for debug declarations only. This object is never loaded as BPF.
+typedef void rustfs_void;
+
+__attribute__((used, noinline))
+unsigned long rustfs_btf_probe(void *data, const void *other, unsigned long length)
+{
+    return data == other ? length : 0;
+}
+
+__attribute__((used, noinline))
+rustfs_void rustfs_btf_void(const void *data)
+{
+    __asm__ volatile("" : : "r"(data) : "memory");
+}

scripts/inspect_btf_prototypes.py

@@ -0,0 +1,179 @@
+#!/usr/bin/env python3
+# Copyright 2026 RustFS Team
+# SPDX-License-Identifier: Apache-2.0
+"""Bounded BTF v1 reader; no inferred offsets, pointer values or runtime semantics."""
+import argparse
+import array
+import hashlib
+import json
+import os
+import stat
+import struct
+import sys
+from pathlib import Path
+
+MAX_BTF_BYTES = 32 * 1024 * 1024
+MAX_FUNCTIONS = 16
+MAX_JSON_BYTES = 1024 * 1024
+
+
+def need(x, label):
+    if not x:
+        raise ValueError(label)
+
+
+class _BTF:
+    def __init__(self, raw):
+        need(isinstance(raw, bytes) and 24 <= len(raw) <= 32 << 20, 'bounded_complete_BTF_blob')
+        magic, version, flags, header, toff, tlen, soff, slen = struct.unpack_from('<HBBIIIII', raw)
+        need((magic, version, flags) == (0xeb9f, 1, 0) and 24 <= header <= len(raw), 'supported_actual_BTF_v1_LE_header')
+        self.raw, self.tb, self.te = raw, header + toff, header + toff + tlen
+        self.sb, self.se = header + soff, header + soff + slen
+        need(header <= self.tb <= self.te <= self.sb < self.se <= len(raw) and self.se == len(raw), 'complete_nonoverlapping_declared_sections')
+        need(raw[self.sb] == 0 and raw[self.se - 1] == 0, 'valid_BTF_string_table_sentinels')
+        self.offsets = array.array('I', [0])
+        offset = self.tb
+        while offset < self.te:
+            need(offset + 12 <= self.te, 'complete_common_type')
+            _, info, _ = struct.unpack_from('<III', raw, offset)
+            kind, count = (info >> 24) & 31, info & 65535
+            need(info & 0x60ff0000 == 0, 'reserved_type_info_bits_zero')
+            fixed = {1: 4, 2: 0, 3: 12, 7: 0, 8: 0, 9: 0, 10: 0, 11: 0, 12: 0, 14: 4, 16: 0, 17: 4, 18: 0}
+            if kind in fixed:
+                size = fixed[kind]
+            elif kind in (4, 5, 15, 19):
+                size = count * 12
+            elif kind in (6, 13):
+                size = count * 8
+            else:
+                raise ValueError('unknown_BTF_kind_not_guessed')
+            need(offset + 12 + size <= self.te and len(self.offsets) <= 500000, 'complete_bounded_type_payload')
+            self.offsets.append(offset)
+            offset += 12 + size
+        need(offset == self.te, 'all_type_bytes_consumed')
+
+    def string(self, at):
+        need(type(at) is int and 0 <= at < self.se - self.sb, 'string_offset_in_actual_table')
+        end = self.raw.find(b'\0', self.sb + at, self.se)
+        need(end >= 0 and end - self.sb - at <= 4096, 'bounded_terminated_name')
+        return self.raw[self.sb + at:end].decode('utf-8', errors='strict')
+
+    def type(self, ident):
+        need(type(ident) is int and 0 < ident < len(self.offsets), 'actual_nonvoid_type_ID')
+        offset = self.offsets[ident]
+        name, info, size = struct.unpack_from('<III', self.raw, offset)
+        kind, count = (info >> 24) & 31, info & 65535
+        return {'ID': ident, 'offset': offset, 'name': self.string(name), 'kind': kind, 'count': count,
+                'kflag': bool(info >> 31), 'size_or_type': size}
+
+    def find(self, name, kind):
+        result = []
+        encoded = name.encode() + b'\0'
+        for i in range(1, len(self.offsets)):
+            off = self.offsets[i]
+            nm, info, _ = struct.unpack_from('<III', self.raw, off)
+            if (info >> 24) & 31 == kind:
+                need(nm < self.se - self.sb, 'name_offset_in_string_section')
+                if self.raw.startswith(encoded, self.sb + nm):
+                    result.append(self.type(i))
+        need(len(result) == 1, 'unique_actual_named_type_or_function')
+        return result[0]
+
+    def unwrap(self, ident, *, allow_void=False):
+        chain, seen = [], set()
+        while True:
+            # Void has no record. Only explicit prototype boundaries opt in.
+            if allow_void and type(ident) is int and ident == 0:
+                return {'ID': 0, 'name': 'void', 'kind': 0}, chain
+            need(ident not in seen and len(chain) < 32, 'acyclic_bounded_qualifier_chain')
+            seen.add(ident)
+            t = self.type(ident)
+            chain.append(t)
+            if t['kind'] == 2:
+                name = struct.unpack_from('<I', self.raw, t['offset'])[0]
+                need(name == 0 and t['count'] == 0 and not t['kflag'], 'actual_pointer_header')
+                if t['size_or_type'] != 0:
+                    self.type(t['size_or_type'])
+            if t['kind'] not in (8, 9, 10, 11, 18):
+                return t, chain
+            need(t['count'] == 0 and not t['kflag'], 'actual_qualifier_header')
+            if t['kind'] in (9, 10, 11):
+                name = struct.unpack_from('<I', self.raw, t['offset'])[0]
+                need(name == 0, 'unnamed_CVR_qualifier')
+            ident = t['size_or_type']
+
+    def prototype(self, name):
+        fn = self.find(name, 12)
+        p = self.type(fn['size_or_type'])
+        need(p['kind'] == 13 and p['count'] <= 16, 'actual_bounded_function_prototype')
+        args = []
+        for i in range(p['count']):
+            nm, ty = struct.unpack_from('<II', self.raw, p['offset'] + 12 + i * 8)
+            t, chain = self.unwrap(ty)
+            arg = {'name': self.string(nm), 'type_ID': ty, 'type': t, 'chain': chain}
+            if t['kind'] == 2:
+                target, target_chain = self.unwrap(t['size_or_type'], allow_void=True)
+                arg.update({'pointee': target, 'pointee_chain': target_chain})
+            args.append(arg)
+        result, chain = self.unwrap(p['size_or_type'], allow_void=True)
+        return {'function': fn, 'prototype': p, 'args': args, 'return': result, 'return_chain': chain}
+
+
+def inspect_prototypes(path, names):
+    """Read bounded source bytes and report declared prototypes, never runtime ABI."""
+    if (not isinstance(names, (list, tuple)) or not names or len(names) > MAX_FUNCTIONS
+            or any(not isinstance(name, str) or not name or "\0" in name
+                   or len(name.encode("utf-8")) > 4096 for name in names)
+            or len(set(names)) != len(names)):
+        raise ValueError("one_to_sixteen_unique_function_names_required")
+    path = Path(path)
+    if not stat.S_ISREG(path.lstat().st_mode):
+        raise ValueError("BTF_input_must_be_a_regular_non_symlink_file")
+
+    def opener(name, flags):
+        return os.open(name, flags | getattr(os, "O_NONBLOCK", 0) | getattr(os, "O_NOFOLLOW", 0))
+
+    with open(path, "rb", opener=opener) as source:
+        if not stat.S_ISREG(os.fstat(source.fileno()).st_mode):
+            raise ValueError("BTF_input_must_be_a_regular_non_symlink_file")
+        raw = source.read(MAX_BTF_BYTES + 1)
+    if len(raw) > MAX_BTF_BYTES:
+        raise ValueError("BTF_source_exceeds_32_MiB")
+    btf = _BTF(raw)
+    return {"schema": 1,
+            "source": {"format": "BTF-v1-little-endian", "bytes": len(raw),
+                       "sha256": hashlib.sha256(raw).hexdigest()},
+            "prototypes": [btf.prototype(name) for name in names]}
+
+
+def encode_document(document):
+    """Reject an oversized report before emitting any JSON prefix."""
+    pieces, length = [], 0
+    for chunk in json.JSONEncoder(ensure_ascii=True, sort_keys=True).iterencode(document):
+        encoded = chunk.encode("utf-8")
+        length += len(encoded)
+        if length + 1 > MAX_JSON_BYTES:
+            raise ValueError("prototype_report_exceeds_1_MiB")
+        pieces.append(encoded)
+    return b"".join(pieces) + b"\n"
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(
+        description="Read-only inspection of declared BTF function prototypes; no probe loading or runtime ABI claims.")
+    parser.add_argument("--btf", required=True, type=Path, help="raw BTF v1 little-endian file (at most 32 MiB)")
+    parser.add_argument("--function", required=True, action="append", dest="functions",
+                        help="exact, unique function name; repeat at most 16 times")
+    args = parser.parse_args(argv)
+    try:
+        output = encode_document(inspect_prototypes(args.btf, args.functions))
+    except OSError:
+        parser.exit(2, "BTF input could not be read\n")
+    except (ValueError, UnicodeError) as error:
+        parser.exit(2, "BTF inspection rejected: " + str(error) + "\n")
+    sys.stdout.buffer.write(output)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/scanner_memory_acceptance.py

@@ -0,0 +1,954 @@
+#!/usr/bin/env python3
+"""Run owned S3 scanner-memory acceptance phases; service lifecycle stays external."""
+
+from __future__ import annotations
+
+import argparse
+from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
+from datetime import datetime, timezone
+import fcntl
+import hashlib
+import ipaddress
+import json
+import math
+import os
+from pathlib import Path
+import re
+import shutil
+import stat
+import subprocess
+import sys
+import threading
+import time
+from urllib.parse import urlsplit
+import urllib.request
+import uuid
+
+SCHEMA = 1
+OWNER_KEY = ".scanner-memory-owner.json"
+LABEL = "rustfs.scanner-memory.run-id"
+GIB = 1024 ** 3
+SAFE_ENV = frozenset({
+    "RUSTFS_SCANNER_ENABLED", "RUSTFS_HEAL_ENABLED", "RUSTFS_SCANNER_SPEED",
+    "RUSTFS_SCANNER_START_DELAY_SECS", "RUSTFS_SCANNER_CYCLE", "RUSTFS_SCANNER_IDLE_MODE",
+    "RUSTFS_SCANNER_CYCLE_MAX_OBJECTS", "RUSTFS_SCANNER_CYCLE_MAX_DURATION_SECS",
+    "RUSTFS_SCANNER_CYCLE_MAX_DIRECTORIES", "RUSTFS_SCANNER_DISK_IO_BUDGET_IOPS",
+    "RUSTFS_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC",
+    "RUSTFS_CONSOLE_ENABLE", "RUSTFS_ERASURE_SET_DRIVE_COUNT", "RUSTFS_DURABILITY_MODE",
+    "RUSTFS_MEMORY_OBSERVABILITY_INTERVAL_SECS", "RUSTFS_OBS_METRICS_EXPORT_ENABLED",
+})
+STATUS_FIELDS = (
+    "current_cycle", "current_cycle_active", "current_cycle_objects_scanned",
+    "current_cycle_directories_scanned", "last_cycle_result", "last_cycle_partial_reason",
+    "last_cycle_end_unix_secs", "last_cycle_objects_scanned", "failed_cycles",
+    "current_disk_bucket_scans_active", "last_cycle_bucket_drive_failures", "usage_publication",
+)
+
+
+class AcceptanceError(ValueError):
+    """A public-safe diagnostic created by this runner, never by an SDK."""
+
+
+class NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, *_args, **_kwargs):
+        raise AcceptanceError("redirected observation endpoints are not supported")
+
+
+def require(condition, message):
+    if not condition:
+        raise AcceptanceError(message)
+
+
+def canonical(value):
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def digest(value):
+    return hashlib.sha256(canonical(value)).hexdigest()
+
+
+def utc():
+    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def local_url(value):
+    parsed = urlsplit(value)
+    require(parsed.scheme in ("http", "https") and parsed.hostname in ("127.0.0.1", "::1")
+            and parsed.port is not None and not parsed.username and not parsed.password
+            and not parsed.query and not parsed.fragment and parsed.path in ("", "/"),
+            "endpoint must use literal 127.0.0.1 or ::1 with an explicit HTTP(S) port")
+    return value.rstrip("/")
+
+
+@contextmanager
+def open_owned(path, mode="r"):
+    require(mode in ("r", "a"), "unsupported evidence file mode")
+    flags = os.O_RDONLY if mode == "r" else os.O_WRONLY | os.O_CREAT | os.O_APPEND
+    try:
+        fd = os.open(path, flags | os.O_NOFOLLOW | os.O_NONBLOCK, 0o600)
+    except OSError:
+        raise AcceptanceError("evidence file is not an owned regular file") from None
+    try:
+        info = os.fstat(fd)
+        require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() and info.st_nlink == 1
+                and info.st_mode & 0o077 == 0, "evidence file must be private, owned and single-link")
+        if mode == "a":
+            os.fchmod(fd, 0o600)
+        stream = os.fdopen(fd, mode, encoding="utf-8")
+        fd = None
+        with stream:
+            yield stream
+    finally:
+        if fd is not None:
+            os.close(fd)
+
+
+def read_json(path, limit=1024 * 1024):
+    with open_owned(path) as stream:
+        text = stream.read(limit + 1)
+    require(len(text) <= limit, "oversized evidence file")
+    return json.loads(text)
+
+
+def atomic_json(path, value):
+    temporary = path.with_name(path.name + "." + uuid.uuid4().hex + ".tmp")
+    with temporary.open("x", encoding="utf-8") as stream:
+        os.chmod(temporary, 0o600)
+        json.dump(value, stream, sort_keys=True)
+        stream.write("\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+    os.replace(temporary, path)
+
+
+def object_key(plan, index):
+    require(type(index) is int and 0 <= index < plan["objects"], "object index outside manifest")
+    prefix = index % plan["root_prefixes"]
+    ordinal = index // plan["root_prefixes"]
+    suffix = f"source:{prefix:06d}/{ordinal:08d}.webp"
+    if plan["layout"] == "wide":
+        return suffix
+    return f"photos/{prefix % 256:02x}/{(prefix // 256) % 256:02x}/" + suffix
+
+
+def key_index(plan, key):
+    pattern = r"source:([0-9]{6,})/([0-9]{8,})\.webp"
+    if plan["layout"] == "nested":
+        pattern = r"photos/[0-9a-f]{2}/[0-9a-f]{2}/" + pattern
+    match = re.fullmatch(pattern, key)
+    require(match is not None, "inventory contains a foreign object key")
+    prefix, ordinal = map(int, match.groups())
+    index = ordinal * plan["root_prefixes"] + prefix
+    require(prefix < plan["root_prefixes"] and index < plan["objects"] and object_key(plan, index) == key,
+            "inventory key does not match the immutable layout")
+    return index
+
+
+def payload(plan):
+    seed = uuid.UUID(plan["run_id"]).bytes
+    blocks = (hashlib.sha256(seed + index.to_bytes(8, "big")).digest()
+              for index in range((plan["object_bytes"] + 31) // 32))
+    return b"".join(blocks)[:plan["object_bytes"]]
+
+
+def owner_body(manifest):
+    return canonical({"schema": SCHEMA, "run_id": manifest["plan"]["run_id"], "plan_sha256": manifest["plan_sha256"]})
+
+
+def validate_manifest(manifest):
+    require(type(manifest) is dict and set(manifest) == {"schema", "plan", "plan_sha256"}
+            and manifest["schema"] == SCHEMA, "invalid manifest envelope")
+    plan = manifest["plan"]
+    require(type(plan) is dict and set(plan) == {
+        "run_id", "created_at", "endpoint", "source_sha", "binary_sha256", "layout", "objects",
+        "object_bytes", "root_prefixes", "memory_bytes", "data_root", "data_volume", "bucket",
+    }, "invalid immutable plan fields")
+    require(digest(plan) == manifest["plan_sha256"], "manifest digest mismatch")
+    require(str(uuid.UUID(plan["run_id"])) == plan["run_id"], "invalid run identity")
+    require(plan["bucket"] == "scanner-memory-" + uuid.UUID(plan["run_id"]).hex, "bucket is not run-owned")
+    local_url(plan["endpoint"])
+    require(re.fullmatch(r"[0-9a-f]{40}", plan["source_sha"]) and re.fullmatch(r"[0-9a-f]{64}", plan["binary_sha256"]),
+            "invalid source or binary identity")
+    require(plan["layout"] in ("wide", "nested"), "invalid layout")
+    for name, maximum in (("objects", 20_000_000), ("object_bytes", 1024 * 1024),
+                          ("root_prefixes", 2_000_000), ("memory_bytes", 64 * GIB)):
+        require(type(plan[name]) is int and 0 < plan[name] <= maximum, "invalid workload bound")
+    require(plan["root_prefixes"] <= plan["objects"], "root prefix count exceeds object count")
+    require(bool(plan["data_root"]) != bool(plan["data_volume"]), "select one immutable storage source")
+    if plan["data_volume"]:
+        require(isinstance(plan["data_volume"], str) and re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}", plan["data_volume"]),
+                "invalid owned data volume name")
+    else:
+        path = Path(plan["data_root"])
+        require(path.is_absolute() and path == path.resolve() and path.is_dir(), "data root must be an existing canonical directory")
+    return plan
+
+
+class Evidence:
+    def __init__(self, root):
+        self.root = root.resolve()
+        require(root == self.root and root.is_dir() and not root.is_symlink(), "run directory must be canonical")
+        require(root.stat().st_uid == os.getuid() and root.stat().st_mode & 0o077 == 0, "run directory must be private and owned")
+        self.manifest = read_json(root / "manifest.json")
+        self.plan = validate_manifest(self.manifest)
+        self.lock = threading.Lock()
+
+    @contextmanager
+    def exclusive(self):
+        fd = os.open(self.root / ".lock", os.O_CREAT | os.O_RDWR | os.O_NOFOLLOW, 0o600)
+        with os.fdopen(fd, "w") as stream:
+            info = os.fstat(stream.fileno())
+            require(stat.S_ISREG(info.st_mode) and info.st_uid == os.getuid() and info.st_nlink == 1
+                    and info.st_mode & 0o077 == 0, "invalid evidence lock ownership")
+            fcntl.flock(stream, fcntl.LOCK_EX | fcntl.LOCK_NB)
+            yield
+
+    def append(self, kind, **fields):
+        row = {"schema": SCHEMA, "at": utc(), "monotonic_ns": time.monotonic_ns(), "kind": kind,
+               "run_id": self.plan["run_id"], **fields}
+        with self.lock, open_owned(self.root / "observations.jsonl", "a") as stream:
+            stream.write(json.dumps(row, sort_keys=True) + "\n")
+            stream.flush()
+        return row
+
+    def state(self):
+        path = self.root / "state.json"
+        state = read_json(path) if path.exists() or path.is_symlink() else {"plan_sha256": self.manifest["plan_sha256"], "next_object": 0}
+        require(state.get("plan_sha256") == self.manifest["plan_sha256"], "resume state belongs to another plan")
+        require(type(state.get("next_object")) is int and 0 <= state["next_object"] <= self.plan["objects"], "invalid resume marker")
+        return state
+
+    def save(self, state):
+        atomic_json(self.root / "state.json", state)
+
+
+class S3:
+    def __init__(self, evidence, workers):
+        import boto3
+        from botocore.config import Config
+        self.evidence = evidence
+        self.plan = evidence.plan
+        self.session = boto3.Session()
+        require(os.environ.get("AWS_ACCESS_KEY_ID") and os.environ.get("AWS_SECRET_ACCESS_KEY"),
+                "set disposable AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY in the environment")
+        self.client = self.session.client("s3", endpoint_url=self.plan["endpoint"], region_name="us-east-1", config=Config(
+            signature_version="s3v4", s3={"addressing_style": "path"}, proxies={}, connect_timeout=5, read_timeout=30,
+            retries={"total_max_attempts": 2}, max_pool_connections=workers))
+
+    def admin(self, path):
+        from botocore.auth import S3SigV4Auth
+        from botocore.awsrequest import AWSRequest
+        request = AWSRequest(method="GET", url=self.plan["endpoint"] + path)
+        S3SigV4Auth(self.session.get_credentials().get_frozen_credentials(), "s3", "us-east-1").add_auth(request)
+        opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+        with opener.open(urllib.request.Request(request.url, headers=dict(request.headers)), timeout=15) as response:
+            raw = response.read(4 * 1024 * 1024 + 1)
+        require(len(raw) <= 4 * 1024 * 1024, "admin response exceeds bound")
+        return json.loads(raw)
+
+    def storage_paths(self):
+        # Ask the serving process for its expanded, effective disk inventory;
+        # Docker Config.Env and /proc/environ alone do not describe CLI overrides.
+        info = self.admin("/rustfs/admin/v3/storageinfo").get("info", {})
+        observations = info.get("observations", [])
+        require(len(observations) == 1 and observations[0].get("status") == "succeeded"
+                and observations[0].get("cached") is False,
+                "storage inventory must be a fresh single-node observation")
+        disks = info.get("disks", [])
+        backend = info.get("backend", {})
+        drives, sets = backend.get("DrivesPerSet", []), backend.get("TotalSets", [])
+        require(drives and len(drives) == len(sets)
+                and all(type(n) is int and n > 0 for n in drives + sets)
+                and sum(n * count for n, count in zip(drives, sets)) == len(disks),
+                "storage inventory does not cover the configured topology")
+        require(all(disk.get("local") is True and disk.get("state") == "ok" for disk in disks),
+                "storage inventory must contain only healthy local disks")
+        return [disk.get("path") for disk in disks]
+
+    def status(self, enabled):
+        status = self.admin("/rustfs/admin/v3/scanner/status")
+        require(status.get("enabled") is enabled, "scanner enabled state differs from required phase")
+        metrics = status.get("metrics")
+        require(type(metrics) is dict, "scanner metrics are unavailable")
+        return {"enabled": enabled, "metrics": {key: metrics.get(key) for key in STATUS_FIELDS}}
+
+    def owned(self, create=False):
+        from botocore.exceptions import ClientError
+        bucket = self.plan["bucket"]
+        try:
+            self.client.head_bucket(Bucket=bucket)
+        except ClientError as error:
+            require(error.response.get("ResponseMetadata", {}).get("HTTPStatusCode") == 404 and create,
+                    "owned bucket is unavailable")
+            self.client.create_bucket(Bucket=bucket)
+            self.client.put_object(Bucket=bucket, Key=OWNER_KEY, Body=owner_body(self.evidence.manifest),
+                                   ContentType="application/json")
+        response = self.client.get_object(Bucket=bucket, Key=OWNER_KEY)
+        with response["Body"] as body:
+            marker = body.read(4097)
+        require(marker == owner_body(self.evidence.manifest), "bucket ownership marker mismatch")
+        require(self.client.get_bucket_versioning(Bucket=bucket).get("Status") is None,
+                "acceptance bucket must remain unversioned")
+
+    def verify(self, samples=16, deadline=None, guard=None):
+        deadline = deadline or time.monotonic() + 3600
+        def check():
+            require(time.monotonic() <= deadline, "inventory verification time budget exhausted")
+            if guard:
+                guard()
+        check()
+        self.owned()
+        plan = self.plan
+        seen = bytearray((plan["objects"] + 7) // 8)
+        count = size = markers = 0
+        token = None
+        while True:
+            check()
+            args = {"Bucket": plan["bucket"], "MaxKeys": 1000}
+            if token:
+                args["ContinuationToken"] = token
+            page = self.client.list_objects_v2(**args)
+            for entry in page.get("Contents", []):
+                if entry["Key"] == OWNER_KEY:
+                    markers += 1
+                    require(entry["Size"] == len(owner_body(self.evidence.manifest)), "ownership object size changed")
+                    continue
+                index = key_index(plan, entry["Key"])
+                mask = 1 << (index % 8)
+                require(not seen[index // 8] & mask, "listing contains a duplicate object")
+                seen[index // 8] |= mask
+                require(entry["Size"] == plan["object_bytes"], "object size mismatch")
+                count += 1
+                size += entry["Size"]
+            if not page.get("IsTruncated"):
+                break
+            next_token = page.get("NextContinuationToken")
+            require(isinstance(next_token, str) and next_token and next_token != token, "listing cursor did not advance")
+            token = next_token
+        require(count == plan["objects"] and markers == 1 and size == count * plan["object_bytes"],
+                "exact object/byte inventory does not match manifest")
+        expected = payload(plan)
+        indices = sorted({index * (plan["objects"] - 1) // max(1, samples - 1) for index in range(samples)})
+        for index in indices:
+            check()
+            response = self.client.get_object(Bucket=plan["bucket"], Key=object_key(plan, index))
+            with response["Body"] as body:
+                actual = body.read(plan["object_bytes"] + 1)
+            require(actual == expected, "sampled payload differs from fixture")
+        result = {"objects": count, "logical_bytes": size, "ownership_objects": markers,
+                  "scanner_expected_objects": count + 1,
+                  "scanner_expected_bytes": size + len(owner_body(self.evidence.manifest)), "payload_samples": len(indices)}
+        check()
+        self.evidence.append("inventory", **result)
+        return result
+
+    def usage(self, expected):
+        account = self.admin("/rustfs/admin/v3/accountinfo")
+        matches = [bucket for bucket in account.get("buckets", []) if bucket.get("name") == self.plan["bucket"]]
+        require(len(matches) == 1, "bucket usage is unavailable")
+        bucket = matches[0]
+        return {"objects": bucket.get("objects"), "size": bucket.get("size"),
+                "converged": account.get("usage_snapshot_converged"),
+                "matches_inventory": bucket.get("objects") == expected["scanner_expected_objects"]
+                and bucket.get("size") == expected["scanner_expected_bytes"]}
+
+
+def command(argv, timeout=30):
+    result = subprocess.run(argv, capture_output=True, timeout=timeout, check=False)
+    require(result.returncode == 0 and len(result.stdout) <= 1024 * 1024, "runtime observation command failed")
+    return result.stdout
+
+
+def key_values(raw, colon=False):
+    result = {}
+    for line in raw.splitlines():
+        pieces = line.split(":" if colon else None, 1)
+        if len(pieces) != 2:
+            continue
+        value = pieces[1].strip().split()
+        if value and value[0].isdigit():
+            result[pieces[0]] = int(value[0]) * (1024 if len(value) > 1 and value[1] == "kB" else 1)
+    return result
+
+
+def parse_proc_stat(raw):
+    end = raw.rfind(")")
+    require(end > 0, "invalid process identity")
+    fields = raw[end + 1:].split()
+    require(len(fields) >= 20 and fields[19].isdigit(), "invalid process start time")
+    return int(fields[19])
+
+
+def address_matches(bound, endpoint):
+    try:
+        actual, expected = ipaddress.ip_address(bound), ipaddress.ip_address(urlsplit(endpoint).hostname)
+    except (ValueError, TypeError):
+        return False
+    return actual.version == expected.version and (actual == expected or actual.is_unspecified)
+
+
+def docker_endpoint_matches(bindings, endpoint):
+    port = str(urlsplit(endpoint).port)
+    return any(protocol.endswith("/tcp") and binding.get("HostPort") == port
+               and address_matches(binding.get("HostIp"), endpoint)
+               for protocol, entries in bindings.items() for binding in (entries or []))
+
+
+def native_endpoint_matches(tables, owned_sockets, endpoint):
+    port = urlsplit(endpoint).port
+    found = False
+    for protocol, table in tables.items():
+        for row in table.splitlines()[1:]:
+            fields = row.split()
+            if len(fields) <= 9 or fields[3] != "0A":
+                continue
+            address, encoded_port = fields[1].rsplit(":", 1)
+            if int(encoded_port, 16) != port:
+                continue
+            width = 8 if protocol == "tcp" else 32
+            require(len(address) == width, "invalid native listener address")
+            packed = b"".join(int(address[index:index + 8], 16).to_bytes(4, sys.byteorder)
+                              for index in range(0, width, 8))
+            bound = ipaddress.ip_address(packed)
+            expected = ipaddress.ip_address(urlsplit(endpoint).hostname)
+            ambiguous_v4 = bound.version == 6 and expected.version == 4 and (bound.is_unspecified or bound.ipv4_mapped == expected)
+            if ambiguous_v4 and fields[9] not in owned_sockets:
+                return False
+            if not address_matches(str(bound), endpoint):
+                continue
+            if fields[9] not in owned_sockets:
+                return False
+            found = True
+    return found
+
+
+def storage_directory(value):
+    require(type(value) is str and value.startswith("/") and str(Path(value)) == value
+            and ".." not in Path(value).parts and not any(char in value for char in "\0\n\r\t"),
+            "storage directory must be an unambiguous absolute path")
+    return Path(value)
+
+
+def owned_storage_paths(owned, paths):
+    owned = storage_directory(owned)
+    require(type(paths) is list and 0 < len(paths) <= 1024, "storage disk paths are unavailable")
+    paths = [storage_directory(path) for path in paths]
+    require(all(path == owned or owned in path.parents for path in paths),
+            "effective storage disk is outside the owned storage root")
+    return sorted(paths)
+
+
+def check_storage_mounts(owned, mountinfo):
+    lines = mountinfo.decode().splitlines()
+    require(lines, "storage mount inventory is unavailable")
+    for line in lines:
+        fields = line.split()
+        require(len(fields) >= 10 and "-" in fields[6:], "storage mount inventory is malformed")
+        mount = storage_directory(re.sub(r"\\([0-7]{3})", lambda match: chr(int(match[1], 8)), fields[4]))
+        require(Path(owned) not in mount.parents, "nested mounts under owned storage are unsupported")
+
+
+def native_storage(process, owned, paths, mountinfo):
+    # Host path resolution is valid only in the serving process's namespace/root.
+    require((process / "ns/mnt").samefile("/proc/self/ns/mnt") and (process / "root").samefile("/"),
+            "native storage requires the observer's mount namespace and root")
+    owned = storage_directory(owned)
+    require(owned.resolve(strict=True) == owned, "owned storage root changed")
+    paths = sorted(str(storage_directory(path).resolve(strict=True)) for path in paths)
+    resolved = owned_storage_paths(str(owned), paths)
+    check_storage_mounts(owned, mountinfo)
+    require(owned.is_dir() and all(path.is_dir() and path.stat().st_dev == owned.stat().st_dev for path in resolved),
+            "effective storage disks do not share the owned filesystem")
+    binding = digest({"owned": str(owned), "paths": paths, "device": owned.stat().st_dev})
+    return f"native 0 0 {shutil.disk_usage(owned).free // 1024} 0 /".encode(), binding
+
+
+class Runtime:
+    def __init__(self, evidence, container=None, pid=None, metrics_url=None, storage_paths=None):
+        require(bool(container) != bool(pid), "select exactly one owned container or native PID")
+        self.evidence, self.plan = evidence, evidence.plan
+        self.container, self.pid = container, pid or 1
+        require(container is None or re.fullmatch(r"[a-zA-Z0-9][a-zA-Z0-9_.-]{0,127}", container), "invalid container identifier")
+        require(type(self.pid) is int and self.pid > 0, "invalid process PID")
+        self.metrics_url = metrics_url
+        if metrics_url:
+            parsed = urlsplit(metrics_url)
+            local_url(f"{parsed.scheme}://{parsed.netloc}")
+            require(not parsed.query and not parsed.fragment, "metrics URL must not contain query or fragment")
+        self.verified = set()
+        self.storage_paths = storage_paths
+
+    def read_files(self):
+        names = ("stat", "status", "smaps_rollup", "cgroup", "environ", "mountinfo")
+        if self.container:
+            raw = json.loads(command(["docker", "inspect", self.container]))
+            require(len(raw) == 1, "ambiguous container identity")
+            item = raw[0]
+            require((item["Config"].get("Labels") or {}).get(LABEL) == self.plan["run_id"], "container is not owned by this run")
+            require(item["State"]["Running"] and not item["State"]["OOMKilled"], "owned container stopped or was OOM-killed")
+            require(item["HostConfig"]["Memory"] == self.plan["memory_bytes"], "Docker memory limit differs from manifest")
+            bindings = item.get("NetworkSettings", {}).get("Ports", {})
+            require(docker_endpoint_matches(bindings, self.plan["endpoint"]),
+                    "S3 TCP endpoint address is not published by the observed container")
+            if self.plan["data_volume"]:
+                mounts = [mount for mount in item.get("Mounts", [])
+                          if mount.get("Type") == "volume" and mount.get("Name") == self.plan["data_volume"]]
+                volume = json.loads(command(["docker", "volume", "inspect", self.plan["data_volume"]]))
+                require(len(volume) == 1 and (volume[0].get("Labels") or {}).get(LABEL) == self.plan["run_id"],
+                        "data volume is not owned by this run")
+            else:
+                mounts = [mount for mount in item.get("Mounts", [])
+                          if mount.get("Type") == "bind" and Path(mount["Source"]).resolve() == Path(self.plan["data_root"])]
+            require(len(mounts) == 1 and mounts[0].get("RW") is True, "container storage does not match the immutable manifest")
+            storage_path = mounts[0]["Destination"]
+            require(callable(self.storage_paths), "effective storage inventory is required")
+            paths = [str(path) for path in owned_storage_paths(storage_path, self.storage_paths())]
+            identity = {"container_id": item["Id"], "started_at": item["State"]["StartedAt"],
+                        "restart_count": item["RestartCount"], "pid": self.pid}
+            script = '[ "$(stat -Lc "%d:%i" /proc/$1/ns/mnt)" = "$(stat -Lc "%d:%i" /proc/self/ns/mnt)" ] || exit 1; '
+            script += '[ "$(stat -Lc "%d:%i" /proc/$1/root)" = "$(stat -Lc "%d:%i" /)" ] || exit 1; '
+            script += 'for n in stat status smaps_rollup cgroup environ mountinfo; do printf "\\n__FILE_%s__\\n" "$n"; cat "/proc/$1/$n" || exit 1; done; '
+            script += 'printf "\\n__FILE_exe_inode__\\n"; stat -Lc "%d:%i" "/proc/$1/exe"; '
+            script += 'for n in memory.current memory.max memory.peak memory.stat memory.events memory.swap.current memory.swap.max; do printf "\\n__FILE_%s__\\n" "$n"; cat "/sys/fs/cgroup/$n" 2>/dev/null || true; done'
+            script += '; printf "\\n__FILE_storage__\\n"; df -Pk "$2" | tail -n 1'
+            script += '; printf "\\n__FILE_storage_root__\\n"; readlink -f -- "$2" || exit 1'
+            script += '; printf "\\n__FILE_storage_devices__\\n"; storage=$2; shift 2; stat -Lc "%d" -- "$storage" "$@" || exit 1'
+            script += '; printf "\\n__FILE_volume_paths__\\n"; for p; do [ -d "$p" ] || exit 1; readlink -f -- "$p" || exit 1; done'
+            data = command(["docker", "exec", item["Id"], "sh", "-c", script, "sh", str(self.pid), storage_path, *paths])
+            chunks = re.split(rb"\n__FILE_([A-Za-z_.]+)__\n", data)
+            files = {chunks[index].decode(): chunks[index + 1].strip(b"\n") for index in range(1, len(chunks), 2)}
+            require(files["storage_root"].decode() == storage_path, "owned storage mount is not canonical")
+            resolved = files["volume_paths"].decode().splitlines()
+            require(len(resolved) == len(paths), "effective storage paths could not be resolved")
+            owned_storage_paths(storage_path, resolved)
+            check_storage_mounts(storage_path, files["mountinfo"])
+            devices = files["storage_devices"].splitlines()
+            require(len(devices) == len(paths) + 1 and devices[0].isdigit() and len(set(devices)) == 1,
+                    "effective storage disks do not share the owned filesystem")
+            identity["storage_binding_sha256"] = digest({"mount": mounts[0], "paths": resolved, "devices": [value.decode() for value in devices]})
+        else:
+            require(self.plan["data_volume"] is None, "native PID sampling requires a host data root")
+            root = Path("/proc") / str(self.pid)
+            require(root.stat().st_uid == os.getuid(), "native process is not owned by this user")
+            sockets = set()
+            for descriptor in (root / "fd").iterdir():
+                try:
+                    target = os.readlink(descriptor)
+                except FileNotFoundError:
+                    continue
+                match = re.fullmatch(r"socket:\[([0-9]+)\]", target)
+                if match:
+                    sockets.add(match[1])
+            tables = {protocol: (root / "net" / protocol).read_text() for protocol in ("tcp", "tcp6")
+                      if (root / "net" / protocol).exists()}
+            require(native_endpoint_matches(tables, sockets, self.plan["endpoint"]),
+                    "S3 TCP endpoint is not exclusively bound by the observed PID")
+            files = {name: (root / name).read_bytes() for name in names}
+            groups = files["cgroup"].decode().splitlines()
+            require(len(groups) == 1 and groups[0].startswith("0::/") and groups[0] != "0::/", "dedicated cgroup v2 is required")
+            relative = groups[0][3:].lstrip("/")
+            require(".." not in Path(relative).parts, "invalid cgroup path")
+            group = Path("/sys/fs/cgroup") / relative
+            require(set((group / "cgroup.procs").read_text().split()) == {str(self.pid)}, "native cgroup contains other processes")
+            for name in ("memory.current", "memory.max", "memory.peak", "memory.stat", "memory.events", "memory.swap.current", "memory.swap.max"):
+                files[name] = (group / name).read_bytes() if (group / name).exists() else b""
+            info = (root / "exe").stat()
+            files["exe_inode"] = f"{info.st_dev}:{info.st_ino}".encode()
+            require(callable(self.storage_paths), "effective storage inventory is required")
+            files["storage"], binding = native_storage(root, self.plan["data_root"], self.storage_paths(), files["mountinfo"])
+            identity = {"boot_id": Path("/proc/sys/kernel/random/boot_id").read_text().strip(), "pid": self.pid,
+                        "storage_binding_sha256": binding}
+        identity.update(start_ticks=parse_proc_stat(files["stat"].decode()), executable_inode=files["exe_inode"].decode())
+        key = digest(identity)
+        if key not in self.verified:
+            if self.container:
+                observed = command(["docker", "exec", identity["container_id"], "sha256sum", f"/proc/{self.pid}/exe"], timeout=60).decode().split()[0]
+            else:
+                with (Path("/proc") / str(self.pid) / "exe").open("rb") as stream:
+                    observed = hashlib.file_digest(stream, "sha256").hexdigest()
+            require(observed == self.plan["binary_sha256"], "running binary does not match manifest")
+            self.verified.add(key)
+        environment = {}
+        for pair in files["environ"].split(b"\0"):
+            name, _, value = pair.partition(b"=")
+            if name.decode(errors="replace") in SAFE_ENV:
+                environment[name.decode()] = value.decode()
+        return identity, files, environment
+
+    def sample(self):
+        identity, files, environment = self.read_files()
+        def number(name):
+            value = files.get(name, b"").strip()
+            return int(value) if value.isdigit() else None
+        require(number("memory.max") == self.plan["memory_bytes"], "kernel memory limit differs from manifest")
+        status = key_values(files["status"].decode(), colon=True)
+        smaps = key_values(files["smaps_rollup"].decode(), colon=True)
+        events = key_values(files["memory.events"].decode())
+        require("oom" in events and "oom_kill" in events, "cgroup OOM counters are unavailable")
+        require(all(events.get(name, 0) == 0 for name in ("oom", "oom_kill", "oom_group_kill")), "cgroup reported an OOM event")
+        require(status.get("VmRSS") is not None and number("memory.current") is not None, "required memory readings are missing")
+        storage = files["storage"].split()
+        require(len(storage) >= 6 and storage[3].isdigit(), "storage free-space observation is unavailable")
+        result = {"storage_available_bytes": int(storage[3]) * 1024, "identity": identity, "identity_sha256": digest(identity), "configuration": environment,
+                  "rss_bytes": status["VmRSS"], "rss_anon_bytes": status.get("RssAnon"),
+                  "smaps_anonymous_bytes": smaps.get("Anonymous"), "cgroup_current_bytes": number("memory.current"),
+                  "cgroup_peak_bytes": number("memory.peak"), "cgroup_limit_bytes": number("memory.max"),
+                  "cgroup_swap_bytes": number("memory.swap.current"), "cgroup_swap_limit_bytes": number("memory.swap.max"),
+                  "cgroup_stat": key_values(files["memory.stat"].decode()), "cgroup_events": events,
+                  "active_heap_bytes": None, "active_heap_status": "unknown: allocator metrics not supplied",
+                  "allocator_endpoint_attribution": "operator-supplied; process binding unverified" if self.metrics_url else None}
+        if self.metrics_url:
+            try:
+                opener = urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())
+                with opener.open(self.metrics_url, timeout=5) as response:
+                    text = response.read(4 * 1024 * 1024 + 1).decode()
+                require(len(text) <= 4 * 1024 * 1024, "oversized metrics response")
+                values = {}
+                for name in ("rustfs_memory_allocator_malloc_requested_bytes_available", "rustfs_memory_allocator_malloc_requested_bytes"):
+                    matches = re.findall(r"^" + name + r"(?:\{[^\n]*\})?\s+([0-9.eE+-]+)\s*$", text, re.MULTILINE)
+                    if len(matches) == 1:
+                        values[name] = float(matches[0])
+                active = values.get("rustfs_memory_allocator_malloc_requested_bytes")
+                if values.get("rustfs_memory_allocator_malloc_requested_bytes_available") == 1 and active is not None and math.isfinite(active) and active >= 0:
+                    result.update(active_heap_bytes=int(active), active_heap_status="observed: allocator requested current bytes")
+                else:
+                    result["active_heap_status"] = "unknown: live allocator statistic unavailable"
+            except Exception:
+                result["active_heap_status"] = "unknown: allocator metrics request failed"
+        return result
+
+
+class Phase:
+    """One process identity and deadline cover work, verification and the last sample."""
+
+    def __init__(self, evidence, runtime, api, label, enabled, args):
+        self.evidence, self.runtime, self.api = evidence, runtime, api
+        self.label, self.enabled, self.interval = label, enabled, args.interval
+        self.phase_id = str(uuid.uuid4())
+        self.started = time.monotonic()
+        self.deadline = self.started + args.seconds
+        self.stop = threading.Event()
+        self.sample_lock = threading.Lock()
+        self.error = None
+        self.identity = None
+        self.latest = None
+        self.initial = None
+        self.stage = "scanner" if enabled else "workload"
+
+    def check(self):
+        require(self.error is None, self.error or "observation failed")
+        require(time.monotonic() <= self.deadline, "phase time budget exhausted; retain evidence and resume explicitly")
+
+    def once(self):
+        with self.sample_lock:
+            self.check()
+            memory = self.runtime.sample()
+            if self.identity is None:
+                self.identity = memory["identity_sha256"]
+            require(memory["identity_sha256"] == self.identity, "process changed during a phase; start a new phase after restart")
+            status = self.api.status(self.enabled)
+            # Close the API window before Phase entry can authorize S3 writes.
+            # Runtime identity also binds the effective storage inventory.
+            memory = self.runtime.sample()
+            require(memory["identity_sha256"] == self.identity,
+                    "process or effective storage changed during status observation")
+            self.check()
+            row = self.evidence.append("sample", phase=self.label, phase_id=self.phase_id, stage=self.stage, memory=memory, scanner=status)
+            self.latest = row
+            return row
+
+    def __enter__(self):
+        self.evidence.append("phase_start", phase=self.label, phase_id=self.phase_id,
+                             time_budget_seconds=self.deadline - self.started)
+        self.initial = self.once()
+        def sample_loop():
+            while not self.stop.wait(self.interval):
+                try:
+                    self.once()
+                except Exception:
+                    self.error = "memory/status observation failed; phase is inconclusive"
+                    self.evidence.append("observation_error", phase=self.label, phase_id=self.phase_id)
+                    self.stop.set()
+        self.thread = threading.Thread(target=sample_loop, daemon=True)
+        self.thread.start()
+        return self
+
+    def enter_stage(self, stage):
+        with self.sample_lock:
+            self.stage = stage
+        self.once()
+
+    def verify(self):
+        self.enter_stage("verification")
+        return self.api.verify(deadline=self.deadline, guard=self.check)
+
+    def result(self, result):
+        return {**result, "phase_id": self.phase_id, "process_identity": self.identity,
+                "elapsed_seconds": time.monotonic() - self.started}
+
+    def __exit__(self, exception_type, *_):
+        self.stop.set()
+        self.thread.join(timeout=60)
+        require(not self.thread.is_alive(), "sampler did not stop")
+        if exception_type is None:
+            self.check()
+            self.once()
+
+
+def free_space(plan, remaining, available=None):
+    required = GIB + math.ceil(remaining * (plan["object_bytes"] + 16384) * 1.2)
+    if available is None:
+        require(plan["data_root"] is not None, "named-volume capacity must be observed inside the owned container")
+        available = shutil.disk_usage(plan["data_root"]).free
+    require(type(available) is int and available >= required, "insufficient free space for remaining objects and metadata reserve")
+
+
+def run_load(evidence, api, runtime, args):
+    state = evidence.state()
+    data = payload(evidence.plan)
+    with Phase(evidence, runtime, api, "load", False, args) as phase, ThreadPoolExecutor(args.workers) as pool:
+        free_space(evidence.plan, evidence.plan["objects"] - state["next_object"], phase.latest["memory"]["storage_available_bytes"])
+        api.owned(create=state["next_object"] == 0)
+        while state["next_object"] < evidence.plan["objects"]:
+            phase.check()
+            start = state["next_object"]
+            end = min(start + args.batch, evidence.plan["objects"])
+            free_space(evidence.plan, evidence.plan["objects"] - start, phase.latest["memory"]["storage_available_bytes"])
+            api.owned()
+            def put(index):
+                phase.check()
+                api.client.put_object(Bucket=evidence.plan["bucket"], Key=object_key(evidence.plan, index), Body=data)
+            list(pool.map(put, range(start, end)))
+            state["next_object"] = end
+            evidence.save(state)
+            evidence.append("load_batch", phase_id=phase.phase_id, completed=end, expected=evidence.plan["objects"])
+            phase.check()
+        inventory = phase.verify()
+    state["load_complete"] = True
+    evidence.save(state)
+    return phase.result({"phase": "load", "complete": True, "inventory": inventory})
+
+
+def run_probe(evidence, api, runtime, args):
+    require(evidence.state().get("load_complete") is True, "finish and verify load before probing")
+    completed = set()
+    with Phase(evidence, runtime, api, args.label, True, args) as phase:
+        api.owned()
+        first_end = phase.initial["scanner"]["metrics"].get("last_cycle_end_unix_secs") or 0
+        while True:
+            phase.check()
+            metrics = phase.latest["scanner"]["metrics"]
+            ended = metrics.get("last_cycle_end_unix_secs")
+            if type(ended) is int and ended > first_end and metrics.get("last_cycle_result") == "success":
+                completed.add(ended)
+            if args.stop_after_progress is not None:
+                progress = metrics.get("current_cycle_objects_scanned")
+                if metrics.get("current_cycle_active") is True and type(progress) is int and progress >= args.stop_after_progress:
+                    result = {"phase": args.label, "active_before_external_restart": True,
+                              "completed_cycles": len(completed), "complete": False}
+                    break
+            elif len(completed) >= args.min_completed_cycles:
+                inventory = phase.verify()
+                usage = api.usage(inventory)
+                phase.check()
+                require(usage["matches_inventory"] and usage["converged"] is True,
+                        "scanner usage is not converged to the exact S3 inventory")
+                result = {"phase": args.label, "completed_cycles": len(completed), "inventory": inventory,
+                          "usage": usage, "complete": True}
+                break
+            require(time.monotonic() < phase.deadline, "required scanner progress was not observed within the phase budget")
+            phase.stop.wait(min(args.interval, max(0, phase.deadline - time.monotonic())))
+    return phase.result(result)
+
+
+def run_burst(evidence, api, runtime, args):
+    require(evidence.state().get("load_complete") is True, "finish load before scanner-off burst")
+    state = evidence.state()
+    count = args.burst_objects or evidence.plan["objects"]
+    require(0 < count <= evidence.plan["objects"], "burst count exceeds owned fixture")
+    burst = state.setdefault("burst", {"id": str(uuid.uuid4()), "next": 0, "objects": count})
+    require(burst["objects"] == count and str(uuid.UUID(burst["id"])) == burst["id"]
+            and type(burst["next"]) is int and 0 <= burst["next"] <= count, "burst resume parameters changed")
+    initial_cursor = burst["next"]
+    evidence.save(state)
+    with Phase(evidence, runtime, api, "burst", False, args) as phase, ThreadPoolExecutor(args.workers) as pool:
+        api.owned()
+        while burst["next"] < count:
+            phase.check()
+            api.owned()
+            end = min(burst["next"] + args.batch, count)
+            free_space(evidence.plan, args.workers, phase.latest["memory"]["storage_available_bytes"])
+            def operation(index):
+                phase.check()
+                key = f".scanner-memory-burst/{burst['id']}/{index:08d}"
+                api.client.copy_object(Bucket=evidence.plan["bucket"], Key=key,
+                                       CopySource={"Bucket": evidence.plan["bucket"], "Key": object_key(evidence.plan, index)})
+                phase.check()
+                head = api.client.head_object(Bucket=evidence.plan["bucket"], Key=key)
+                require(head["ContentLength"] == evidence.plan["object_bytes"], "copied object size mismatch")
+                phase.check()
+                api.client.delete_object(Bucket=evidence.plan["bucket"], Key=key)
+            list(pool.map(operation, range(burst["next"], end)))
+            burst["next"] = end
+            evidence.save(state)
+            evidence.append("burst_batch", phase_id=phase.phase_id, completed=end, expected=count)
+            phase.check()
+        inventory = phase.verify()
+        phase.enter_stage("idle")
+        idle_deadline = time.monotonic() + args.idle_seconds
+        require(idle_deadline <= phase.deadline, "insufficient phase budget for the requested idle observation")
+        evidence.append("idle_start", phase="burst", phase_id=phase.phase_id, idle_seconds=args.idle_seconds)
+        while time.monotonic() < idle_deadline:
+            phase.check()
+            phase.stop.wait(min(args.interval, max(0, idle_deadline - time.monotonic())))
+    return phase.result({"phase": "burst", "complete": True, "objects": count, "burst_id": burst["id"],
+                         "cursor_start": initial_cursor, "cursor_end": burst["next"],
+                         "newly_completed_objects": burst["next"] - initial_cursor,
+                         "new_logical_copy_head_delete_requests": (burst["next"] - initial_cursor) * 3,
+                         "cumulative_completed_objects": burst["next"],
+                         "cumulative_logical_copy_head_delete_requests": burst["next"] * 3,
+                         "observed_http_request_count": None, "idle_seconds": args.idle_seconds, "inventory": inventory,
+                         "reclamation_verdict": "measurement only; RSS retention alone does not prove a leak"})
+
+
+def summarize(evidence):
+    samples = {}
+    results = []
+    errors = 0
+    path = evidence.root / "observations.jsonl"
+    if path.exists() or path.is_symlink():
+        with open_owned(path) as stream:
+            for line in stream:
+                require(len(line) <= 1024 * 1024, "oversized evidence row")
+                row = json.loads(line)
+                require(row.get("run_id") == evidence.plan["run_id"], "foreign run evidence")
+                if row["kind"] == "sample":
+                    memory = row["memory"]
+                    key = row["phase_id"] + ":" + memory["identity_sha256"] + ":" + row["stage"]
+                    group = samples.setdefault(key, {"phase": row["phase"], "phase_id": row["phase_id"],
+                                                     "process_identity": memory["identity_sha256"], "stage": row["stage"], "samples": 0,
+                                                     "rss_peak": 0, "sampled_current_peak_bytes": 0,
+                                                     "kernel_reported_peak_bytes": None, "first_rss": memory["rss_bytes"],
+                                                     "active_heap_samples": 0})
+                    group["samples"] += 1
+                    group["rss_peak"] = max(group["rss_peak"], memory["rss_bytes"])
+                    group["sampled_current_peak_bytes"] = max(group["sampled_current_peak_bytes"], memory["cgroup_current_bytes"])
+                    if memory.get("cgroup_peak_bytes") is not None:
+                        group["kernel_reported_peak_bytes"] = max(group["kernel_reported_peak_bytes"] or 0, memory["cgroup_peak_bytes"])
+                    group["last_rss"] = memory["rss_bytes"]
+                    group["active_heap_samples"] += memory["active_heap_bytes"] is not None
+                elif row["kind"] == "phase_result":
+                    results.append(row["result"])
+                elif row["kind"] in ("phase_error", "observation_error"):
+                    errors += 1
+    interrupted = set()
+    restart_observed = False
+    for result in results:
+        identity = result.get("process_identity")
+        if result.get("active_before_external_restart"):
+            interrupted.add(identity)
+        elif interrupted and identity and identity not in interrupted and result.get("complete") and result.get("completed_cycles", 0) > 0:
+            restart_observed = True
+    summary = {"schema": SCHEMA, "plan_sha256": evidence.manifest["plan_sha256"], "plan": evidence.plan,
+               "phases": samples, "results": results, "observation_errors": errors,
+               "restart_window_observed": restart_observed,
+               "full_scale": evidence.plan["objects"] == 2_000_000 and evidence.plan["object_bytes"] >= 50 * 1024,
+               "allocator_leak_verdict": "unknown; correlate live heap, anonymous RSS and cgroup file/slab across idle windows",
+               "overall_acceptance_passed": False}
+    atomic_json(evidence.root / "summary.json", summary)
+    return summary
+
+
+def prepare(args):
+    require(not args.run_dir.exists(), "prepare requires a new run directory")
+    identifier = str(uuid.uuid4())
+    plan = {"run_id": identifier, "created_at": utc(), "endpoint": local_url(args.endpoint),
+            "source_sha": args.source_sha, "binary_sha256": args.binary_sha256, "layout": args.layout,
+            "objects": args.objects, "object_bytes": args.object_bytes, "root_prefixes": args.root_prefixes,
+            "memory_bytes": args.memory_bytes, "data_root": str(args.data_root.resolve()) if args.data_root else None,
+            "data_volume": args.data_volume,
+            "bucket": "scanner-memory-" + uuid.UUID(identifier).hex}
+    manifest = {"schema": SCHEMA, "plan": plan, "plan_sha256": digest(plan)}
+    validate_manifest(manifest)
+    if plan["data_root"]:
+        free_space(plan, plan["objects"])
+    args.run_dir.mkdir(mode=0o700)
+    atomic_json(args.run_dir / "manifest.json", manifest)
+    return {"run_id": identifier, "bucket": plan["bucket"], "container_label": LABEL + "=" + identifier,
+            "manifest": str(args.run_dir / "manifest.json")}
+
+
+def main(argv=None):
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("action", choices=("prepare", "load", "verify", "probe", "burst", "summary"))
+    parser.add_argument("run_dir", type=Path)
+    parser.add_argument("--endpoint")
+    parser.add_argument("--source-sha")
+    parser.add_argument("--binary-sha256")
+    parser.add_argument("--layout", choices=("wide", "nested"), default="wide")
+    parser.add_argument("--objects", type=int, default=2_000_000)
+    parser.add_argument("--object-bytes", type=int, default=50 * 1024)
+    parser.add_argument("--root-prefixes", type=int, default=170_000)
+    parser.add_argument("--memory-bytes", type=int, default=4 * GIB)
+    parser.add_argument("--data-root", type=Path)
+    parser.add_argument("--data-volume")
+    parser.add_argument("--container")
+    parser.add_argument("--pid", type=int)
+    parser.add_argument("--allocator-metrics-url")
+    parser.add_argument("--workers", type=int, default=48)
+    parser.add_argument("--batch", type=int, default=256)
+    parser.add_argument("--seconds", type=float, default=3600)
+    parser.add_argument("--interval", type=float, default=5)
+    parser.add_argument("--label", default="scanner_probe")
+    parser.add_argument("--min-completed-cycles", type=int, default=2)
+    parser.add_argument("--stop-after-progress", type=int)
+    parser.add_argument("--burst-objects", type=int)
+    parser.add_argument("--idle-seconds", type=float, default=1800)
+    args = parser.parse_args(argv)
+    evidence = None
+    try:
+        args.run_dir = args.run_dir.absolute()
+        require(1 <= args.workers <= 256 and 1 <= args.batch <= 4096, "invalid client concurrency or batch bound")
+        require(math.isfinite(args.seconds) and 0 < args.seconds <= 7 * 86400
+                and math.isfinite(args.interval) and 0.1 <= args.interval <= 300
+                and math.isfinite(args.idle_seconds) and 0 < args.idle_seconds <= 86400, "invalid time bounds")
+        require(re.fullmatch(r"[a-zA-Z0-9_-]{1,64}", args.label), "invalid phase label")
+        require(args.min_completed_cycles >= 0 and (args.stop_after_progress is None or args.stop_after_progress > 0),
+                "invalid cycle/progress bounds")
+        if args.action == "prepare":
+            require(args.endpoint and args.source_sha and args.binary_sha256 and (args.data_root or args.data_volume),
+                    "prepare requires endpoint, source, binary and storage identity")
+            result = prepare(args)
+        else:
+            evidence = Evidence(args.run_dir)
+            with evidence.exclusive():
+                if args.action == "summary":
+                    result = summarize(evidence)
+                else:
+                    api = S3(evidence, args.workers)
+                    if args.action == "verify":
+                        result = api.verify(deadline=time.monotonic() + args.seconds)
+                    else:
+                        runtime = Runtime(evidence, args.container, args.pid, args.allocator_metrics_url, api.storage_paths)
+                        result = {"load": run_load, "probe": run_probe, "burst": run_burst}[args.action](evidence, api, runtime, args)
+                    evidence.append("phase_result", result=result)
+                    summarize(evidence)
+        print(json.dumps(result, sort_keys=True))
+        return 0
+    except AcceptanceError as error:
+        message = str(error)
+    except Exception:
+        message = "acceptance phase failed; raw credentials and server errors were withheld"
+    if evidence:
+        try:
+            evidence.append("phase_error", phase=args.action, reason=message)
+            summarize(evidence)
+        except Exception:
+            pass
+    print(message, file=sys.stderr)
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/security/check_performance_ab_workflow.sh

@@ -1,43 +0,0 @@
-#!/usr/bin/env bash
-set -euo pipefail
-
-workflow=".github/workflows/performance-ab.yml"
-
-require_absent_pattern() {
-  local pattern="$1"
-  local description="$2"
-
-  if grep -Eq -- "$pattern" "$workflow"; then
-    echo "invalid performance A/B workflow contract: $description" >&2
-    exit 1
-  fi
-}
-
-require_present_pattern() {
-  local pattern="$1"
-  local description="$2"
-
-  if ! grep -Eq -- "$pattern" "$workflow"; then
-    echo "invalid performance A/B workflow contract: $description" >&2
-    exit 1
-  fi
-}
-
-require_absent_pattern '(^|[^[:alnum:]_])pull_request(_target)?([^[:alnum:]_]|$)' "the workflow must not contain PR event handling"
-require_absent_pattern 'pull-requests[[:space:]]*:[[:space:]]*write' "the workflow must not receive PR write permission"
-require_absent_pattern 'permissions[[:space:]]*:[[:space:]]*write-all' "the workflow must not receive broad write permission"
-require_absent_pattern '^[[:space:]]*push:' "the workflow must not spend a release build on every main push"
-require_present_pattern 'listWorkflowRuns' "the scheduled baseline must come from workflow history"
-require_present_pattern 'status:[[:space:]]*"success"' "the scheduled baseline must be a successful run"
-require_present_pattern 'SCHEDULED_BASELINE_SHA' "the resolved scheduled baseline must reach the comparison"
-require_present_pattern "SCHEDULED_BASELINE_SHA:-\\\$candidate_sha" "the first scheduled run must seed from its verified candidate"
-require_present_pattern 'git merge-base --is-ancestor' "every baseline must stay on candidate history"
-require_present_pattern 'update the selected ref before comparing' "stale manual refs must fail before the benchmark"
-require_present_pattern 'setup failed before the rig ran' "an earlier setup failure must not become an empty exit status"
-require_present_pattern 'Cache successful candidate baseline' "a successful candidate must become the next cached baseline"
-if ! sed -n '/^  warp-ab:/,/^  alert-on-failure:/p' "$workflow" | grep -Eq '^    timeout-minutes:[[:space:]]*180([[:space:]]|$)'; then
-  echo "invalid performance A/B workflow contract: the cold-cache path must fit both builds, the A/B run, and evidence publication" >&2
-  exit 1
-fi
-
-echo "Performance A/B workflow contract ok."

scripts/test_inspect_btf_prototypes.py

@@ -0,0 +1,454 @@
+#!/usr/bin/env python3
+# Copyright 2026 RustFS Team
+# SPDX-License-Identifier: Apache-2.0
+"""Prototype/CLI contracts plus an optional independent bpftool reference."""
+
+import hashlib
+import json
+import os
+import re
+import selectors
+import struct
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+import inspect_btf_prototypes as reader
+
+HERE = Path(__file__).parent
+
+STRINGS = b"\0demo\0value\0u64\0Root\0Alias\0"
+
+
+def typ(kind, third, name="", count=0, kflag=False):
+    offset = STRINGS.index(name.encode() + b"\0") if name else 0
+    return struct.pack("<III", offset, (kind << 24) | count | (int(kflag) << 31), third)
+
+
+def blob(types):
+    raw = b"".join(types)
+    header = struct.pack("<HBBIIIII", 0xEB9F, 1, 0, 24, 0, len(raw), len(raw), len(STRINGS))
+    return header + raw + STRINGS
+
+
+def integer():
+    return typ(1, 8, "u64") + struct.pack("<I", 64)
+
+
+def function(types, result, args):
+    proto_id = len(types) + 1
+    param_name = STRINGS.index(b"value\0")
+    params = b"".join(struct.pack("<II", param_name, arg) for arg in args)
+    return blob(types + [typ(13, result, count=len(args)) + params, typ(12, proto_id, "demo")])
+
+
+def structure(types, field, size=8):
+    member = struct.pack("<III", STRINGS.index(b"value\0"), field, 0)
+    return blob(types + [typ(4, size, "Root", count=1) + member])
+
+
+class VoidBoundaryTests(unittest.TestCase):
+    def test_scalar_and_existing_pointer_declarations_are_retained(self):
+        result = reader._BTF(function([integer(), typ(2, 1)], 1, [2])).prototype("demo")
+        self.assertEqual(result["prototype"]["size_or_type"], 1)
+        self.assertEqual(result["args"][0]["type_ID"], 2)
+        self.assertEqual(result["args"][0]["type"]["kind"], 2)
+        self.assertEqual(result["args"][0]["pointee"]["ID"], 1)
+        self.assertEqual(result["return"]["size_or_type"], 8)
+
+    def test_void_pointer_argument_has_explicit_void_pointee(self):
+        raw = function([typ(2, 0)], 0, [1])
+        result = reader._BTF(raw).prototype("demo")
+        arg = result["args"][0]
+        self.assertEqual(arg["type"]["kind"], 2)
+        self.assertEqual(arg["type"]["size_or_type"], 0)
+        self.assertEqual(arg["pointee"], {"ID": 0, "name": "void", "kind": 0})
+        self.assertEqual(arg["pointee_chain"], [])
+        self.assertEqual(result["return"], {"ID": 0, "name": "void", "kind": 0})
+
+    def test_qualified_void_pointer_keeps_declared_chain(self):
+        raw = function([typ(10, 0), typ(2, 1)], 0, [2])
+        arg = reader._BTF(raw).prototype("demo")["args"][0]
+        self.assertEqual(arg["pointee"]["kind"], 0)
+        self.assertEqual([t["ID"] for t in arg["pointee_chain"]], [1])
+        self.assertEqual(arg["pointee_chain"][0]["kind"], 10)
+
+    def test_qualified_void_return_is_explicit(self):
+        raw = function([typ(8, 0, "Alias")], 1, [])
+        result = reader._BTF(raw).prototype("demo")
+        self.assertEqual(result["return"], {"ID": 0, "name": "void", "kind": 0})
+        self.assertEqual([t["ID"] for t in result["return_chain"]], [1])
+
+    def test_resolved_missing_nonzero_ids_are_not_void(self):
+        fixtures = [
+            function([], 0, [999]),
+            function([typ(2, 999)], 0, [1]),
+            function([typ(8, 999, "Alias"), typ(2, 1)], 0, [2]),
+            function([], 999, []),
+            function([typ(2, 999)], 1, []),
+            function([typ(2, 999), typ(2, 1)], 0, [2]),
+        ]
+        for raw in fixtures:
+            with self.subTest(raw=raw.hex()):
+                with self.assertRaisesRegex(ValueError, "actual_nonvoid_type_ID"):
+                    reader._BTF(raw).prototype("demo")
+
+    def test_named_direct_void_parameter_remains_unsupported(self):
+        with self.assertRaisesRegex(ValueError, "actual_nonvoid_type_ID"):
+            reader._BTF(function([], 0, [0])).prototype("demo")
+
+    def test_void_is_not_a_stored_type_record(self):
+        with self.assertRaisesRegex(ValueError, "unknown_BTF_kind_not_guessed"):
+            reader._BTF(blob([typ(0, 0)]))
+
+    def test_qualifier_cycle_is_rejected(self):
+        raw = function([typ(10, 2), typ(8, 1, "Alias")], 0, [1])
+        with self.assertRaisesRegex(ValueError, "acyclic_bounded_qualifier_chain"):
+            reader._BTF(raw).prototype("demo")
+
+    def test_qualifier_record_limit_is_preserved(self):
+        for count, accepted in [(31, True), (32, False)]:
+            types = [integer()]
+            for _ in range(count):
+                types.append(typ(10, len(types)))
+            raw = function(types, 0, [len(types)])
+            with self.subTest(scalar_qualifiers=count):
+                if accepted:
+                    self.assertEqual(reader._BTF(raw).prototype("demo")["args"][0]["type"]["kind"], 1)
+                else:
+                    with self.assertRaisesRegex(ValueError, "acyclic_bounded_qualifier_chain"):
+                        reader._BTF(raw).prototype("demo")
+        for count, accepted in [(32, True), (33, False)]:
+            types = [typ(10, index) for index in range(count)]
+            types.append(typ(2, count))
+            raw = function(types, 0, [len(types)])
+            with self.subTest(void_qualifiers=count):
+                if accepted:
+                    result = reader._BTF(raw).prototype("demo")["args"][0]
+                    self.assertEqual(len(result["pointee_chain"]), 32)
+                else:
+                    with self.assertRaisesRegex(ValueError, "acyclic_bounded_qualifier_chain"):
+                        reader._BTF(raw).prototype("demo")
+
+    def test_void_descriptor_does_not_infer_layout_or_record(self):
+        result = reader._BTF(function([typ(2, 0)], 0, [1])).prototype("demo")
+        void = result["args"][0]["pointee"]
+        self.assertEqual(set(void), {"ID", "name", "kind"})
+        self.assertNotIn("offset", void)
+        self.assertNotIn("size_or_type", void)
+        self.assertNotIn("pointer_width", result["args"][0])
+
+    def test_malformed_resolved_pointer_headers_are_rejected(self):
+        fixtures = [typ(2, 0, count=1), typ(2, 0, kflag=True), typ(2, 0, "value")]
+        for pointer in fixtures:
+            for args, ret in [([1], 0), ([], 1)]:
+                with self.subTest(pointer=pointer.hex(), args=args):
+                    with self.assertRaisesRegex(ValueError, "actual_pointer_header"):
+                        reader._BTF(function([pointer], ret, args)).prototype("demo")
+
+    def test_malformed_resolved_qualifier_headers_are_rejected(self):
+        for qualifier in [typ(10, 0, count=1), typ(10, 0, kflag=True), typ(10, 0, "value")]:
+            raw = function([qualifier, typ(2, 1)], 0, [2])
+            with self.subTest(qualifier=qualifier.hex()):
+                with self.assertRaisesRegex(ValueError, "actual_qualifier_header|unnamed_CVR_qualifier"):
+                    reader._BTF(raw).prototype("demo")
+
+    def test_incomplete_source_is_rejected(self):
+        raw = function([integer()], 1, [1])
+        for truncated in [raw[:23], raw[:-1]]:
+            with self.subTest(length=len(truncated)):
+                with self.assertRaises(ValueError):
+                    reader._BTF(truncated)
+
+    def test_default_record_and_unwrap_interfaces_stay_nonvoid(self):
+        btf = reader._BTF(function([], 0, []))
+        for operation in [lambda: btf.type(0), lambda: btf.unwrap(0), lambda: btf.unwrap(False), lambda: btf.unwrap(999)]:
+            with self.assertRaisesRegex(ValueError, "actual_nonvoid_type_ID"):
+                operation()
+
+
+class CliTests(unittest.TestCase):
+    def run_cli(self, data, *names):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "input.btf"
+            path.write_bytes(data)
+            command = [sys.executable, "-B", str(HERE / "inspect_btf_prototypes.py"), "--btf", str(path)]
+            for name in names:
+                command.extend(["--function", name])
+            result = subprocess.run(command, capture_output=True, timeout=10)
+            self.assertEqual(path.read_bytes(), data)
+            self.assertEqual(sorted(p.name for p in Path(directory).iterdir()), ["input.btf"])
+            return result
+
+    def test_complete_json_has_observed_source_digest(self):
+        raw = function([typ(2, 0)], 0, [1])
+        result = self.run_cli(raw, "demo")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        value = json.loads(result.stdout)
+        self.assertEqual(value["schema"], 1)
+        self.assertEqual(value["source"]["bytes"], len(raw))
+        self.assertEqual(value["source"]["sha256"], hashlib.sha256(raw).hexdigest())
+        self.assertEqual(value["prototypes"][0]["args"][0]["pointee"]["ID"], 0)
+
+    def test_multiple_functions_keep_complete_request_order(self):
+        raw = blob([typ(13, 0), typ(12, 1, "demo"), typ(12, 1, "value")])
+        result = self.run_cli(raw, "value", "demo")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        document = json.loads(result.stdout)
+        self.assertEqual([p["function"]["name"] for p in document["prototypes"]], ["value", "demo"])
+
+    def test_later_missing_function_emits_no_partial_report(self):
+        result = self.run_cli(function([], 0, []), "demo", "missing")
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+        self.assertIn(b"unique_actual_named_type_or_function", result.stderr)
+
+    def test_corrupt_input_is_an_error_not_an_empty_success(self):
+        result = self.run_cli(b"not a BTF source", "demo")
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+
+    def test_duplicate_names_are_rejected(self):
+        result = self.run_cli(function([], 0, []), "demo", "demo")
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+
+    def test_absent_input_is_rejected_without_traceback(self):
+        with tempfile.TemporaryDirectory() as directory:
+            result = subprocess.run(
+                [sys.executable, "-B", str(HERE / "inspect_btf_prototypes.py"),
+                 "--btf", str(Path(directory) / "missing"), "--function", "demo"],
+                capture_output=True, timeout=10,
+            )
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+        self.assertNotIn(b"Traceback", result.stderr)
+
+    def test_source_limit_is_enforced_before_parsing(self):
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "oversized"
+            with path.open("wb") as output:
+                output.seek(reader.MAX_BTF_BYTES)
+                output.write(b"x")
+            with self.assertRaisesRegex(ValueError, "BTF_source_exceeds_32_MiB"):
+                reader.inspect_prototypes(path, ["demo"])
+
+    def test_report_limit_rejects_before_output(self):
+        with self.assertRaisesRegex(ValueError, "prototype_report_exceeds_1_MiB"):
+            reader.encode_document({"large": "x" * reader.MAX_JSON_BYTES})
+
+    def test_large_valid_prototype_report_has_empty_stdout_on_rejection(self):
+        strings = STRINGS + b"Q" * 4095 + b"\0"
+        types = [integer()]
+        for _ in range(31):
+            types.append(struct.pack("<III", len(STRINGS), 8 << 24, len(types)))
+        params = struct.pack("<II", STRINGS.index(b"value\0"), 32) * 16
+        types.extend([typ(13, 0, count=16) + params, typ(12, 33, "demo")])
+        records = b"".join(types)
+        raw = struct.pack("<HBBIIIII", 0xEB9F, 1, 0, 24, 0, len(records),
+                          len(records), len(strings)) + records + strings
+        result = self.run_cli(raw, "demo")
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+        self.assertIn(b"prototype_report_exceeds_1_MiB", result.stderr)
+
+    def test_function_count_is_bounded_before_file_read(self):
+        with self.assertRaisesRegex(ValueError, "one_to_sixteen"):
+            reader.inspect_prototypes("not-read", [str(n) for n in range(17)])
+
+    def test_selector_shapes_are_rejected_before_file_read(self):
+        for names in [None, "demo", [[]], [1], [""], ["a\0b"]]:
+            with self.subTest(names=names):
+                with self.assertRaisesRegex(ValueError, "one_to_sixteen"):
+                    reader.inspect_prototypes("not-read", names)
+
+    def test_symlink_and_fifo_are_rejected_without_blocking(self):
+        with tempfile.TemporaryDirectory() as directory:
+            source = Path(directory) / "source"
+            source.write_bytes(function([], 0, []))
+            link = Path(directory) / "link"
+            link.symlink_to(source)
+            fifo = Path(directory) / "fifo"
+            os.mkfifo(fifo)
+            for path in [link, fifo]:
+                with self.subTest(path=path.name):
+                    with self.assertRaisesRegex(ValueError, "regular_non_symlink"):
+                        reader.inspect_prototypes(path, ["demo"])
+
+    def test_ambiguous_function_identity_is_rejected(self):
+        raw = blob([typ(13, 0), typ(12, 1, "demo"), typ(12, 1, "demo")])
+        result = self.run_cli(raw, "demo")
+        self.assertEqual(result.returncode, 2)
+        self.assertEqual(result.stdout, b"")
+
+
+def bounded_reference_command(command, timeout=30):
+    """Bound both pipe bytes and child lifetime, including a silent child."""
+    process = subprocess.Popen(
+        command, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+    )
+    buffers = {"stdout": bytearray(), "stderr": bytearray()}
+    limits = {"stdout": reader.MAX_JSON_BYTES, "stderr": 64 * 1024}
+    deadline = time.monotonic() + timeout
+    try:
+        with selectors.DefaultSelector() as selector:
+            selector.register(process.stdout, selectors.EVENT_READ, "stdout")
+            selector.register(process.stderr, selectors.EVENT_READ, "stderr")
+            while selector.get_map():
+                remaining = deadline - time.monotonic()
+                ready = selector.select(max(0, remaining))
+                if remaining <= 0 or not ready:
+                    raise ValueError("native_reference_timeout")
+                for key, _ in ready:
+                    chunk = os.read(key.fileobj.fileno(), 64 * 1024)
+                    if not chunk:
+                        selector.unregister(key.fileobj)
+                        continue
+                    buffers[key.data].extend(chunk)
+                    if len(buffers[key.data]) > limits[key.data]:
+                        raise ValueError("native_reference_" + key.data + "_exceeded_its_bound")
+        code = process.wait(timeout=max(0.01, deadline - time.monotonic()))
+        if code != 0:
+            raise ValueError("native_reference_failed returncode=" + str(code))
+        if buffers["stderr"]:
+            print("Native reference diagnostics: " + repr(bytes(buffers["stderr"])), file=sys.stderr)
+        return bytes(buffers["stdout"])
+    except (ValueError, subprocess.TimeoutExpired) as error:
+        raise ValueError(str(error) + "; captured_stderr=" + repr(bytes(buffers["stderr"]))) from error
+    finally:
+        process.stdout.close()
+        process.stderr.close()
+        if process.poll() is None:
+            process.kill()
+            process.wait()
+
+
+class NativeCaptureTests(unittest.TestCase):
+    def test_native_stderr_and_exact_failure_status_are_retained(self):
+        command = [sys.executable, "-c", "import sys; sys.stderr.write('reference marker'); sys.exit(7)"]
+        with self.assertRaisesRegex(ValueError, "returncode=7.*reference marker"):
+            bounded_reference_command(command)
+
+    def test_native_success_returns_complete_stdout(self):
+        self.assertEqual(bounded_reference_command([sys.executable, "-c", "print('complete')"]), b"complete\n")
+
+    def test_native_output_overflow_is_refused(self):
+        command = [sys.executable, "-c", "import os; os.write(1, b'x' * (1024 * 1024 + 1))"]
+        with self.assertRaisesRegex(ValueError, "stdout_exceeded_its_bound"):
+            bounded_reference_command(command)
+
+    def test_silent_native_child_cannot_outlive_the_deadline(self):
+        with self.assertRaisesRegex(ValueError, "native_reference_timeout"):
+            bounded_reference_command([sys.executable, "-c", "import threading; threading.Event().wait()"], timeout=0.1)
+
+
+def native_reference(path):
+    """Compare a small compiler-produced fixture with bpftool, not this decoder."""
+    path = Path(path)
+    with path.open("rb") as source:
+        raw = source.read(64 * 1024 + 1)
+    if len(raw) > 64 * 1024:
+        raise ValueError("native_reference_requires_the_small_compiler_fixture")
+    native = bounded_reference_command(
+        [os.environ.get("BPFTOOL", "bpftool"), "btf", "dump", "file", str(path), "format", "raw"])
+    records, current = {}, None
+    for line in native.decode("utf-8").splitlines():
+        match = re.fullmatch(r"\[(\d+)\] (\w+) '(.*)'(?: (.*))?", line)
+        if match:
+            ident, kind, name, extra = match.groups()
+            current = {"kind": kind, "name": name, "extra": extra or "", "args": []}
+            records[int(ident)] = current
+        elif current is not None and current["kind"] == "FUNC_PROTO":
+            argument = re.fullmatch(r"\s*'(.*)' type_id=(\d+)", line)
+            if not argument:
+                raise ValueError("unexpected_native_prototype_row")
+            current["args"].append((argument.group(1), int(argument.group(2))))
+    names = ["rustfs_btf_probe", "rustfs_btf_void"]
+    command = [sys.executable, "-B", str(HERE / "inspect_btf_prototypes.py"), "--btf", str(path)]
+    for name in names:
+        command.extend(["--function", name])
+    result = subprocess.run(command, capture_output=True, timeout=30, check=True)
+    report = json.loads(result.stdout)
+    if report["source"]["sha256"] != hashlib.sha256(raw).hexdigest() or path.read_bytes() != raw:
+        raise ValueError("native_reference_source_changed")
+    if (report["schema"] != 1 or report["source"]["bytes"] != len(raw)
+            or [p["function"]["name"] for p in report["prototypes"]] != names):
+        raise ValueError("native_reference_membership_mismatch")
+    kinds = {"INT": 1, "PTR": 2, "ARRAY": 3, "STRUCT": 4, "UNION": 5,
+             "ENUM": 6, "FWD": 7, "TYPEDEF": 8, "VOLATILE": 9, "CONST": 10,
+             "RESTRICT": 11, "FUNC": 12, "FUNC_PROTO": 13, "VAR": 14,
+             "DATASEC": 15, "FLOAT": 16, "DECL_TAG": 17, "TYPE_TAG": 18, "ENUM64": 19}
+
+    def check_record(actual):
+        if actual["ID"] == 0:
+            if actual != {"ID": 0, "name": "void", "kind": 0} or 0 in records:
+                raise ValueError("native_reference_void_mismatch")
+            return
+        expected = records[actual["ID"]]
+        name = "" if expected["name"] == "(anon)" else expected["name"]
+        if actual["kind"] != kinds[expected["kind"]] or actual["name"] != name:
+            raise ValueError("native_reference_record_mismatch")
+        declared = re.search(r"(?:\btype_id|\bret_type_id|\bsize)=(\d+)", expected["extra"])
+        if declared and actual["size_or_type"] != int(declared.group(1)):
+            raise ValueError("native_reference_type_or_size_mismatch")
+        count = re.search(r"\bvlen=(\d+)", expected["extra"])
+        if count and actual["count"] != int(count.group(1)):
+            raise ValueError("native_reference_parameter_count_mismatch")
+
+    def check_chain(start, chain, leaf):
+        current = start
+        for index, record in enumerate(chain):
+            if record["ID"] != current:
+                raise ValueError("native_reference_chain_disconnected")
+            check_record(record)
+            if record["kind"] in (8, 9, 10, 11, 18):
+                current = record["size_or_type"]
+            elif index != len(chain) - 1 or record != leaf:
+                raise ValueError("native_reference_chain_leaf_mismatch")
+        check_record(leaf)
+        if current != leaf["ID"] or (leaf["ID"] != 0 and (not chain or chain[-1] != leaf)):
+            raise ValueError("native_reference_incomplete_chain")
+
+    for name, actual in zip(names, report["prototypes"]):
+        matches = [(ident, value) for ident, value in records.items()
+                   if value["kind"] == "FUNC" and value["name"] == name]
+        if len(matches) != 1:
+            raise ValueError("native_reference_function_is_not_unique")
+        ident, function = matches[0]
+        proto_id = int(re.search(r"\btype_id=(\d+)", function["extra"]).group(1))
+        expected = records[proto_id]
+        return_id = int(re.search(r"\bret_type_id=(\d+)", expected["extra"]).group(1))
+        args = [(arg["name"], arg["type_ID"]) for arg in actual["args"]]
+        if (actual["function"]["ID"] != ident or actual["prototype"]["ID"] != proto_id
+                or actual["prototype"]["size_or_type"] != return_id or args != expected["args"]):
+            raise ValueError("native_reference_declaration_mismatch")
+        for record in [actual["function"], actual["prototype"], actual["return"], *actual["return_chain"]]:
+            check_record(record)
+        check_chain(return_id, actual["return_chain"], actual["return"])
+        for arg in actual["args"]:
+            check_chain(arg["type_ID"], arg["chain"], arg["type"])
+            for record in [arg["type"], *arg["chain"], *arg.get("pointee_chain", [])]:
+                check_record(record)
+            if "pointee" in arg:
+                check_record(arg["pointee"])
+            if arg["type"]["kind"] == 2:
+                pointer = records[arg["type"]["ID"]]
+                target = int(re.search(r"\btype_id=(\d+)", pointer["extra"]).group(1))
+                if pointer["kind"] != "PTR" or target != arg["type"]["size_or_type"]:
+                    raise ValueError("native_reference_pointer_mismatch")
+                check_chain(target, arg["pointee_chain"], arg["pointee"])
+                for qualifier in arg["pointee_chain"]:
+                    record = records[qualifier["ID"]]
+                    if int(re.search(r"\btype_id=(\d+)", record["extra"]).group(1)) != qualifier["size_or_type"]:
+                        raise ValueError("native_reference_qualifier_mismatch")
+    print("Validated both compiler-produced prototypes against independent bpftool output")
+
+
+if __name__ == "__main__":
+    if len(sys.argv) == 3 and sys.argv[1] == "--native-reference":
+        native_reference(sys.argv[2])
+    else:
+        unittest.main()

scripts/test_scanner_memory_acceptance.py

@@ -0,0 +1,477 @@
+#!/usr/bin/env python3
+"""Bounded ownership, inventory, memory and restart evidence tests; no live service."""
+
+import copy
+import io
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+import tempfile
+import unittest
+from unittest.mock import Mock, patch
+from types import SimpleNamespace
+import uuid
+
+import scanner_memory_acceptance as acceptance
+
+
+class FakeClient:
+    def __init__(self, pages, data):
+        self.pages, self.data = iter(pages), data
+
+    def list_objects_v2(self, **_):
+        return next(self.pages)
+
+    def get_object(self, **_):
+        return {"Body": io.BytesIO(self.data)}
+
+
+class AcceptanceTest(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.root = Path(self.temporary.name).resolve()
+        self.root.chmod(0o700)
+        data = self.root / "data"
+        data.mkdir()
+        run_id = str(uuid.uuid4())
+        self.plan = {"run_id": run_id, "created_at": acceptance.utc(), "endpoint": "http://127.0.0.1:19000",
+                     "source_sha": "a" * 40, "binary_sha256": "b" * 64, "layout": "wide", "objects": 7,
+                     "object_bytes": 51200, "root_prefixes": 3, "memory_bytes": 4 * acceptance.GIB,
+                     "data_root": str(data), "data_volume": None, "bucket": "scanner-memory-" + uuid.UUID(run_id).hex}
+        self.manifest = {"schema": 1, "plan": self.plan, "plan_sha256": acceptance.digest(self.plan)}
+        acceptance.atomic_json(self.root / "manifest.json", self.manifest)
+        self.evidence = acceptance.Evidence(self.root)
+
+    def api(self, entries, data=None, split=4):
+        api = acceptance.S3.__new__(acceptance.S3)
+        api.plan, api.evidence = self.plan, self.evidence
+        api.owned = lambda: None
+        pages = []
+        for offset in range(0, len(entries), split):
+            more = offset + split < len(entries)
+            pages.append({"Contents": entries[offset:offset + split], "IsTruncated": more,
+                          "NextContinuationToken": str(offset + split) if more else None})
+        api.client = FakeClient(pages, acceptance.payload(self.plan) if data is None else data)
+        return api
+
+    def entries(self):
+        return [{"Key": acceptance.OWNER_KEY, "Size": len(acceptance.owner_body(self.manifest))}] + [
+            {"Key": acceptance.object_key(self.plan, index), "Size": self.plan["object_bytes"]}
+            for index in range(self.plan["objects"])]
+
+    def test_layout_roundtrip_and_root_width(self):
+        for layout in ("wide", "nested"):
+            self.plan["layout"] = layout
+            keys = [acceptance.object_key(self.plan, index) for index in range(7)]
+            self.assertEqual(len(set(keys)), 7)
+            self.assertEqual([acceptance.key_index(self.plan, key) for key in keys], list(range(7)))
+            self.assertEqual(len({key.split('/')[0] for key in keys}), 3 if layout == "wide" else 1)
+        for key in ("photos/00/00/source:000001/00000000.webp", "foreign", "source:999999/00000000.webp"):
+            with self.assertRaises(acceptance.AcceptanceError):
+                acceptance.key_index(self.plan, key)
+
+    def test_exact_inventory_payload_and_marker(self):
+        result = self.api(self.entries()).verify(samples=3)
+        self.assertEqual(result["objects"], 7)
+        self.assertEqual(result["logical_bytes"], 7 * 51200)
+        self.assertEqual(result["scanner_expected_objects"], 8)
+        self.assertEqual(result["payload_samples"], 3)
+
+    def test_inventory_rejects_missing_duplicate_foreign_and_wrong_size(self):
+        entries = self.entries()
+        variants = [entries[:-1], entries + [entries[1]], entries + [{"Key": "foreign", "Size": 51200}],
+                    [*entries[:-1], {**entries[-1], "Size": 1}], entries[1:]]
+        for variant in variants:
+            with self.subTest(variant=variant), self.assertRaises(acceptance.AcceptanceError):
+                self.api(variant).verify()
+        with self.assertRaises(acceptance.AcceptanceError):
+            self.api(entries, b"corrupt payload").verify()
+
+    def test_inventory_timeout_is_not_success(self):
+        with self.assertRaises(acceptance.AcceptanceError):
+            self.api(self.entries()).verify(deadline=acceptance.time.monotonic() - 1)
+
+    def test_manifest_and_resume_marker_fail_closed(self):
+        changed = copy.deepcopy(self.manifest)
+        changed["plan"]["objects"] = 8
+        with self.assertRaises(acceptance.AcceptanceError):
+            acceptance.validate_manifest(changed)
+        for state in ({"plan_sha256": "foreign", "next_object": 0},
+                      {"plan_sha256": self.manifest["plan_sha256"], "next_object": 8}):
+            acceptance.atomic_json(self.root / "state.json", state)
+            with self.assertRaises(acceptance.AcceptanceError):
+                self.evidence.state()
+
+    def test_runtime_requires_ownership_before_exec(self):
+        runtime = acceptance.Runtime(self.evidence, container="foreign")
+        raw = [{"Config": {"Labels": {acceptance.LABEL: str(uuid.uuid4())}}}]
+        with patch.object(acceptance, "command", return_value=json.dumps(raw).encode()) as command:
+            with self.assertRaises(acceptance.AcceptanceError):
+                runtime.read_files()
+            self.assertEqual(command.call_count, 1)
+            self.assertEqual(command.call_args.args[0][:2], ["docker", "inspect"])
+
+    def container_observation(self):
+        container = {"Id": "owned-id", "Config": {"Labels": {acceptance.LABEL: self.plan["run_id"]}},
+                     "State": {"Running": True, "OOMKilled": False, "StartedAt": "2026-01-01T00:00:00Z"},
+                     "HostConfig": {"Memory": self.plan["memory_bytes"]}, "RestartCount": 0,
+                     "Mounts": [{"Type": "bind", "Source": self.plan["data_root"], "Destination": "/data", "RW": True}],
+                     "NetworkSettings": {"Ports": {"9000/tcp": [{"HostPort": "19000", "HostIp": "127.0.0.1"}]}}}
+        files = {"stat": "1 (rustfs) " + " ".join(["S"] + ["0"] * 18 + ["1234"]),
+                 "status": "VmRSS: 20 kB\nRssAnon: 12 kB", "smaps_rollup": "Anonymous: 12 kB",
+                 "cgroup": "0::/", "environ": "RUSTFS_VOLUMES=/data\0RUSTFS_SCANNER_ENABLED=false\0RUSTFS_SECRET_KEY=private-test-value\0",
+                 "exe_inode": "1:5678", "memory.max": str(self.plan["memory_bytes"]), "memory.current": "40960",
+                 "memory.peak": "50000", "memory.stat": "anon 12288\nfile 20000", "memory.events": "oom 0\noom_kill 0",
+                 "storage": "disk 100 10 90 10% /data", "storage_root": "/data",
+                 "storage_devices": "1\n1", "volume_paths": "/data",
+                 "mountinfo": "1 0 0:1 / / rw - ext4 disk rw\n2 1 0:2 / /data rw - ext4 disk rw"}
+        def command(argv, **_):
+            if argv[:2] == ["docker", "inspect"]:
+                return json.dumps([container]).encode()
+            if argv[3] == "sha256sum":
+                return (self.plan["binary_sha256"] + "  /proc/1/exe\n").encode()
+            return "".join(f"\n__FILE_{name}__\n{value}\n" for name, value in files.items()).encode()
+        return container, files, command
+
+    def test_capacity_rejects_unused_owned_mount_before_load_writes(self):
+        runtime = acceptance.Runtime(self.evidence, container="owned")
+        runtime.storage_paths = lambda: ["/foreign"]
+        _, files, command = self.container_observation()
+        files["environ"] = "RUSTFS_VOLUMES=/foreign\0"
+        files["storage"] = "disk 20000000 10 10000000 10% /data"
+        api = SimpleNamespace(owned=Mock(), client=Mock(), status=lambda _: {}, verify=lambda **_: {})
+        args = SimpleNamespace(seconds=10, interval=1, workers=1, batch=1)
+        with patch.object(acceptance, "command", side_effect=command):
+            with self.assertRaisesRegex(acceptance.AcceptanceError, "storage"):
+                acceptance.run_load(self.evidence, api, runtime, args)
+        api.owned.assert_not_called()
+        api.client.put_object.assert_not_called()
+
+    def test_owned_container_collects_real_fields_without_emitting_private_environment(self):
+        runtime = acceptance.Runtime(self.evidence, container="owned")
+        runtime.storage_paths = lambda: ["/data"]
+        container, files, command = self.container_observation()
+        with patch.object(acceptance, "command", side_effect=command) as calls:
+            sample = runtime.sample()
+            runtime.sample()
+        self.assertEqual(sum(call.args[0][3:4] == ["sha256sum"] for call in calls.call_args_list), 1)
+        self.assertEqual(sample["storage_available_bytes"], 90 * 1024)
+        self.assertEqual(sample["configuration"], {"RUSTFS_SCANNER_ENABLED": "false"})
+        self.assertNotIn("private-test-value", json.dumps(sample))
+        self.assertNotIn("SECRET", json.dumps(sample))
+        container["NetworkSettings"]["Ports"] = {}
+        with patch.object(acceptance, "command", side_effect=command), self.assertRaises(acceptance.AcceptanceError):
+            runtime.sample()
+
+    def test_storage_inventory_uses_complete_live_server_configuration(self):
+        api = acceptance.S3.__new__(acceptance.S3)
+        info = {"observations": [{"status": "succeeded", "cached": False}],
+                "disks": [{"path": "/data/one", "local": True, "state": "ok"},
+                          {"path": "/data/two", "local": True, "state": "ok"}],
+                "backend": {"DrivesPerSet": [2], "TotalSets": [1]}}
+        api.admin = Mock(return_value={"info": info})
+        self.assertEqual(api.storage_paths(), ["/data/one", "/data/two"])
+        api.admin.assert_called_once_with("/rustfs/admin/v3/storageinfo")
+        bad = []
+        for field, value in (("observations", []), ("disks", info["disks"][:1]),
+                             ("backend", {})):
+            bad.append({**info, field: value})
+        for field, value in (("status", "failed"), ("cached", True)):
+            bad.append({**info, "observations": [{**info["observations"][0], field: value}]})
+        bad.append({**info, "observations": info["observations"] * 2})
+        for field, value in (("local", False), ("state", "offline")):
+            bad.append({**info, "disks": [{**disk, field: value} for disk in info["disks"]]})
+        for invalid in bad:
+            with self.subTest(inventory=invalid), self.assertRaises(acceptance.AcceptanceError):
+                api.admin.return_value = {"info": invalid}
+                api.storage_paths()
+
+    def test_effective_storage_paths_fail_closed_on_ambiguous_or_foreign_roots(self):
+        for paths in (["/foreign"], ["/data-other"], ["/data", "/foreign"], ["data"],
+                      ["/data/../foreign"], ["http://node/data"], ["/data/"], [None], []):
+            with self.subTest(paths=paths), self.assertRaises(acceptance.AcceptanceError):
+                acceptance.owned_storage_paths("/data", paths)
+        self.assertEqual(acceptance.owned_storage_paths("/data", ["/data", "/data/disk1"]),
+                         [Path("/data"), Path("/data/disk1")])
+
+    def test_container_rejects_symlink_escape_nested_mount_and_foreign_filesystem(self):
+        runtime = acceptance.Runtime(self.evidence, container="owned", storage_paths=lambda: ["/data"])
+        for field, value in (("volume_paths", "/foreign"), ("storage_root", "/foreign"),
+                             ("storage_devices", "1\n2"), ("volume_paths", ""),
+                             ("mountinfo", "1 0 0:1 / / rw - ext4 disk rw\n2 1 0:2 / /data/nested rw - ext4 disk rw")):
+            with self.subTest(field=field, value=value):
+                _, files, command = self.container_observation()
+                files[field] = value
+                with patch.object(acceptance, "command", side_effect=command), self.assertRaises(acceptance.AcceptanceError):
+                    runtime.sample()
+
+    def test_container_uses_effective_inventory_instead_of_shadowed_environment(self):
+        runtime = acceptance.Runtime(self.evidence, container="owned", storage_paths=lambda: ["/data"])
+        _, files, command = self.container_observation()
+        # The serving process reports /data after positional CLI args override this env value.
+        files["environ"] += "RUSTFS_VOLUMES=/foreign\0"
+        with patch.object(acceptance, "command", side_effect=command):
+            first = runtime.sample()
+            runtime.storage_paths = lambda: ["/foreign"]
+            with self.assertRaises(acceptance.AcceptanceError):
+                runtime.sample()
+        self.assertIn("storage_binding_sha256", first["identity"])
+        self.assertNotIn("/foreign", json.dumps(first))
+
+    def test_storage_identity_is_stable_when_disk_inventory_order_changes(self):
+        paths = ["/data/one", "/data/two"]
+        runtime = acceptance.Runtime(self.evidence, container="owned", storage_paths=lambda: paths)
+        _, files, command = self.container_observation()
+        files["volume_paths"] = "/data/one\n/data/two"
+        files["storage_devices"] = "1\n1\n1"
+        with patch.object(acceptance, "command", side_effect=command) as calls:
+            first = runtime.sample()
+            paths.reverse()
+            second = runtime.sample()
+        self.assertEqual(first["identity_sha256"], second["identity_sha256"])
+        observations = [call.args[0] for call in calls.call_args_list if call.args[0][3:5] == ["sh", "-c"]]
+        self.assertEqual([argv[-2:] for argv in observations], [["/data/one", "/data/two"]] * 2)
+
+    def test_named_volume_requires_effective_disks_inside_owned_mount(self):
+        self.evidence.plan["data_root"] = None
+        self.evidence.plan["data_volume"] = "owned-volume"
+        container, _, command = self.container_observation()
+        container["Mounts"] = [{"Type": "volume", "Name": "owned-volume", "Destination": "/data", "RW": True}]
+        def inspect_volume(argv, **kwargs):
+            if argv[:3] == ["docker", "volume", "inspect"]:
+                return json.dumps([{"Labels": {acceptance.LABEL: self.plan["run_id"]}}]).encode()
+            return command(argv, **kwargs)
+        runtime = acceptance.Runtime(self.evidence, container="owned", storage_paths=lambda: ["/data"])
+        with patch.object(acceptance, "command", side_effect=inspect_volume):
+            runtime.sample()
+            runtime.storage_paths = lambda: ["/foreign"]
+            with self.assertRaises(acceptance.AcceptanceError):
+                runtime.sample()
+
+    def test_native_capacity_binds_actual_disk_paths_before_observing_free_space(self):
+        owned = self.plan["data_root"]
+        child = Path(owned) / "disk"
+        child.mkdir()
+        foreign = self.root / "foreign"
+        foreign.mkdir()
+        link = Path(owned) / "alias"
+        link.symlink_to(foreign, target_is_directory=True)
+        mountinfo = b"1 0 0:1 / / rw - ext4 disk rw"
+        with patch.object(Path, "samefile", return_value=True), patch.object(acceptance.shutil, "disk_usage") as usage:
+            usage.return_value = SimpleNamespace(free=4096)
+            storage, binding = acceptance.native_storage(Path("/proc/123"), owned, [str(child)], mountinfo)
+            self.assertEqual(storage.split()[3], b"4")
+            self.assertEqual(len(binding), 64)
+            usage.assert_called_once_with(Path(owned))
+            for paths in ([str(foreign)], [str(link)], [owned, str(foreign)], []):
+                usage.reset_mock()
+                with self.subTest(paths=paths), self.assertRaises(acceptance.AcceptanceError):
+                    acceptance.native_storage(Path("/proc/123"), owned, paths, mountinfo)
+                usage.assert_not_called()
+            nested = mountinfo + f"\n2 1 0:2 / {owned}/nested rw - ext4 disk rw".encode()
+            usage.reset_mock()
+            with self.assertRaises(acceptance.AcceptanceError):
+                acceptance.native_storage(Path("/proc/123"), owned, [owned], nested)
+            usage.assert_not_called()
+        with patch.object(Path, "samefile", return_value=False), self.assertRaises(acceptance.AcceptanceError):
+            acceptance.native_storage(Path("/proc/123"), owned, [owned], mountinfo)
+
+    def test_memory_separates_rss_anonymous_page_cache_and_unknown_heap(self):
+        runtime = acceptance.Runtime(self.evidence, pid=123)
+        files = {"memory.max": str(4 * acceptance.GIB).encode(), "memory.current": b"40960", "memory.peak": b"50000",
+                 "status": b"VmRSS:\t20 kB\nRssAnon:\t12 kB\n", "smaps_rollup": b"Anonymous: 12 kB\n",
+                 "memory.events": b"low 0\nhigh 0\nmax 0\noom 0\noom_kill 0\n",
+                 "memory.stat": b"anon 12288\nfile 20000\nslab 8672\n", "storage": b"disk 100 10 90 10% /data"}
+        with patch.object(runtime, "read_files", return_value=({"pid": 123}, files, {"RUSTFS_SCANNER_ENABLED": "true"})):
+            sample = runtime.sample()
+        self.assertEqual(sample["rss_bytes"], 20480)
+        self.assertEqual(sample["rss_anon_bytes"], 12288)
+        self.assertEqual(sample["cgroup_stat"]["file"], 20000)
+        self.assertIsNone(sample["active_heap_bytes"])
+        files["memory.events"] = b"oom 1\noom_kill 1\n"
+        with patch.object(runtime, "read_files", return_value=({"pid": 123}, files, {})), self.assertRaises(acceptance.AcceptanceError):
+            runtime.sample()
+
+    def test_process_stat_parentheses(self):
+        raw = "123 (worker name ) tail) " + " ".join(["S"] + ["0"] * 18 + ["9876"])
+        self.assertEqual(acceptance.parse_proc_stat(raw), 9876)
+
+    def test_restart_requires_later_success_on_different_process(self):
+        self.evidence.append("phase_result", result={"process_identity": "b", "complete": True, "completed_cycles": 2})
+        self.evidence.append("phase_result", result={"process_identity": "a", "active_before_external_restart": True})
+        self.assertFalse(acceptance.summarize(self.evidence)["restart_window_observed"])
+        self.evidence.append("phase_result", result={"process_identity": "a", "complete": True, "completed_cycles": 2})
+        self.assertFalse(acceptance.summarize(self.evidence)["restart_window_observed"])
+        self.evidence.append("phase_result", result={"process_identity": "c", "complete": True, "completed_cycles": 2})
+        self.assertTrue(acceptance.summarize(self.evidence)["restart_window_observed"])
+        self.assertFalse(acceptance.summarize(self.evidence)["overall_acceptance_passed"])
+
+    def test_probe_counts_new_success_cycles_and_requires_exact_usage(self):
+        self.evidence.save({"plan_sha256": self.manifest["plan_sha256"], "next_object": 7, "load_complete": True})
+        counter = iter(range(100, 1000))
+        api = SimpleNamespace(owned=lambda: None,
+                              status=lambda _: {"enabled": True, "metrics": {
+                                  "last_cycle_end_unix_secs": next(counter), "last_cycle_result": "success"}},
+                              verify=lambda **_: {"objects": 7},
+                              usage=lambda _: {"matches_inventory": True, "converged": True})
+        runtime = SimpleNamespace(sample=lambda: {"identity_sha256": "process-a"})
+        args = SimpleNamespace(seconds=10, interval=0.01, label="cycles", stop_after_progress=None, min_completed_cycles=2)
+        result = acceptance.run_probe(self.evidence, api, runtime, args)
+        self.assertEqual(result["completed_cycles"], 2)
+        self.assertTrue(result["complete"])
+        self.assertIn("phase_id", result)
+
+    def test_runtime_binding_checks_tcp_address_and_family(self):
+        endpoint = self.plan["endpoint"]
+        self.assertTrue(acceptance.docker_endpoint_matches({"9000/tcp": [{"HostPort": "19000", "HostIp": "0.0.0.0"}]}, endpoint))
+        for protocol, ip in (("9000/tcp", "127.0.0.2"), ("9000/udp", "127.0.0.1"),
+                             ("9000/tcp", "::"), ("9000/tcp", "")):
+            with self.subTest(protocol=protocol, ip=ip):
+                self.assertFalse(acceptance.docker_endpoint_matches({protocol: [{"HostPort": "19000", "HostIp": ip}]}, endpoint))
+        self.assertTrue(acceptance.docker_endpoint_matches({"9000/tcp": [{"HostPort": "19000", "HostIp": "::"}]}, "http://[::1]:19000"))
+        def table(ip, inode="123"):
+            packed = acceptance.ipaddress.ip_address(ip).packed
+            encoded = "".join(f"{int.from_bytes(packed[index:index+4], sys.byteorder):08X}" for index in range(0, len(packed), 4))
+            return f"header\n0: {encoded}:4A38 00000000:0000 0A 0 0 0 0 0 {inode}\n"
+        self.assertTrue(acceptance.native_endpoint_matches({"tcp": table("127.0.0.1")}, {"123"}, endpoint))
+        self.assertFalse(acceptance.native_endpoint_matches({"tcp": table("127.0.0.2")}, {"123"}, endpoint))
+        competing = table("127.0.0.1") + table("127.0.0.1", "456").split("\n", 1)[1]
+        self.assertFalse(acceptance.native_endpoint_matches({"tcp": competing}, {"123"}, endpoint))
+        self.assertTrue(acceptance.native_endpoint_matches({"tcp6": table("::1")}, {"123"}, "http://[::1]:19000"))
+        self.assertFalse(acceptance.native_endpoint_matches({"tcp": table("127.0.0.1"), "tcp6": table("::", "456")}, {"123"}, endpoint))
+
+    def test_evidence_symlink_and_hardlink_never_mutate_external_file(self):
+        target = self.root / "external"
+        target.write_bytes(b"unchanged external data")
+        target.chmod(0o600)
+        output = self.root / "observations.jsonl"
+        for link in (lambda: output.symlink_to(target), lambda: os.link(target, output)):
+            link()
+            with self.assertRaises(acceptance.AcceptanceError):
+                self.evidence.append("must-not-write")
+            with self.assertRaises(acceptance.AcceptanceError):
+                acceptance.summarize(self.evidence)
+            self.assertEqual(target.read_bytes(), b"unchanged external data")
+            self.assertEqual(target.stat().st_mode & 0o777, 0o600)
+            output.unlink()
+
+    def test_evidence_fifo_rejection_is_bounded_before_type_check(self):
+        fifo = self.root / "observations.jsonl"
+        os.mkfifo(fifo, 0o600)
+        program = """import sys
+from pathlib import Path
+import scanner_memory_acceptance as acceptance
+try:
+    with acceptance.open_owned(Path(sys.argv[1]), sys.argv[2]):
+        pass
+except acceptance.AcceptanceError:
+    sys.exit(0)
+sys.exit(1)
+"""
+        for mode in ("r", "a"):
+            with self.subTest(mode=mode):
+                result = subprocess.run([sys.executable, "-c", program, str(fifo), mode],
+                                        cwd=Path(__file__).resolve().parent, capture_output=True, timeout=5, check=False)
+                self.assertEqual(result.returncode, 0, "FIFO input must be rejected without waiting for another endpoint")
+
+    def test_last_load_batch_cannot_succeed_after_deadline(self):
+        clock = SimpleNamespace(now=0.0)
+        last_key = acceptance.object_key(self.plan, 6)
+        def put(**kwargs):
+            if kwargs["Key"] == last_key:
+                clock.now = 2.0
+        api = SimpleNamespace(owned=lambda **_: None, status=lambda _: {"enabled": False, "metrics": {}},
+                              client=SimpleNamespace(put_object=put),
+                              verify=lambda **_: self.fail("expired phase must not start verification"))
+        runtime = SimpleNamespace(sample=lambda: {"identity_sha256": "a", "storage_available_bytes": 1 << 50})
+        args = SimpleNamespace(seconds=1, interval=300, workers=1, batch=7)
+        with patch.object(acceptance.time, "monotonic", side_effect=lambda: clock.now), self.assertRaises(acceptance.AcceptanceError):
+            acceptance.run_load(self.evidence, api, runtime, args)
+        self.assertEqual(self.evidence.state()["next_object"], 7, "completed work remains resumable")
+        self.assertNotIn("load_complete", self.evidence.state())
+
+    def test_final_verification_cannot_switch_process_identity(self):
+        identity = SimpleNamespace(value="old")
+        runtime = SimpleNamespace(sample=lambda: {"identity_sha256": identity.value})
+        api = SimpleNamespace(status=lambda _: {"enabled": True, "metrics": {}},
+                              verify=lambda **_: setattr(identity, "value", "new"))
+        args = SimpleNamespace(seconds=10, interval=300)
+        with self.assertRaises(acceptance.AcceptanceError):
+            with acceptance.Phase(self.evidence, runtime, api, "probe", True, args) as phase:
+                phase.verify()
+
+    def test_initial_status_restart_is_rejected_before_load_writes(self):
+        identity = SimpleNamespace(value="before-status")
+        runtime = SimpleNamespace(sample=lambda: {
+            "identity_sha256": identity.value, "storage_available_bytes": 1 << 50})
+        def status(_):
+            identity.value = "after-status"
+            return {"enabled": False, "metrics": {}}
+        api = SimpleNamespace(status=status, owned=Mock(), client=Mock(), verify=lambda **_: {})
+        args = SimpleNamespace(seconds=10, interval=300, workers=1, batch=7)
+        with self.assertRaises(acceptance.AcceptanceError):
+            acceptance.run_load(self.evidence, api, runtime, args)
+        api.owned.assert_not_called()
+        api.client.put_object.assert_not_called()
+
+    def test_completed_burst_resume_reports_zero_new_requests(self):
+        self.evidence.save({"plan_sha256": self.manifest["plan_sha256"], "next_object": 7, "load_complete": True,
+                            "burst": {"id": str(uuid.uuid4()), "next": 7, "objects": 7}})
+        runtime = SimpleNamespace(sample=lambda: {"identity_sha256": "a", "storage_available_bytes": 1 << 50})
+        api = SimpleNamespace(owned=lambda: None, status=lambda _: {"enabled": False, "metrics": {}},
+                              verify=lambda **_: {"objects": 7})
+        args = SimpleNamespace(seconds=10, interval=300, workers=1, batch=7, burst_objects=7, idle_seconds=0.001)
+        result = acceptance.run_burst(self.evidence, api, runtime, args)
+        self.assertEqual(result["newly_completed_objects"], 0)
+        self.assertEqual(result["new_logical_copy_head_delete_requests"], 0)
+        self.assertEqual(result["cumulative_logical_copy_head_delete_requests"], 21)
+        self.assertEqual(result["cursor_start"], result["cursor_end"])
+
+    def test_summary_never_splices_retries_or_processes_into_one_memory_trend(self):
+        for phase_id, process, rss in (("attempt-a", "process-a", 100), ("attempt-b", "process-b", 10)):
+            self.evidence.append("sample", phase="burst", phase_id=phase_id, stage="idle",
+                                 memory={"identity_sha256": process, "rss_bytes": rss, "cgroup_current_bytes": rss * 2,
+                                         "cgroup_peak_bytes": rss * 3, "active_heap_bytes": None})
+        groups = acceptance.summarize(self.evidence)["phases"]
+        self.assertEqual(len(groups), 2)
+        for group in groups.values():
+            self.assertEqual(group["first_rss"], group["last_rss"])
+            self.assertEqual(group["sampled_current_peak_bytes"], group["first_rss"] * 2)
+            self.assertEqual(group["kernel_reported_peak_bytes"], group["first_rss"] * 3)
+
+    def test_signed_observations_never_follow_redirects(self):
+        with self.assertRaises(acceptance.AcceptanceError):
+            acceptance.NoRedirect().redirect_request(None, None, 302, "redirect", {}, "https://foreign.example")
+
+    def test_admin_signer_includes_s3_payload_hash(self):
+        from botocore.credentials import Credentials
+        from types import SimpleNamespace
+        api = acceptance.S3.__new__(acceptance.S3)
+        api.plan = self.plan
+        api.session = SimpleNamespace(get_credentials=lambda: Credentials("test-access", "test-secret"))
+        observed = []
+        def open_request(request, **_):
+            observed.append({key.lower(): value for key, value in request.header_items()})
+            return io.BytesIO(b"{}")
+        opener = SimpleNamespace(open=open_request)
+        with patch("urllib.request.build_opener", return_value=opener):
+            self.assertEqual(api.admin("/rustfs/admin/v3/scanner/status"), {})
+        self.assertEqual(observed[0]["x-amz-content-sha256"], acceptance.hashlib.sha256(b"").hexdigest())
+        self.assertIn("x-amz-content-sha256", observed[0]["authorization"].lower())
+
+    def test_capacity_guard_is_fail_closed(self):
+        with patch.object(acceptance.shutil, "disk_usage", return_value=acceptance.shutil._ntuple_diskusage(10, 9, 1)):
+            with self.assertRaises(acceptance.AcceptanceError):
+                acceptance.free_space(self.plan, 1)
+
+    def test_only_loopback_endpoints_without_embedded_credentials(self):
+        for endpoint in ("http://localhost:9000", "http://remote.example:9000", "http://user:secret@localhost:9000", "http://localhost:9000/?token=secret"):
+            with self.assertRaises(acceptance.AcceptanceError):
+                acceptance.local_url(endpoint)
+
+
+if __name__ == "__main__":
+    unittest.main()