chore: restore main's Cargo and baseline files after merging main

* fix(ci): run the mint lane on ubuntu-latest, which has a Docker daemon

* fix(docker): let BuildKit populate TARGETARCH for the rc download

* fix(docker): derive the rc architecture when TARGETARCH is not set

.config/e2e-smoke-selection.txt

@@ -1 +1 @@
-sha256=fbabc660e6e01fd8ff6b9d1de7d68e58242d60c942f85aed5074b0ec2918c154
+sha256=21e9a1561ac20382503c38353bbb9b830658534a4f277c254a42d0afc6ebed8b

.config/test-count-baseline.txt

@@ -25,6 +25,7 @@ rustfs-data-usage	58	0
 rustfs-ecstore	7858	2
 rustfs-extension-schema	23	0
 rustfs-filemeta	361	0
+rustfs-gateway-adapter	74	0
 rustfs-heal	1128	3
 rustfs-heal-contracts	8	0
 rustfs-iam	415	0

.github/scheduled-validations.json

@@ -1021,6 +1021,26 @@
       "policy_reason": "Preserve operator-disabled state; enabling requires a separate policy decision.",
       "origin": "baseline"
     },
+    {
+      "workflow": ".github/workflows/manual-main-packages.yml",
+      "name": "Manual main packages",
+      "roles": [
+        "manual"
+      ],
+      "events": [
+        "workflow_dispatch"
+      ],
+      "cadence": "event",
+      "timezone": "UTC",
+      "schedule": [],
+      "owner": "@overtrue",
+      "policy": "active",
+      "monitor": "none",
+      "result_owner": "manual-main-packages.yml",
+      "disposition": "Packages the origin/main tip as workflow artifacts only. No GitHub Release or download-channel writes.",
+      "monitor_reason": "Operator-requested packaging has no periodic freshness obligation.",
+      "origin": "manual-main-packages"
+    },
     {
       "workflow": ".github/workflows/minio-interop.yml",
       "name": "minio-interop",

.github/workflows/manual-main-packages.yml

@@ -0,0 +1,415 @@
+# Copyright 2026 RustFS Team
+# SPDX-License-Identifier: Apache-2.0
+
+# Manual packaging of the current origin/main tip. Workflow artifacts only.
+# Dispatch uses this file from the selected branch. The build fetches
+# refs/heads/main and checks out that commit, not the dispatch ref or a tag.
+
+name: Manual main packages
+
+on:
+  workflow_dispatch:
+    inputs:
+      architecture:
+        description: "Linux targets: x86 (x86_64 musl+gnu), arm (aarch64 musl+gnu), or all"
+        required: true
+        type: choice
+        options:
+          - x86
+          - arm
+          - all
+        default: all
+      version:
+        description: "Binary and package version: MAJOR.MINOR.PATCH, or MAJOR.MINOR.PATCH-(alpha|beta|rc).N. No leading v."
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+env:
+  CARGO_TERM_COLOR: always
+  RUST_BACKTRACE: 1
+  CARGO_INCREMENTAL: 0
+
+jobs:
+  prepare:
+    name: Resolve main tip and version
+    if: github.repository == 'rustfs/rustfs'
+    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    outputs:
+      source_sha: ${{ steps.pin.outputs.source_sha }}
+      version: ${{ steps.version.outputs.version }}
+      toolchain: ${{ steps.inputs.outputs.toolchain }}
+      console: ${{ steps.inputs.outputs.console }}
+      matrix: ${{ steps.matrix.outputs.matrix }}
+    steps:
+      - name: Checkout main
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: main
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Pin the tip of origin/main
+        id: pin
+        shell: bash
+        run: |
+          set -euo pipefail
+          GIT_TERMINAL_PROMPT=0 git fetch --no-tags origin refs/heads/main
+          sha=$(git rev-parse FETCH_HEAD)
+          git checkout --detach "$sha"
+          test "$(git rev-parse HEAD)" = "$sha"
+          printf 'source_sha=%s\n' "$sha" >> "$GITHUB_OUTPUT"
+          printf 'Pinned origin/main %s\n' "$sha"
+
+      - name: Validate package version
+        id: version
+        shell: bash
+        env:
+          PACKAGE_VERSION: ${{ inputs.version }}
+        run: |
+          set -euo pipefail
+          bash scripts/release/package_manual_artifacts.sh validate \
+            --version "$PACKAGE_VERSION" \
+            --arch x86_64 \
+            --libc gnu | tee "$RUNNER_TEMP/package-version.txt"
+          printf 'version=%s\n' "$PACKAGE_VERSION" >> "$GITHUB_OUTPUT"
+
+      - name: Install stable Rust
+        uses: dtolnay/rust-toolchain@29eef336d9b2848a0b548edc03f92a220660cdb8 # stable
+        with:
+          toolchain: stable
+
+      - name: Resolve console assets and toolchain
+        id: inputs
+        shell: bash
+        env:
+          GH_TOKEN: ${{ github.token }}
+        run: |
+          set -euo pipefail
+          toolchain=$(rustc --version | awk '{print $2}')
+          [[ "$toolchain" =~ ^[0-9]+\.[0-9]+\.[0-9]+$ ]]
+          gh api repos/rustfs/console/releases/latest > console-release.json
+          TOOLCHAIN="$toolchain" python3 - <<'PY'
+          import json, os, sys
+          sys.path.insert(0, "scripts/release")
+          import candidate
+          console = candidate.console_asset(candidate.read_json("console-release.json"))
+          candidate.emit_output("console", json.dumps(console, separators=(",", ":")))
+          candidate.emit_output("toolchain", os.environ["TOOLCHAIN"])
+          print(f"toolchain={os.environ['TOOLCHAIN']}")
+          print(f"console={console['tag']} {console['name']}")
+          PY
+
+      - name: Select Linux targets
+        id: matrix
+        shell: bash
+        env:
+          ARCHITECTURE: ${{ inputs.architecture }}
+        run: |
+          set -euo pipefail
+          matrix=$(bash scripts/release/package_manual_artifacts.sh matrix "$ARCHITECTURE")
+          count=$(python3 -c 'import json,sys; print(len(json.loads(sys.argv[1])["include"]))' "$matrix")
+          [[ "$count" -gt 0 ]]
+          printf 'matrix=%s\n' "$matrix" >> "$GITHUB_OUTPUT"
+          printf 'Selected %s target(s) for %s\n' "$count" "$ARCHITECTURE"
+
+  package:
+    name: Package RustFS (${{ matrix.target_id }})
+    needs: prepare
+    if: github.repository == 'rustfs/rustfs'
+    # x86_64 uses sm-standard-4, the official release compile pool: the workspace
+    # release build needs that disk (placement rule: disk) and reads build.yml's
+    # cache without writing it. aarch64 uses ubicloud-standard-16-arm so the
+    # packaged binary and console smoke execute natively (placement rule: hardware).
+    runs-on: ${{ matrix.os }}
+    timeout-minutes: 210
+    env:
+      FORCE_JAVASCRIPT_ACTIONS_TO_NODE24: "true"
+      RUSTUP_TOOLCHAIN: ${{ needs.prepare.outputs.toolchain }}
+      RUSTFLAGS: "${{ matrix.rustflags }}"
+    strategy:
+      fail-fast: false
+      matrix: ${{ fromJson(needs.prepare.outputs.matrix) }}
+    steps:
+      - name: Checkout the pinned origin/main commit
+        uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7
+        with:
+          ref: ${{ needs.prepare.outputs.source_sha }}
+          fetch-depth: 0
+          persist-credentials: false
+
+      - name: Confirm the checkout is that main commit
+        shell: bash
+        env:
+          SOURCE_SHA: ${{ needs.prepare.outputs.source_sha }}
+        run: |
+          set -euo pipefail
+          test "$(git rev-parse HEAD)" = "$SOURCE_SHA"
+          GIT_TERMINAL_PROMPT=0 git fetch --no-tags origin refs/heads/main
+          git merge-base --is-ancestor "$SOURCE_SHA" FETCH_HEAD
+
+      - name: Setup Rust environment
+        uses: ./.github/actions/setup
+        with:
+          rust-version: ${{ needs.prepare.outputs.toolchain }}
+          target: ${{ matrix.target }}
+          cache-shared-key: build-${{ matrix.target }}${{ matrix.glibc && format('-glibc-{0}', matrix.glibc) || '' }}${{ matrix.os == 'ubicloud-standard-16-arm' && '-ubicloud-arm' || '' }}
+          # build.yml owns this key. This job only restores it.
+          cache-save-if: 'false'
+          install-cross-tools: ${{ matrix.cross }}
+          install-build-packaging-tools: 'true'
+          install-test-tools: 'false'
+
+      - name: Download static console assets
+        shell: bash
+        env:
+          GH_TOKEN: ${{ github.token }}
+          CONSOLE_IDENTITY: ${{ needs.prepare.outputs.console }}
+        run: |
+          set -euo pipefail
+          python3 - <<'PYCONSOLE'
+          import json, os, pathlib, subprocess, zipfile
+          value = json.loads(os.environ["CONSOLE_IDENTITY"])
+          subprocess.run(["python3", ".github/actions/setup/download_release_asset.py",
+              "--repository", "rustfs/console", "--tag", value["tag"], "--asset", value["name"],
+              "--sha256", value["sha256"], "--output", "console.zip"], check=True)
+          destination = pathlib.Path("rustfs/static")
+          destination.mkdir(parents=True, exist_ok=True)
+          with zipfile.ZipFile("console.zip") as archive:
+              for member in archive.infolist():
+                  path = pathlib.PurePosixPath(member.filename)
+                  if path.is_absolute() or ".." in path.parts or "\\" in member.filename:
+                      raise SystemExit("unsafe Console archive member")
+              archive.extractall(destination)
+          if not (destination / "index.html").is_file():
+              raise SystemExit("Console archive has no index.html")
+          PYCONSOLE
+          rm -f console.zip
+
+      - name: Build RustFS
+        shell: bash
+        env:
+          RUSTFS_BUILD_VERSION: ${{ needs.prepare.outputs.version }}
+          BUILD_TARGET: ${{ matrix.target }}
+          BUILD_OS: ${{ matrix.build_os }}
+          BUILD_FEATURES: ${{ matrix.features }}
+          BUILD_GLIBC: ${{ matrix.glibc }}
+          BUILD_CROSS: ${{ matrix.cross }}
+        run: |
+          set -euo pipefail
+          touch rustfs/build.rs
+
+          feature_args=()
+          if [[ -n "$BUILD_FEATURES" ]]; then
+            feature_args=(--features "$BUILD_FEATURES")
+          fi
+
+          # gnu targets pin glibc 2.31 through cargo-zigbuild (<target>.<glibc>).
+          target_arg="$BUILD_TARGET"
+          if [[ -n "$BUILD_GLIBC" ]]; then
+            target_arg="${target_arg}.${BUILD_GLIBC}"
+          fi
+
+          profile_args=()
+          if [[ "$BUILD_TARGET" == *-linux-gnu && "$BUILD_FEATURES" == pyroscope ]]; then
+            profile_args=(--config 'profile.release.strip="none"')
+          fi
+
+          case "$BUILD_CROSS" in
+            true)
+              cargo zigbuild --release --locked --target "$target_arg" "${feature_args[@]}" "${profile_args[@]}" -p rustfs --bin rustfs
+              ;;
+            false)
+              cargo build --release --locked --target "$target_arg" "${feature_args[@]}" "${profile_args[@]}" -p rustfs --bin rustfs
+              ;;
+            *)
+              printf 'unexpected cross flag: %q\n' "$BUILD_CROSS" >&2
+              exit 1
+              ;;
+          esac
+
+          bin="target/${BUILD_TARGET}/release/rustfs"
+          version_output=$("$bin" --version)
+          first=${version_output%%$'\n'*}
+          if [[ "$first" != "rustfs ${RUSTFS_BUILD_VERSION}" ]]; then
+            printf 'binary version line is %q, expected %q\n' "$first" "rustfs ${RUSTFS_BUILD_VERSION}" >&2
+            printf '%s\n' "$version_output" >&2
+            exit 1
+          fi
+          if ! grep -Fxq "build target : ${BUILD_TARGET}" <<< "$version_output" || \
+             ! grep -Fxq "build os     : ${BUILD_OS}" <<< "$version_output" || \
+             ! grep -Fxq "rust channel : stable-${BUILD_TARGET}" <<< "$version_output"; then
+            printf 'release binary reports an unexpected compile target\n' >&2
+            printf '%s\n' "$version_output" >&2
+            exit 1
+          fi
+          if [[ -n "$BUILD_GLIBC" ]]; then
+            python3 scripts/release/check_glibc.py "$BUILD_GLIBC" "$bin"
+          fi
+
+      - name: Generate the CPU symbol catalog
+        if: endsWith(matrix.target, '-linux-gnu') && matrix.features == 'pyroscope'
+        shell: bash
+        env:
+          BUILD_TARGET: ${{ matrix.target }}
+        run: |
+          set -euo pipefail
+          cargo metadata --locked --format-version 1 --filter-platform "$BUILD_TARGET" \
+            --features rustfs/pyroscope > "$RUNNER_TEMP/cpu-catalog-metadata.json"
+          features=$(jq -ce '
+            (.packages[] | select(.name == "rustfs") | .id) as $id |
+            .resolve.nodes[] | select(.id == $id) | .features | map(select(. != "default")) | sort
+          ' "$RUNNER_TEMP/cpu-catalog-metadata.json")
+          release_dir="target/${BUILD_TARGET}/release"
+          rm -f "$release_dir/rustfs.cpu-symbol-catalog.json" "$release_dir/rustfs.cpu-symbol-catalog.sha256"
+          cargo run --locked -p rustfs-connect-symbol-catalog -- \
+            "$release_dir/rustfs" "$(git rev-parse HEAD)" "$BUILD_TARGET" "$features" \
+            "$release_dir/rustfs.cpu-symbol-catalog.json" > "$release_dir/rustfs.cpu-symbol-catalog.sha256"
+
+      - name: Install DEB and RPM tools
+        shell: bash
+        run: |
+          set -euo pipefail
+          sudo apt-get update
+          sudo apt-get install -y ruby ruby-dev build-essential rpm fakeroot cpio
+          sudo gem install fpm --version 1.17.0 --no-document
+
+      - name: Package zip, deb, and rpm
+        shell: bash
+        env:
+          PACKAGE_VERSION: ${{ needs.prepare.outputs.version }}
+          ARCH: ${{ matrix.arch }}
+          LIBC: ${{ matrix.libc }}
+          BUILD_TARGET: ${{ matrix.target }}
+          BUILD_FEATURES: ${{ matrix.features }}
+          SOURCE_SHA: ${{ needs.prepare.outputs.source_sha }}
+        run: |
+          set -euo pipefail
+          binary="target/${BUILD_TARGET}/release/rustfs"
+          extra=()
+          if [[ "$BUILD_TARGET" == *-linux-gnu && "$BUILD_FEATURES" == pyroscope ]]; then
+            extra=(
+              --zip-extra "target/${BUILD_TARGET}/release/rustfs.cpu-symbol-catalog.json"
+              --zip-extra "target/${BUILD_TARGET}/release/rustfs.cpu-symbol-catalog.sha256"
+            )
+          fi
+          bash scripts/release/package_manual_artifacts.sh package \
+            --version "$PACKAGE_VERSION" \
+            --arch "$ARCH" \
+            --libc "$LIBC" \
+            --binary "$binary" \
+            --output dist \
+            "${extra[@]}"
+          printf 'Packaged %s from origin/main %s\n' "$PACKAGE_VERSION" "$SOURCE_SHA"
+
+      - name: Verify the packaged binary
+        shell: bash
+        env:
+          PACKAGE_VERSION: ${{ needs.prepare.outputs.version }}
+          ARCH: ${{ matrix.arch }}
+          LIBC: ${{ matrix.libc }}
+          BUILD_TARGET: ${{ matrix.target }}
+          BUILD_OS: ${{ matrix.build_os }}
+          BUILD_FEATURES: ${{ matrix.features }}
+        run: |
+          set -euo pipefail
+          stem="rustfs-linux-${ARCH}-${LIBC}-v${PACKAGE_VERSION}"
+          package_file="dist/${stem}.zip"
+          [[ -f "$package_file" && -f "dist/${stem}.deb" && -f "dist/${stem}.rpm" ]]
+          [[ "$(find dist -maxdepth 1 -type f | wc -l)" -eq 3 ]]
+          archive_members=$(unzip -Z1 "$package_file" | sort)
+          if [[ "$BUILD_TARGET" == *-linux-gnu && "$BUILD_FEATURES" == pyroscope ]]; then
+            expected_members=$(printf '%s\n' rustfs rustfs.cpu-symbol-catalog.json rustfs.cpu-symbol-catalog.sha256 | sort)
+          else
+            expected_members=$(printf '%s\n' rustfs)
+          fi
+          if [[ "$archive_members" != "$expected_members" ]]; then
+            printf 'Linux package does not contain the expected files\n' >&2
+            printf '%s\n' "$archive_members" >&2
+            exit 1
+          fi
+          binary_dir=$(mktemp -d)
+          trap 'rm -rf "$binary_dir"' EXIT
+          unzip -q "$package_file" rustfs -d "$binary_dir"
+          version_output=$("$binary_dir/rustfs" --version)
+          first=${version_output%%$'\n'*}
+          if [[ "$first" != "rustfs ${PACKAGE_VERSION}" ]]; then
+            printf 'packaged binary version line is %q, expected %q\n' "$first" "rustfs ${PACKAGE_VERSION}" >&2
+            exit 1
+          fi
+          if ! grep -Fxq "build target : ${BUILD_TARGET}" <<< "$version_output" || \
+             ! grep -Fxq "build os     : ${BUILD_OS}" <<< "$version_output"; then
+            printf 'packaged binary reports an unexpected compile target\n' >&2
+            exit 1
+          fi
+
+      - name: Verify packaged console
+        if: matrix.target == 'x86_64-unknown-linux-gnu' || matrix.target == 'aarch64-unknown-linux-gnu'
+        shell: bash
+        env:
+          PACKAGE_VERSION: ${{ needs.prepare.outputs.version }}
+          ARCH: ${{ matrix.arch }}
+          LIBC: ${{ matrix.libc }}
+        run: |
+          set -euo pipefail
+          package_file="dist/rustfs-linux-${ARCH}-${LIBC}-v${PACKAGE_VERSION}.zip"
+
+          find_free_port() {
+            python3 - <<'PY'
+          import socket
+
+          with socket.socket() as sock:
+              sock.bind(("127.0.0.1", 0))
+              print(sock.getsockname()[1])
+          PY
+          }
+
+          package_dir=$(mktemp -d)
+          data_dir=$(mktemp -d)
+          log_file="${RUNNER_TEMP}/rustfs-console-smoke.log"
+          server_pid=""
+          trap '
+            if [[ -n "${server_pid:-}" ]]; then
+              kill "$server_pid" 2>/dev/null || true
+              wait "$server_pid" 2>/dev/null || true
+            fi
+            rm -rf "$package_dir" "$data_dir"
+          ' EXIT
+
+          unzip -q "$package_file" -d "$package_dir"
+          api_port=$(find_free_port)
+          console_port=$(find_free_port)
+          console_url="http://127.0.0.1:${console_port}/rustfs/console/"
+
+          "$package_dir/rustfs" server \
+            --address "127.0.0.1:${api_port}" \
+            --console-enable \
+            --console-address "127.0.0.1:${console_port}" \
+            --access-key console-smoke \
+            --secret-key console-smoke-secret \
+            "$data_dir" >"$log_file" 2>&1 &
+          server_pid=$!
+
+          for _ in {1..40}; do
+            response=$(curl --silent --output /dev/null --write-out '%{http_code} %{content_type}' "$console_url" || true)
+            if [[ "$response" == 200\ text/html* ]]; then
+              exit 0
+            fi
+            sleep 0.25
+          done
+
+          printf 'Console endpoint did not return 200 text/html: %s\n' "$response" >&2
+          cat "$log_file"
+          exit 1
+
+      - name: Upload packages
+        uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1
+        with:
+          name: packages-${{ matrix.target_id }}-v${{ needs.prepare.outputs.version }}
+          path: dist/*
+          if-no-files-found: error
+          compression-level: 0
+          retention-days: 30

.github/workflows/mint.yml

@@ -112,9 +112,11 @@ jobs:
     # GitHub-hosted: reliably provides a running Docker daemon + buildx +
     # python3. See the header note (ci-2) for why the self-hosted sm-standard-4
     # label was abandoned (no docker.sock on the pod its only run landed on).
+    # sm-standard-2 has no docker.sock either (weekly runs failed at "Enable
+    # buildx" from 2026-09-27 to 2026-11-08), so keep this job on ubuntu-latest.
     # TODO(ci-8): scheduled-failure alerting (auto-open issue on a red weekly
     # run) is added by the ci-8 composite action; do not implement it here.
-    runs-on: sm-standard-2
+    runs-on: ubuntu-latest
     timeout-minutes: 120
     steps:
       - uses: actions/checkout@f548e57e544e1ff5a4c46bf1e1b8685f8e4a348a # v7

Cargo.lock

@@ -4698,6 +4698,12 @@ version = "0.3.34"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "cd417de3d1d015fc3bfd2b1ea46dfc7bab72ef86f1cc7cc9c78e728b34a6d1fd"
 
+[[package]]
+name = "futures-timer"
+version = "3.0.4"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "af43fadb8a98512d547e37b4e92e0ced13e205c061b87b4623eff01d918d6968"
+
 [[package]]
 name = "futures-util"
 version = "0.3.34"
@@ -7914,10 +7920,34 @@ version = "0.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "010378780309880b08997fae13be7834dba947d36393bd372f2b1556deb2a2f6"
 dependencies = [
+ "phf_macros",
  "phf_shared 0.14.0",
  "serde",
 ]
 
+[[package]]
+name = "phf_generator"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "aeb62e0959d5a1bebc965f4d15d9e2b7cea002b6b0f5ba8cde6cc26738467100"
+dependencies = [
+ "fastrand",
+ "phf_shared 0.14.0",
+]
+
+[[package]]
+name = "phf_macros"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5fa8d0ca26d424d27630da600c6624696e7dec8bf7b3b492b383c5dc49e5e085"
+dependencies = [
+ "phf_generator",
+ "phf_shared 0.14.0",
+ "proc-macro2",
+ "quote",
+ "syn 2.0.119",
+]
+
 [[package]]
 name = "phf_shared"
 version = "0.12.1"
@@ -9999,6 +10029,114 @@ dependencies = [
  "xxhash-rust",
 ]
 
+[[package]]
+name = "rustfs-gateway"
+version = "0.57.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "arc-swap",
+ "bytes",
+ "form_urlencoded",
+ "futures-timer",
+ "futures-util",
+ "http 1.5.0",
+ "http-body",
+ "http-body-util",
+ "httparse",
+ "hyper",
+ "pin-project-lite",
+ "rustfs-gateway-core",
+ "rustfs-gateway-http",
+ "rustfs-gateway-macros",
+ "rustfs-gateway-sig",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "rustfs-gateway-xml",
+ "sha2 0.11.1",
+ "tokio",
+ "tower",
+ "tracing",
+ "url",
+]
+
+[[package]]
+name = "rustfs-gateway-adapter"
+version = "1.0.5"
+dependencies = [
+ "http 1.5.0",
+ "rustfs-credentials",
+ "rustfs-gateway",
+ "rustfs-gateway-http",
+ "rustfs-iam",
+ "rustfs-policy",
+ "serde_json",
+ "time",
+ "tokio",
+]
+
+[[package]]
+name = "rustfs-gateway-core"
+version = "0.54.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "bytes",
+ "http 1.5.0",
+ "md-5 0.11.0",
+ "percent-encoding",
+ "rustfs-gateway-http",
+ "rustfs-gateway-sig",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "rustfs-gateway-xml",
+ "serde",
+ "serde_json",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustfs-gateway-http"
+version = "0.1.4"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "hmac 0.13.0",
+ "http 1.5.0",
+ "httparse",
+ "memchr",
+ "phf 0.14.0",
+ "rustfs-gateway-stream",
+ "rustfs-gateway-types",
+ "sha2 0.11.1",
+ "smallvec",
+ "subtle",
+ "zeroize",
+]
+
+[[package]]
+name = "rustfs-gateway-macros"
+version = "0.1.4"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 3.0.9",
+]
+
+[[package]]
+name = "rustfs-gateway-sig"
+version = "0.13.0"
+source = "git+https://github.com/rustfs/gateway.git?rev=21283e874c2381bab611e9347acbc5b92e9e2b3c#21283e874c2381bab611e9347acbc5b92e9e2b3c"
+dependencies = [
+ "hmac 0.13.0",
+ "http 1.5.0",
+ "rustfs-gateway-http",
+ "sha1 0.11.0",
+ "sha2 0.11.1",
+ "smallvec",
+ "subtle",
+ "zeroize",
+]
+
 [[package]]
 name = "rustfs-gateway-stream"
 version = "0.2.0"
@@ -12603,10 +12741,10 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
 dependencies = [
  "fastrand",
- "getrandom 0.4.3",
+ "getrandom 0.3.4",
  "once_cell",
  "rustix",
- "windows-sys 0.61.2",
+ "windows-sys 0.52.0",
 ]
 
 [[package]]

Cargo.toml

@@ -26,6 +26,7 @@ members = [
     "crates/ecstore", # Erasure coding storage implementation
     "crates/e2e_test", # End-to-end test suite
     "crates/filemeta", # File metadata management
+    "crates/gateway-adapter", # RustFS implementations of the gateway extension traits
     "crates/heal", # Erasure set and object healing
     "crates/heal-contracts", # Heal request/response channel contracts
     "crates/iam", # Identity and Access Management
@@ -93,6 +94,7 @@ redundant_clone = "warn"
 [workspace.dependencies]
 # RustFS Internal Crates
 rustfs = { path = "./rustfs", version = "1.0.5" }
+rustfs-gateway-adapter = { path = "crates/gateway-adapter", version = "1.0.5" }
 rustfs-heal = { path = "crates/heal", version = "1.0.5" }
 rustfs-heal-contracts = { path = "crates/heal-contracts", version = "1.0.5" }
 rustfs-scanner-metrics = { path = "crates/scanner-metrics", version = "1.0.5" }
@@ -149,6 +151,12 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.5" }
 # the s3s-facing edge of the rustfs binary.
 rustfs-gateway-types = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.32.0", default-features = false }
 rustfs-gateway-stream = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.2.0", default-features = false }
+# The gateway framework itself, consumed for its extension traits by the ring-2
+# adapter. `default-features = false` keeps the `server` feature, and with it the
+# server crate, out of the build.
+rustfs-gateway = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", version = "0.57.0", default-features = false }
+# Test-only: builds the `EffectiveHost` a host-resolution test queries with.
+rustfs-gateway-http = { git = "https://github.com/rustfs/gateway.git", rev = "21283e874c2381bab611e9347acbc5b92e9e2b3c", default-features = false }
 
 # Async Runtime and Networking
 async-channel = "2.5.0"

Dockerfile.source

@@ -24,9 +24,10 @@
 #
 # Supports cross-compilation for amd64 and arm64 via TARGETPLATFORM.
 
-ARG TARGETPLATFORM
-ARG BUILDPLATFORM
-ARG TARGETARCH
+# Do not declare TARGETPLATFORM, BUILDPLATFORM or TARGETARCH here: a global
+# declaration without a value shadows BuildKit's automatic values, so the
+# stage-level re-declarations below see empty strings. They are only
+# declared inside the stage that uses them.
 ARG RUSTFS_BUILD_FEATURES=""
 ARG RC_RELEASE=latest
 
@@ -169,7 +170,24 @@ RUN --mount=type=cache,target=/usr/local/cargo/registry \
     ;; \
     esac
 
-RUN sh /usr/src/rustfs/scripts/download_rc.sh "$TARGETARCH" musl "$RC_RELEASE" /usr/local/bin/rc
+# TARGETARCH is only provided by BuildKit. Fall back to TARGETPLATFORM (for
+# example set by docker-compose) and then to the build host, like the cargo
+# step above, so every path that builds this file passes a real architecture.
+RUN set -eu; \
+    rc_arch="${TARGETARCH:-}"; \
+    if [ -z "${rc_arch}" ]; then \
+    case "${TARGETPLATFORM:-}" in \
+    linux/amd64) rc_arch=amd64 ;; \
+    linux/arm64) rc_arch=arm64 ;; \
+    *) \
+    case "$(uname -m)" in \
+    x86_64) rc_arch=amd64 ;; \
+    aarch64|arm64) rc_arch=arm64 ;; \
+    *) echo "TARGETARCH is empty and TARGETPLATFORM='${TARGETPLATFORM:-}' is not linux/amd64 or linux/arm64; cannot choose an rc architecture on $(uname -m)" >&2; exit 1 ;; \
+    esac ;; \
+    esac; \
+    fi; \
+    sh /usr/src/rustfs/scripts/download_rc.sh "${rc_arch}" musl "${RC_RELEASE}" /usr/local/bin/rc
 
 # -----------------------------
 # Development stage (keeps toolchain)

crates/e2e_test/README.md

@@ -29,6 +29,7 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern:
 | **distributed 4×4** | [`src/distributed/`](src/distributed) | Nightly `e2e-distributed` lane: S3, object lock/WORM, versioning, bucket/site replication, quota, expand/decommission/rebalance, concurrency, chaos, 4-node upgrade of historical data and IAM AK/SK. Map: [`docs/testing/distributed-e2e.md`](../../docs/testing/distributed-e2e.md) |
 | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup |
 | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory |
+| **dual-stack differential** | `dual_stack_differential` | Two single-node servers answer one SDK script for all 73 S3 operations and every differing `(op, field, a, b)` is reported. Side A is the binary under test on `RUSTFS_S3_STACK=legacy`; side B runs `RUSTFS_E2E_STACK_B`. Set `RUSTFS_E2E_REFERENCE_BINARY` to a released `rustfs` binary to make side A that binary instead (started without `RUSTFS_S3_STACK`, which older releases do not know). A missing, non-file or non-executable path fails the run; it never skips or falls back to the same-binary mode. The report in `<target>/dual-stack/<run>.json` records `reference_side` and each side's `--version` output |
 
 The external-tool `storage_metric_ownership_test` validates the OTLP/Collector/Prometheus path, including a rolling upgrade and node failures. See the [storage metrics guide](../../docs/operations/storage-metrics.md) for its required binaries and focused command.
 

crates/e2e_test/src/common.rs

@@ -774,6 +774,18 @@ impl RustFSTestEnvironment {
             .await
     }
 
+    /// Like [`Self::start_rustfs_server_from_binary`], but leaves other running RustFS processes
+    /// alone, so that two servers of different binaries can be alive at once.
+    pub async fn start_rustfs_server_from_binary_without_cleanup(
+        &mut self,
+        binary_path: &Path,
+        extra_args: Vec<&str>,
+        extra_env: &[(&str, &str)],
+    ) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        self.start_rustfs_server_inner_with_binary(binary_path, extra_args, extra_env, false)
+            .await
+    }
+
     /// Start RustFS server with basic configuration
     pub async fn start_rustfs_server(&mut self, extra_args: Vec<&str>) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
         self.start_rustfs_server_inner(extra_args, &[], true).await

crates/e2e_test/src/distributed/multipart_fence_acceptance_test.rs

@@ -1285,7 +1285,9 @@ async fn list_parts_disk_loss_after_metadata_read(
     let error = result.expect_err("one missing directory and three unavailable participants cannot prove an empty upload");
     let response = error.raw_response().expect("ListParts must return an S3 service response");
     assert_eq!(response.status().as_u16(), 503);
-    assert_eq!(error.as_service_error().and_then(|err| err.meta().code()), Some("SlowDownRead"));
+    // Since #9149, SlowDownRead is reserved for GET/HEAD object reads; a
+    // ListParts read-quorum failure is reported as ServiceUnavailable.
+    assert_eq!(error.as_service_error().and_then(|err| err.meta().code()), Some("ServiceUnavailable"));
     assert!(response.headers().get("x-rustfs-error-cause").is_none());
     assert!(response.headers().get("retry-after").is_none());
     assert!(

crates/e2e_test/src/distributed/upgrade_test.rs

@@ -493,7 +493,12 @@ async fn compare_degraded_legacy_list_parts(dist: &mut DistCluster, seed: &Upgra
     assert!(legacy.parts().is_empty());
     let current = current.expect_err("current coordinator must not infer empty from missing directories and offline disks");
     assert_eq!(current.raw_response().expect("current S3 response").status().as_u16(), 503);
-    assert_eq!(current.as_service_error().and_then(|error| error.meta().code()), Some("SlowDownRead"));
+    // Since #9149, SlowDownRead is reserved for GET/HEAD object reads; a
+    // ListParts read-quorum failure is reported as ServiceUnavailable.
+    assert_eq!(
+        current.as_service_error().and_then(|error| error.meta().code()),
+        Some("ServiceUnavailable")
+    );
     dist.cluster.start_node_from_binary(3, previous).await?;
     wait_for_upgrade_write_readiness(&clients, &seed.history_bucket, "mixed-list-restored", CREDENTIAL_TIMEOUT).await?;
     for (path, before) in original_files {

crates/e2e_test/src/dual_stack_differential/compare.rs

@@ -232,6 +232,12 @@ pub(crate) struct Report {
     pub run: String,
     pub stack_a: String,
     pub stack_b: String,
+    /// Which side is the pinned release binary: `"a"`, or `None` when both sides are the binary
+    /// under test.
+    pub reference_side: Option<&'static str>,
+    /// `--version` output of the binary each side actually ran.
+    pub version_a: String,
+    pub version_b: String,
     pub region_a: Option<String>,
     pub region_b: Option<String>,
     pub compared_ops: usize,
@@ -480,10 +486,13 @@ mod tests {
     fn the_report_lists_each_difference_as_op_field_a_b() {
         let directory = std::env::temp_dir().join(format!("dual-stack-report-{}", uuid::Uuid::new_v4()));
         let report = Report {
-            schema: 1,
+            schema: 2,
             run: "unit".into(),
             stack_a: "legacy".into(),
             stack_b: "legacy".into(),
+            reference_side: Some("a"),
+            version_a: "rustfs 1.0.4".into(),
+            version_b: "rustfs 1.1.0".into(),
             region_a: None,
             region_b: Some("eu-west-1".into()),
             compared_ops: 1,
@@ -498,6 +507,9 @@ mod tests {
         let _ = std::fs::remove_dir_all(&directory);
         assert_eq!(path.file_name().and_then(|name| name.to_str()), Some("unit.json"));
         assert_eq!(json["compared_ops"], 1);
+        assert_eq!(json["reference_side"], "a");
+        assert_eq!(json["version_a"], "rustfs 1.0.4");
+        assert_eq!(json["version_b"], "rustfs 1.1.0");
         assert_eq!(
             json["differences"],
             serde_json::json!([{

crates/e2e_test/src/dual_stack_differential/mod.rs

@@ -14,7 +14,10 @@
 
 //! Dual-stack differential: one `rustfs` binary, two processes, one SDK script (rustfs/backlog#2740).
 //!
-//! Stack A always runs `RUSTFS_S3_STACK=legacy`. Stack B runs the value of `RUSTFS_E2E_STACK_B`
+//! Stack A runs `RUSTFS_S3_STACK=legacy` of the binary under test, unless `RUSTFS_E2E_REFERENCE_BINARY`
+//! names a released binary: then A is that binary, started without `RUSTFS_S3_STACK` (older
+//! releases do not know it), and a bad path fails the run instead of falling back (see
+//! `reference`). The report names which side is the reference and the `--version` each side ran. Stack B runs the value of `RUSTFS_E2E_STACK_B`
 //! (default `legacy` until the gateway stack is assembled), and `RUSTFS_E2E_STACK_B_REGION`, when
 //! set, starts B in another region so that a difference can be injected by hand. Both servers
 //! answer the script in `scenarios`, which covers every operation of `impl S3 for FS`; `compare`
@@ -23,16 +26,20 @@
 //! is process-global, so two stacks in one process would read each other's buckets.
 
 mod compare;
+mod reference;
 mod scenarios;
 
-use crate::common::{RustFSTestEnvironment, init_logging, workspace_root};
+use crate::common::{RustFSTestEnvironment, init_logging, rustfs_binary_path, workspace_root};
 use compare::{Difference, Observation, Report, compare, compared_cases, compared_ops, render};
-use std::path::PathBuf;
+use reference::{SideA, binary_version};
+use std::path::{Path, PathBuf};
 use std::time::{SystemTime, UNIX_EPOCH};
 
 type BoxError = Box<dyn std::error::Error + Send + Sync>;
 
 const STACK_A: &str = "legacy";
+/// Label of side A in the report when it is the released reference binary.
+const STACK_A_REFERENCE: &str = "reference";
 const ENV_STACK_B: &str = "RUSTFS_E2E_STACK_B";
 const ENV_STACK_B_REGION: &str = "RUSTFS_E2E_STACK_B_REGION";
 /// Becomes `gateway` once rustfs/backlog#2734 T2.5 assembles the gateway stack.
@@ -52,22 +59,31 @@ const CONTROL_REGION_B: &str = "eu-west-1";
 const MIN_NORMAL_SUCCESSES: usize = 68;
 
 struct PairConfig {
+    side_a: SideA,
     stack_b: String,
     region_a: Option<String>,
     region_b: Option<String>,
 }
 
 impl PairConfig {
-    fn from_env() -> Self {
-        Self {
+    fn from_env() -> Result<Self, BoxError> {
+        Ok(Self {
+            side_a: SideA::from_env()?,
             stack_b: std::env::var(ENV_STACK_B).unwrap_or_else(|_| DEFAULT_STACK_B.to_owned()),
             region_a: None,
             region_b: std::env::var(ENV_STACK_B_REGION).ok(),
+        })
+    }
+
+    fn stack_a_label(&self) -> &'static str {
+        match self.side_a {
+            SideA::SameBinary => STACK_A,
+            SideA::Reference(_) => STACK_A_REFERENCE,
         }
     }
 }
 
-/// Starts one server on `stack`, overriding its region only when asked.
+/// Starts one server of the binary under test on `stack`, overriding its region only when asked.
 async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
     let mut env = RustFSTestEnvironment::new().await?;
     let mut vars = vec![("RUSTFS_S3_STACK", stack)];
@@ -78,12 +94,28 @@ async fn start_stack(stack: &str, region: Option<&str>) -> Result<RustFSTestEnvi
     Ok(env)
 }
 
+/// Starts one server of a released binary. `RUSTFS_S3_STACK` is not passed: older releases do not
+/// know it and have exactly one stack.
+async fn start_reference(binary: &Path, region: Option<&str>) -> Result<RustFSTestEnvironment, BoxError> {
+    let mut env = RustFSTestEnvironment::new().await?;
+    let mut vars = Vec::new();
+    if let Some(region) = region {
+        vars.push(("RUSTFS_REGION", region));
+    }
+    env.start_rustfs_server_from_binary_without_cleanup(binary, vec![], &vars)
+        .await?;
+    Ok(env)
+}
+
 /// Starts stack A and stack B side by side, each with its own port and data directory.
 async fn spawn_pair(config: &PairConfig) -> Result<(RustFSTestEnvironment, RustFSTestEnvironment), BoxError> {
-    tokio::try_join!(
-        start_stack(STACK_A, config.region_a.as_deref()),
-        start_stack(&config.stack_b, config.region_b.as_deref())
-    )
+    let side_a = async {
+        match &config.side_a {
+            SideA::SameBinary => start_stack(STACK_A, config.region_a.as_deref()).await,
+            SideA::Reference(binary) => start_reference(binary, config.region_a.as_deref()).await,
+        }
+    };
+    tokio::try_join!(side_a, start_stack(&config.stack_b, config.region_b.as_deref()))
 }
 
 fn report_directory() -> PathBuf {
@@ -104,16 +136,34 @@ struct Run {
 
 /// Runs the script against both stacks concurrently and writes the report.
 async fn run_differential(label: &str, config: &PairConfig) -> Run {
+    let binary_under_test = rustfs_binary_path();
+    let (version_a, version_b) = match &config.side_a {
+        SideA::SameBinary => {
+            let version = binary_version(&binary_under_test)
+                .await
+                .expect("capture the binary under test version");
+            (version.clone(), version)
+        }
+        SideA::Reference(binary) => (
+            binary_version(binary).await.expect("capture the reference binary version"),
+            binary_version(&binary_under_test)
+                .await
+                .expect("capture the binary under test version"),
+        ),
+    };
     let (server_a, server_b) = spawn_pair(config).await.expect("start stack A and stack B");
     let (a, b) = tokio::join!(Box::pin(scenarios::run(&server_a)), Box::pin(scenarios::run(&server_b)));
     let millis = SystemTime::now()
         .duration_since(UNIX_EPOCH)
         .map_or(0, |elapsed| elapsed.as_millis());
     let report = Report {
-        schema: 1,
+        schema: 2,
         run: format!("{label}-{millis}-{}", std::process::id()),
-        stack_a: STACK_A.to_owned(),
+        stack_a: config.stack_a_label().to_owned(),
         stack_b: config.stack_b.clone(),
+        reference_side: matches!(config.side_a, SideA::Reference(_)).then_some("a"),
+        version_a,
+        version_b,
         region_a: config.region_a.clone(),
         region_b: config.region_b.clone(),
         compared_ops: compared_ops(&a, &b).len(),
@@ -131,7 +181,7 @@ mod tests {
     #[tokio::test]
     async fn both_stacks_answer_every_operation_identically() {
         init_logging();
-        let config = PairConfig::from_env();
+        let config = PairConfig::from_env().expect("RUSTFS_E2E_REFERENCE_BINARY must be valid when set");
         let run = run_differential("self-comparison", &config).await;
         let report = &run.report;
 
@@ -146,9 +196,11 @@ mod tests {
         assert_eq!(report.compared_ops, scenarios::OPERATIONS.len(), "report: {}", run.path.display());
         assert!(
             report.differences.is_empty(),
-            "stack {} and stack {} differ on {} of {} cases (report: {}):\n{}",
+            "stack {} ({}) and stack {} ({}) differ on {} of {} cases (report: {}):\n{}",
             report.stack_a,
+            report.version_a,
             report.stack_b,
+            report.version_b,
             report.differences.len(),
             report.compared_cases,
             run.path.display(),
@@ -160,7 +212,11 @@ mod tests {
             .iter()
             .filter(|observation| observation.case == "negative" && !(400..500).contains(&observation.status))
             .collect();
-        assert!(not_4xx.is_empty(), "negative cases must fail with a 4xx on {STACK_A}: {not_4xx:#?}");
+        assert!(
+            not_4xx.is_empty(),
+            "negative cases must fail with a 4xx on side A ({}): {not_4xx:#?}",
+            report.version_a
+        );
         let successes = run
             .a
             .iter()
@@ -168,7 +224,8 @@ mod tests {
             .count();
         assert!(
             successes >= MIN_NORMAL_SUCCESSES,
-            "only {successes} normal cases succeeded on {STACK_A}: {:#?}",
+            "only {successes} normal cases succeeded on side A ({}): {:#?}",
+            report.version_a,
             run.a
         );
     }
@@ -177,6 +234,7 @@ mod tests {
     async fn differential_detects_an_injected_difference() {
         init_logging();
         let config = PairConfig {
+            side_a: SideA::SameBinary,
             stack_b: STACK_A.to_owned(),
             region_a: Some(CONTROL_REGION_A.to_owned()),
             region_b: Some(CONTROL_REGION_B.to_owned()),

crates/e2e_test/src/dual_stack_differential/reference.rs

@@ -0,0 +1,163 @@
+// 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.
+
+//! Which binary side A of the differential runs, and what it reports about itself.
+//!
+//! `RUSTFS_E2E_REFERENCE_BINARY` names a released `rustfs` binary to use as the reference. Unset,
+//! side A is the binary under test on `RUSTFS_S3_STACK=legacy`. Set, it must resolve to an
+//! executable file or the run fails: a reference that was asked for and silently replaced by the
+//! binary under test would report an agreement with itself as agreement with the release.
+
+use std::ffi::OsString;
+use std::path::{Path, PathBuf};
+
+type BoxError = Box<dyn std::error::Error + Send + Sync>;
+
+pub(super) const ENV_REFERENCE_BINARY: &str = "RUSTFS_E2E_REFERENCE_BINARY";
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub(super) enum SideA {
+    /// Both sides are the binary under test; A runs `RUSTFS_S3_STACK=legacy`.
+    SameBinary,
+    /// A is this released binary, started without `RUSTFS_S3_STACK`.
+    Reference(PathBuf),
+}
+
+impl SideA {
+    pub(super) fn from_env() -> Result<Self, BoxError> {
+        Self::resolve(std::env::var_os(ENV_REFERENCE_BINARY))
+    }
+
+    /// `None` is the only value that selects same-binary mode; every other value must validate.
+    pub(super) fn resolve(value: Option<OsString>) -> Result<Self, BoxError> {
+        let Some(value) = value else {
+            return Ok(Self::SameBinary);
+        };
+        let path = PathBuf::from(value);
+        validate_executable(&path)?;
+        Ok(Self::Reference(path))
+    }
+}
+
+fn validate_executable(path: &Path) -> Result<(), BoxError> {
+    if path.as_os_str().is_empty() {
+        return Err(format!("{ENV_REFERENCE_BINARY} is set but empty").into());
+    }
+    let metadata = std::fs::metadata(path)
+        .map_err(|error| format!("{ENV_REFERENCE_BINARY} points to {}, which cannot be read: {error}", path.display()))?;
+    if !metadata.is_file() {
+        return Err(format!("{ENV_REFERENCE_BINARY} points to {}, which is not a file", path.display()).into());
+    }
+    if !is_executable(&metadata) {
+        return Err(format!("{ENV_REFERENCE_BINARY} points to {}, which is not executable", path.display()).into());
+    }
+    Ok(())
+}
+
+#[cfg(unix)]
+fn is_executable(metadata: &std::fs::Metadata) -> bool {
+    use std::os::unix::fs::PermissionsExt;
+    metadata.permissions().mode() & 0o111 != 0
+}
+
+#[cfg(not(unix))]
+fn is_executable(_metadata: &std::fs::Metadata) -> bool {
+    true
+}
+
+/// What `<binary> --version` printed, trimmed. Fails when the binary does not run, exits
+/// non-zero, or prints nothing, because an empty version would report nothing as observed.
+pub(super) async fn binary_version(binary: &Path) -> Result<String, BoxError> {
+    let output = tokio::process::Command::new(binary).arg("--version").output().await?;
+    if !output.status.success() {
+        return Err(format!("{} --version failed with {}", binary.display(), output.status).into());
+    }
+    let version = String::from_utf8_lossy(&output.stdout).trim().to_owned();
+    if version.is_empty() {
+        return Err(format!("{} --version printed nothing", binary.display()).into());
+    }
+    Ok(version)
+}
+
+#[cfg(all(test, unix))]
+mod tests {
+    use super::*;
+    use std::os::unix::fs::PermissionsExt;
+
+    fn script(dir: &Path, name: &str, body: &str, mode: u32) -> PathBuf {
+        let path = dir.join(name);
+        std::fs::write(&path, body).expect("write the stand-in");
+        std::fs::set_permissions(&path, std::fs::Permissions::from_mode(mode)).expect("chmod");
+        path
+    }
+
+    #[test]
+    fn missing_path_is_an_error_not_a_fallback() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let error = SideA::resolve(Some(dir.path().join("absent").into_os_string())).expect_err("must fail closed");
+        assert!(error.to_string().contains("cannot be read"), "{error}");
+    }
+
+    #[test]
+    fn non_executable_file_is_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho v\n", 0o644);
+        let error = SideA::resolve(Some(path.into_os_string())).expect_err("must fail closed");
+        assert!(error.to_string().contains("not executable"), "{error}");
+    }
+
+    #[test]
+    fn directory_is_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let error = SideA::resolve(Some(dir.path().as_os_str().to_owned())).expect_err("must fail closed");
+        assert!(error.to_string().contains("not a file"), "{error}");
+    }
+
+    #[test]
+    fn empty_value_is_an_error() {
+        let error = SideA::resolve(Some(OsString::new())).expect_err("must fail closed");
+        assert!(error.to_string().contains("empty"), "{error}");
+    }
+
+    #[test]
+    fn unset_selects_same_binary_mode() {
+        assert_eq!(SideA::resolve(None).expect("unset is valid"), SideA::SameBinary);
+    }
+
+    #[test]
+    fn executable_file_selects_reference_mode() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho v\n", 0o755);
+        assert_eq!(
+            SideA::resolve(Some(path.clone().into_os_string())).expect("valid"),
+            SideA::Reference(path)
+        );
+    }
+
+    #[tokio::test]
+    async fn version_output_is_captured_from_the_binary() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let path = script(dir.path(), "rustfs", "#!/bin/sh\necho 'rustfs 1.0.4 (abc123)'\n", 0o755);
+        assert_eq!(binary_version(&path).await.expect("version"), "rustfs 1.0.4 (abc123)");
+    }
+
+    #[tokio::test]
+    async fn failing_or_silent_version_is_an_error() {
+        let dir = tempfile::tempdir().expect("tempdir");
+        let failing = script(dir.path(), "failing", "#!/bin/sh\nexit 3\n", 0o755);
+        assert!(binary_version(&failing).await.is_err());
+        let silent = script(dir.path(), "silent", "#!/bin/sh\nexit 0\n", 0o755);
+        assert!(binary_version(&silent).await.is_err());
+    }
+}

crates/ecstore/src/disk/local.rs

@@ -11541,11 +11541,14 @@ impl LocalDisk {
     ) -> Result<()> {
         multipart_staging::reject_reserved_anchor(path)?;
         let fuse_with_admission = admission.is_some() && !crash_inject::is_armed(CrashPoint::MetaWriteAfterTmpBeforeRename, path);
+        // Held until the staging root is published, by the filesystem work
+        // itself, so a concurrent listing cannot prune the marker in between.
+        let mut index_pin = None;
         let prepared_index = if volume == RUSTFS_META_MULTIPART_BUCKET {
             if fuse_with_admission {
                 self.multipart_index.prepare_record_head(path, &buf).await
             } else {
-                self.multipart_index.record_head(path, &buf).await?;
+                index_pin = self.multipart_index.record_head(path, &buf).await?;
                 None
             }
         } else {
@@ -11571,8 +11574,10 @@ impl LocalDisk {
             let root = self.publication_root.clone();
             let lease = os::acquire_namespace_mutation_lease_with_owner(&file_path, namespace_owner).await;
             os::run_blocking_namespace_operation_with_sync(lease, do_sync, move || {
-                if let Some(record) = prepared_index {
-                    record.run_std()?;
+                let _index_pin = prepared_index.map(|record| record.run_std()).transpose()?;
+                #[cfg(test)]
+                if let Some(pin) = &_index_pin {
+                    pin.pause_for_test();
                 }
                 super::fs::create_dir_all_std(&tmp_volume_dir)?;
                 #[cfg(test)]
@@ -11616,12 +11621,16 @@ impl LocalDisk {
             return Err(DiskError::Unexpected);
         }
 
+        let publication_owner: Option<Arc<dyn Send + Sync>> = match index_pin {
+            Some(pin) => Some(Arc::new((namespace_owner.clone(), pin))),
+            None => namespace_owner.clone(),
+        };
         os::rename_all_with_precondition_and_parent_sync(
             &tmp_file_path,
             &file_path,
             &volume_dir,
             &self.publication_root,
-            namespace_owner.clone(),
+            publication_owner,
             admission.as_ref().map(|admission| admission.precondition()),
             admission.is_some() && sync && durability.syncs_commit_metadata(),
         )

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

@@ -521,8 +521,12 @@ impl LocalDisk {
             state.preflight_rejection = Some(LocalRenamePreflightRejection(()));
         })?;
         crate::hp_guard!("LocalDisk::rename_data");
-        if dst_volume == super::super::RUSTFS_META_MULTIPART_BUCKET {
-            self.multipart_index.record_info(dst_path, &fi).await?;
+        if dst_volume == super::super::RUSTFS_META_MULTIPART_BUCKET
+            && let Some(pin) = self.multipart_index.record_info(dst_path, &fi).await?
+        {
+            // The owner chain reaches every lease this rename publishes under,
+            // so the index marker stays pinned until that work has returned.
+            state.namespace_owner = Some(Arc::new((state.namespace_owner.take(), pin)));
         }
         #[cfg(feature = "e2e-test-hooks")]
         if fi.is_healing() {

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

@@ -4,6 +4,11 @@
 //! Bucket-to-staging-root candidates: one marker per staging root, recorded
 //! before the root is published and pruned once the root is gone. Source
 //! metadata stays authoritative.
+//!
+//! A marker without a staging root is either a leftover or an upload whose
+//! root is still being published. The publisher proves the second case by
+//! holding a [`PublicationPin`] from before it records the marker until its
+//! filesystem work has returned; no elapsed time is taken as that proof.
 
 use super::{LocalDisk, read_metadata_only_std};
 use crate::disk::{
@@ -29,9 +34,6 @@ use uuid::Uuid;
 
 const INDEX: &str = ".rustfs.sys/multipart-index-v1";
 const REBUILD_DIRENTS_PER_JOB: usize = 64;
-/// A marker is recorded before its staging root is published. A marker younger
-/// than this without a root is an upload in flight, never a leftover to prune.
-const MULTIPART_INDEX_PRUNE_GRACE: std::time::Duration = std::time::Duration::from_secs(300);
 type DirectoryEntries = Box<dyn Iterator<Item = io::Result<String>> + Send>;
 type PhysicalRoot = (u64, [u8; 16]);
 static OWNERS: LazyLock<parking_lot::Mutex<HashMap<PhysicalRoot, Weak<IndexOwner>>>> =
@@ -56,9 +58,15 @@ pub(super) struct IndexOwner {
     // drop so a cancelled scan can be resumed by the next caller.
     driving: AtomicBool,
     disabled: AtomicBool,
+    // Markers whose staging root some publisher is creating right now, with
+    // the number of publishers. Boot-local like the index itself: every writer
+    // of this physical drive in this process shares this owner.
+    publishing: parking_lot::Mutex<HashMap<PathBuf, usize>>,
     #[cfg(test)]
     rebuild_head_pause: parking_lot::Mutex<Option<RebuildHeadPause>>,
     #[cfg(test)]
+    publication_pause: parking_lot::Mutex<Option<RebuildHeadPause>>,
+    #[cfg(test)]
     observed_rebuild_queue: parking_lot::Mutex<Option<Arc<os::DiskBlockingQueue>>>,
     #[cfg(test)]
     rebuild_single_job: AtomicBool,
@@ -120,23 +128,6 @@ impl IndexDirectory {
             std::fs::File::open(self.path.join(name))
         }
     }
-    fn modified(&self, name: &str) -> io::Result<std::time::SystemTime> {
-        #[cfg(unix)]
-        {
-            self.open_file(name)?.metadata()?.modified()
-        }
-        #[cfg(not(unix))]
-        {
-            std::fs::metadata(self.path.join(name))?.modified()
-        }
-    }
-    /// Whether the marker is still inside the publication grace period. A
-    /// marker dated in the future (clock stepped back) is never a leftover.
-    fn marker_is_young(&self, name: &str) -> bool {
-        self.modified(name)
-            .ok()
-            .is_some_and(|modified| modified.elapsed().map_or(true, |age| age < MULTIPART_INDEX_PRUNE_GRACE))
-    }
     fn remove_file(&self, name: &str) -> io::Result<()> {
         #[cfg(unix)]
         {
@@ -280,9 +271,12 @@ impl IndexOwner {
             initialized: OnceLock::new(),
             driving: AtomicBool::new(false),
             disabled: AtomicBool::new(false),
+            publishing: parking_lot::Mutex::new(HashMap::new()),
             #[cfg(test)]
             rebuild_head_pause: parking_lot::Mutex::new(None),
             #[cfg(test)]
+            publication_pause: parking_lot::Mutex::new(None),
+            #[cfg(test)]
             observed_rebuild_queue: parking_lot::Mutex::new(None),
             #[cfg(test)]
             rebuild_single_job: AtomicBool::new(false),
@@ -394,8 +388,8 @@ impl IndexOwner {
             // The index is boot-local: every start rescans the staging area
             // before serving it, so a marker lost to a crash is recreated before
             // it can be consulted. Creation is enough; no fsync of the marker or
-            // its ancestors. An existing marker is touched so pruning sees the
-            // new upload's age, not the previous one's.
+            // its ancestors. An existing marker needs nothing: the caller's
+            // publication pin, not the marker's age, keeps it from being pruned.
             match openat(
                 &parent,
                 source_root,
@@ -407,16 +401,7 @@ impl IndexOwner {
                         return Err(io::Error::other("index marker is not a file"));
                     }
                 }
-                Err(rustix::io::Errno::EXIST) => {
-                    // Best effort: a filesystem that refuses to set times keeps
-                    // the previous upload's age, which only shortens the prune
-                    // grace for this key; index completeness does not depend on it.
-                    if let Ok(marker) =
-                        openat(&parent, source_root, OFlags::WRONLY | OFlags::NOFOLLOW | OFlags::CLOEXEC, Mode::empty())
-                    {
-                        let _ = std::fs::File::from(marker).set_modified(std::time::SystemTime::now());
-                    }
-                }
+                Err(rustix::io::Errno::EXIST) => {}
                 Err(error) => return Err(error.into()),
             }
         }
@@ -432,6 +417,22 @@ impl IndexOwner {
         Ok(())
     }
 
+    /// Declare that the caller is about to publish the staging root of
+    /// `marker`. Must precede the marker lease the caller records under, so a
+    /// prune that found no pin has released that lease before the marker is
+    /// (re)created.
+    fn pin_publication(self: &Arc<Self>, marker: PathBuf) -> PublicationPin {
+        *self.publishing.lock().entry(marker.clone()).or_default() += 1;
+        PublicationPin {
+            owner: Arc::clone(self),
+            marker,
+        }
+    }
+
+    fn is_publishing(&self, marker: &Path) -> bool {
+        self.publishing.lock().contains_key(marker)
+    }
+
     fn ensure_current_root(&self) -> io::Result<()> {
         #[cfg(unix)]
         let current = {
@@ -513,9 +514,12 @@ impl IndexOwner {
         }
     }
 
-    pub(super) async fn record_info(self: &Arc<Self>, path: &str, info: &FileInfo) -> Result<()> {
+    /// Record the marker of `path`'s staging root. The returned pin must stay
+    /// alive until the filesystem work that publishes the root has returned,
+    /// including work that outlives a cancelled caller.
+    pub(super) async fn record_info(self: &Arc<Self>, path: &str, info: &FileInfo) -> Result<Option<PublicationPin>> {
         if canonical_upload(path).is_none() {
-            return Ok(());
+            return Ok(None);
         }
         let owner = self.clone();
         let path = path.to_owned();
@@ -531,6 +535,7 @@ impl IndexOwner {
                     .join(path.split('/').next().unwrap_or_default())
             })
             .unwrap_or_else(|| owner.base.join(INDEX));
+        let pin = self.pin_publication(marker.clone());
         let lease = os::acquire_namespace_mutation_lease_with_owner(&marker, None).await;
         os::run_blocking_namespace_operation_with_sync(lease, true, move || {
             match owner.record(&path, &info) {
@@ -544,7 +549,7 @@ impl IndexOwner {
             }
         })
         .await?;
-        Ok(())
+        Ok(Some(pin))
     }
 
     pub(super) async fn prepare_record_head(self: &Arc<Self>, path: &str, bytes: &[u8]) -> Option<PreparedHeadRecord> {
@@ -578,20 +583,23 @@ impl IndexOwner {
                     .join(upload.split('/').next().unwrap_or_default())
             })
             .unwrap_or_else(|| self.base.join(INDEX));
+        let pin = self.pin_publication(marker.clone());
         let lease = os::acquire_namespace_mutation_lease_with_owner(&marker, None).await;
         Some(PreparedHeadRecord {
-            owner: Arc::clone(self),
             upload: upload.to_owned(),
             info,
+            pin,
             _lease: lease,
         })
     }
 
-    pub(super) async fn record_head(self: &Arc<Self>, path: &str, bytes: &[u8]) -> Result<()> {
+    /// Record the marker of the staging root that `path` belongs to. See
+    /// [`Self::record_info`] for how long the returned pin must live.
+    pub(super) async fn record_head(self: &Arc<Self>, path: &str, bytes: &[u8]) -> Result<Option<PublicationPin>> {
         let Some(prepared) = self.prepare_record_head(path, bytes).await else {
-            return Ok(());
+            return Ok(None);
         };
-        prepared.run().await
+        prepared.run().await.map(Some)
     }
 
     async fn page(self: &Arc<Self>, bucket: &str, marker: Option<&str>, disk_id: Uuid) -> Result<MultipartIndexPage> {
@@ -642,17 +650,20 @@ impl IndexOwner {
             // to every key ever multipart-initiated on this drive.
             let mut live = Vec::with_capacity(roots.len());
             let mut stale = Vec::new();
+            let markers = owner.base.join(INDEX).join(sha(&bucket));
             for root in roots {
+                // The marker precedes its staging root: a pinned one is an
+                // upload in flight and stays a candidate. The pin is read
+                // before the root, so a publisher that finished in between has
+                // already made its root visible.
+                if owner.is_publishing(&markers.join(&root)) {
+                    live.push(root);
+                    continue;
+                }
                 match owner.directory(&Path::new(RUSTFS_META_MULTIPART_BUCKET).join(&root)) {
                     Ok(_) => live.push(root),
                     Err(error) if matches!(error.kind(), io::ErrorKind::NotFound | io::ErrorKind::NotADirectory) => {
-                        // The marker precedes its staging root; a young one is an
-                        // upload in flight and stays a candidate.
-                        if directory.as_ref().is_some_and(|index| index.marker_is_young(&root)) {
-                            live.push(root);
-                        } else {
-                            stale.push(root);
-                        }
+                        stale.push(root);
                     }
                     Err(error) => return Err(DiskError::from(error)),
                 }
@@ -677,9 +688,11 @@ impl IndexOwner {
     }
 
     /// Remove one marker whose staging root was gone when a page examined it.
-    /// The marker's own namespace lease serializes this with `record_info`; the
-    /// staging root and the marker's age are re-checked under that lease, so an
-    /// upload recorded in between keeps its marker.
+    /// The marker's own namespace lease serializes this with recording. Under
+    /// that lease the publication pin is read first and the staging root
+    /// second: a publisher pinned before the read keeps the marker, one that
+    /// already finished has made its root visible, and one that pins later
+    /// records its marker only after this lease is released.
     async fn prune_marker(self: &Arc<Self>, bucket: &str, root: &str) -> Result<()> {
         let owner = self.clone();
         let bucket_directory = Path::new(INDEX).join(sha(bucket));
@@ -690,6 +703,9 @@ impl IndexOwner {
         // not queue behind every other mutation on the one-worker foreground lane.
         os::run_blocking_namespace_operation_with_sync(lease, true, move || {
             owner.ensure_current_root()?;
+            if owner.is_publishing(&marker) {
+                return Ok(());
+            }
             match owner.directory(&Path::new(RUSTFS_META_MULTIPART_BUCKET).join(&root)) {
                 Ok(_) => return Ok(()),
                 Err(error) if matches!(error.kind(), io::ErrorKind::NotFound | io::ErrorKind::NotADirectory) => {}
@@ -700,9 +716,6 @@ impl IndexOwner {
                 Err(error) if error.kind() == io::ErrorKind::NotFound => return Ok(()),
                 Err(error) => return Err(error),
             };
-            if directory.marker_is_young(&root) {
-                return Ok(());
-            }
             match directory.remove_file(&root) {
                 Err(error) if error.kind() != io::ErrorKind::NotFound => Err(error),
                 _ => Ok(()),
@@ -713,34 +726,69 @@ impl IndexOwner {
     }
 }
 
-pub(super) struct PreparedHeadRecord {
+/// Proof that the staging root of one marker is being published. While any pin
+/// for a marker is alive, pruning leaves that marker alone.
+#[must_use = "dropping the pin lets a concurrent listing prune the marker before its staging root exists"]
+pub(super) struct PublicationPin {
     owner: Arc<IndexOwner>,
+    marker: PathBuf,
+}
+
+impl PublicationPin {
+    /// Test seam between recording the marker and publishing the staging root.
+    #[cfg(test)]
+    pub(super) fn pause_for_test(&self) {
+        let pause = self.owner.publication_pause.lock().take();
+        if let Some(pause) = pause {
+            let _pause = os::enter_blocking_test_hook();
+            let _ = pause.started.send(());
+            let _ = pause.release.recv_timeout(std::time::Duration::from_secs(10));
+        }
+    }
+}
+
+impl Drop for PublicationPin {
+    fn drop(&mut self) {
+        let mut publishing = self.owner.publishing.lock();
+        if let Some(count) = publishing.get_mut(&self.marker) {
+            *count -= 1;
+            if *count == 0 {
+                publishing.remove(&self.marker);
+            }
+        }
+    }
+}
+
+pub(super) struct PreparedHeadRecord {
     upload: String,
     info: FileInfo,
+    pin: PublicationPin,
     _lease: Arc<os::NamespaceMutationLease>,
 }
 
 impl PreparedHeadRecord {
-    pub(super) fn run_std(self) -> io::Result<()> {
+    /// Record the marker and release its lease. The caller publishes the
+    /// staging root while it holds the returned pin.
+    pub(super) fn run_std(self) -> io::Result<PublicationPin> {
         let Self {
-            owner,
             upload,
             info,
+            pin,
             _lease,
         } = self;
-        match owner.record(&upload, &info) {
+        match pin.owner.record(&upload, &info) {
             Err(error) if error.kind() == io::ErrorKind::InvalidData => {
-                owner.disabled.store(true, Ordering::Release);
-                Ok(())
+                pin.owner.disabled.store(true, Ordering::Release);
+                Ok(pin)
             }
-            result => result,
+            Err(error) => Err(error),
+            Ok(()) => Ok(pin),
         }
     }
 
-    async fn run(self) -> Result<()> {
+    async fn run(self) -> Result<PublicationPin> {
         let lease = Arc::clone(&self._lease);
-        os::run_blocking_namespace_operation_with_sync(lease, true, move || self.run_std()).await?;
-        Ok(())
+        Ok(os::run_blocking_namespace_operation_with_sync(lease, true, move || self.run_std()).await?)
     }
 }
 
@@ -958,47 +1006,274 @@ mod tests {
         let live = format!("{:064x}", 1);
         let finished = format!("{:064x}", 2);
         let in_flight = format!("{:064x}", 3);
-        let clock_stepped = format!("{:064x}", 4);
-        for name in [&live, &finished, &in_flight, &clock_stepped] {
+        for name in [&live, &finished, &in_flight] {
             std::fs::write(directory.join(name), b"1").expect("marker");
         }
         // Completion and abort remove the staging root and leave the marker.
         std::fs::create_dir_all(root.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&live)).expect("live staging root");
-        set_marker_modified(&directory.join(&finished), std::time::SystemTime::now() - MULTIPART_INDEX_PRUNE_GRACE * 2);
-        // A marker dated after a backwards clock step is never a leftover.
-        set_marker_modified(
-            &directory.join(&clock_stepped),
-            std::time::SystemTime::now() + MULTIPART_INDEX_PRUNE_GRACE * 2,
-        );
-        let page = disk.multipart_index.page("bucket", None, Uuid::new_v4()).await.expect("page");
-        // The in-flight marker has no staging root yet but is young: a candidate, not a leftover.
-        assert_eq!(page.roots, vec![live.clone(), in_flight.clone(), clock_stepped.clone()]);
+        let owner = &disk.multipart_index;
+        let publishing = owner.pin_publication(owner.base.join(INDEX).join(sha("bucket")).join(&in_flight));
+        let second_publisher = owner.pin_publication(owner.base.join(INDEX).join(sha("bucket")).join(&in_flight));
+        let page = owner.page("bucket", None, Uuid::new_v4()).await.expect("page");
+        // The in-flight marker has no staging root yet but is pinned: a candidate, not a leftover.
+        assert_eq!(page.roots, vec![live.clone(), in_flight.clone()]);
         assert!(page.next_marker.is_none());
         assert!(directory.join(&live).exists(), "a live root keeps its marker");
-        assert!(directory.join(&in_flight).exists(), "a marker inside the grace period is kept");
-        assert!(directory.join(&clock_stepped).exists(), "a future-dated marker is kept");
-        assert!(!directory.join(&finished).exists(), "a finished root's marker is pruned");
-        // The next page no longer pays for the pruned marker.
-        let page = disk
-            .multipart_index
+        assert!(directory.join(&in_flight).exists(), "a pinned marker is kept");
+        assert!(!directory.join(&finished).exists(), "a finished root's marker is pruned at once");
+        // One of two publishers of the same root finishing does not unpin it.
+        drop(publishing);
+        let page = owner
             .page("bucket", None, Uuid::new_v4())
             .await
-            .expect("page after pruning");
-        assert_eq!(page.roots, vec![live, in_flight, clock_stepped]);
+            .expect("page with one publisher left");
+        assert_eq!(page.roots, vec![live.clone(), in_flight.clone()]);
+        assert!(directory.join(&in_flight).exists(), "the remaining publisher keeps the marker");
+        // A publication that ended without a staging root leaves a leftover.
+        drop(second_publisher);
+        let page = owner
+            .page("bucket", None, Uuid::new_v4())
+            .await
+            .expect("page after the failed publication");
+        assert_eq!(page.roots, vec![live]);
+        assert!(!directory.join(&in_flight).exists(), "an unpinned marker without a root is pruned");
+        assert!(owner.publishing.lock().is_empty(), "finished publishers leave no pin behind");
     }
 
-    /// Move a marker's modification time outside the publication grace period.
-    fn age_marker(marker: &Path) {
-        set_marker_modified(marker, std::time::SystemTime::now() - MULTIPART_INDEX_PRUNE_GRACE * 2);
+    fn pause_next_publication(owner: &IndexOwner) -> (tokio::sync::oneshot::Receiver<()>, std::sync::mpsc::Sender<()>) {
+        let (started, waiting) = tokio::sync::oneshot::channel();
+        let (release, released) = std::sync::mpsc::channel();
+        *owner.publication_pause.lock() = Some(RebuildHeadPause {
+            started,
+            release: released,
+        });
+        (waiting, release)
     }
 
-    fn set_marker_modified(marker: &Path, modified: std::time::SystemTime) {
+    /// The window the pin closes: the marker is recorded and its lease is
+    /// released, the staging root is not published yet, and a listing examines
+    /// the bucket. The marker is dated a day back so that no grace period could
+    /// be what keeps it.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn multipart_index_listing_during_source_publication_keeps_the_new_upload_listed() {
+        let root = tempfile::tempdir().expect("disk root");
+        let disk = Arc::new(open(root.path()).await);
+        let (path, info) = upload("bucket", "published-while-listed");
+        let source_root = path.split('/').next().expect("source sha").to_owned();
+        let marker = root.path().join(INDEX).join(sha("bucket")).join(&source_root);
+        let source_head = root
+            .path()
+            .join(RUSTFS_META_MULTIPART_BUCKET)
+            .join(&path)
+            .join(STORAGE_FORMAT_FILE);
+        let (paused, release) = pause_next_publication(&disk.multipart_index);
+        let writer = tokio::spawn({
+            let disk = Arc::clone(&disk);
+            let path = path.clone();
+            async move { disk.write_metadata("bucket", RUSTFS_META_MULTIPART_BUCKET, &path, info).await }
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), paused)
+            .await
+            .expect("the publication reaches the pause")
+            .expect("pause sender");
+        assert!(marker.exists(), "the marker precedes the staging root");
+        assert!(!source_head.exists(), "the staging root is not published yet");
         std::fs::File::options()
             .write(true)
-            .open(marker)
+            .open(&marker)
             .expect("marker")
-            .set_modified(modified)
-            .expect("set marker time");
+            .set_modified(std::time::SystemTime::now() - std::time::Duration::from_secs(86_400))
+            .expect("age the marker");
+
+        let identity = Uuid::new_v4();
+        let page = disk
+            .multipart_index
+            .page("bucket", None, identity)
+            .await
+            .expect("page during publication");
+        assert_eq!(page.roots, vec![source_root.clone()], "an upload in flight stays a candidate");
+        assert!(marker.exists(), "a listing must not prune the marker of an upload in flight");
+
+        release.send(()).expect("resume the publication");
+        writer.await.expect("writer task").expect("the upload metadata is published");
+        assert!(source_head.exists());
+        let page = disk
+            .multipart_index
+            .page("bucket", None, identity)
+            .await
+            .expect("page after publication");
+        assert_eq!(page.roots, vec![source_root], "the published upload is listed");
+        assert!(marker.exists());
+        assert!(disk.multipart_index.publishing.lock().is_empty());
+    }
+
+    /// A timed-out caller drops its future while the filesystem work goes on.
+    /// The pin belongs to that work, so the marker survives until the staging
+    /// root the detached work publishes is visible.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn multipart_index_cancelled_publisher_keeps_its_marker_until_the_detached_work_returns() {
+        let root = tempfile::tempdir().expect("disk root");
+        let disk = Arc::new(open(root.path()).await);
+        let (path, info) = upload("bucket", "caller-timed-out");
+        let source_root = path.split('/').next().expect("source sha").to_owned();
+        let marker = root.path().join(INDEX).join(sha("bucket")).join(&source_root);
+        let source_head = root
+            .path()
+            .join(RUSTFS_META_MULTIPART_BUCKET)
+            .join(&path)
+            .join(STORAGE_FORMAT_FILE);
+        let (paused, release) = pause_next_publication(&disk.multipart_index);
+        let writer = tokio::spawn({
+            let disk = Arc::clone(&disk);
+            let path = path.clone();
+            async move { disk.write_metadata("bucket", RUSTFS_META_MULTIPART_BUCKET, &path, info).await }
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), paused)
+            .await
+            .expect("the publication reaches the pause")
+            .expect("pause sender");
+        writer.abort();
+        assert!(writer.await.expect_err("the caller is gone").is_cancelled());
+
+        let identity = Uuid::new_v4();
+        let page = disk
+            .multipart_index
+            .page("bucket", None, identity)
+            .await
+            .expect("page after the caller left");
+        assert_eq!(page.roots, vec![source_root.clone()]);
+        assert!(marker.exists(), "the detached publication still owns the marker");
+
+        release.send(()).expect("resume the detached publication");
+        tokio::time::timeout(std::time::Duration::from_secs(10), async {
+            while !source_head.exists() || !disk.multipart_index.publishing.lock().is_empty() {
+                tokio::time::sleep(std::time::Duration::from_millis(5)).await;
+            }
+        })
+        .await
+        .expect("the detached publication completes");
+        let page = disk
+            .multipart_index
+            .page("bucket", None, identity)
+            .await
+            .expect("page after the detached publication");
+        assert_eq!(page.roots, vec![source_root], "the upload the detached work published is listed");
+        assert!(marker.exists());
+    }
+
+    /// A page classifies a marker as stale before the publisher pins it; the
+    /// prune that follows must read the pin again under the marker lease.
+    #[tokio::test]
+    async fn multipart_index_prune_rechecks_the_pin_under_the_marker_lease() {
+        let root = tempfile::tempdir().expect("disk root");
+        let disk = open(root.path()).await;
+        let owner = &disk.multipart_index;
+        let directory = root.path().join(INDEX).join(sha("bucket"));
+        std::fs::create_dir_all(&directory).expect("marker directory");
+        let name = format!("{:064x}", 9);
+        std::fs::write(directory.join(&name), b"1").expect("marker");
+
+        let pin = owner.pin_publication(owner.base.join(INDEX).join(sha("bucket")).join(&name));
+        owner.prune_marker("bucket", &name).await.expect("prune of a pinned marker");
+        assert!(directory.join(&name).exists(), "a publisher that pinned after the page keeps its marker");
+
+        drop(pin);
+        owner.prune_marker("bucket", &name).await.expect("prune of a leftover");
+        assert!(!directory.join(&name).exists(), "the same marker is a leftover once nobody publishes it");
+    }
+
+    /// The publisher pins before it queues on the marker lease. A prune that
+    /// holds the lease and found no pin has therefore unlinked the marker
+    /// before the publisher records it, and the record recreates it.
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn multipart_index_publisher_pins_before_it_waits_for_the_marker_lease() {
+        let root = tempfile::tempdir().expect("disk root");
+        let disk = Arc::new(open(root.path()).await);
+        let owner = Arc::clone(&disk.multipart_index);
+        let (path, info) = upload("bucket", "queued-behind-a-prune");
+        let source_root = path.split('/').next().expect("source sha").to_owned();
+        let marker = owner.base.join(INDEX).join(sha("bucket")).join(&source_root);
+        let visible_marker = root.path().join(INDEX).join(sha("bucket")).join(&source_root);
+
+        // Stand in for a prune that owns the marker lease and already read "no pin".
+        let prune_lease = os::acquire_namespace_mutation_lease_with_owner(&marker, None).await;
+        assert!(!owner.is_publishing(&marker));
+        let writer = tokio::spawn({
+            let disk = Arc::clone(&disk);
+            let path = path.clone();
+            async move { disk.write_metadata("bucket", RUSTFS_META_MULTIPART_BUCKET, &path, info).await }
+        });
+        tokio::time::timeout(std::time::Duration::from_secs(5), async {
+            while !owner.is_publishing(&marker) {
+                tokio::time::sleep(std::time::Duration::from_millis(2)).await;
+            }
+        })
+        .await
+        .expect("the publisher pins while the marker lease is still held by the prune");
+        assert!(!visible_marker.exists(), "the marker is recorded only under the lease");
+
+        drop(prune_lease);
+        writer.await.expect("writer task").expect("the upload metadata is published");
+        assert!(visible_marker.exists(), "the record after the prune recreates the marker");
+        let page = owner.page("bucket", None, Uuid::new_v4()).await.expect("page");
+        assert_eq!(page.roots, vec![source_root]);
+    }
+
+    /// Stale markers at the start of a page, at its cursor, and filling a whole
+    /// page: every page still satisfies the receiver's contract and advances.
+    #[tokio::test]
+    async fn multipart_index_pages_of_stale_markers_keep_valid_cursors() {
+        let root = tempfile::tempdir().expect("disk root");
+        let disk = open(root.path()).await;
+        let directory = root.path().join(INDEX).join(sha("bucket"));
+        std::fs::create_dir_all(&directory).expect("marker directory");
+        // Page one is entirely stale. Page two is live except its first name
+        // and the name its cursor lands on. Two live names follow.
+        let total = MULTIPART_INDEX_PAGE_SIZE * 2 + 2;
+        let stale = |i: usize| i <= MULTIPART_INDEX_PAGE_SIZE || i == MULTIPART_INDEX_PAGE_SIZE * 2 - 1;
+        for i in 0..total {
+            let name = format!("{i:064x}");
+            std::fs::write(directory.join(&name), b"1").expect("marker");
+            if !stale(i) {
+                std::fs::create_dir_all(root.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&name)).expect("staging root");
+            }
+        }
+        let disk_id = Uuid::new_v4();
+        let first = disk.multipart_index.page("bucket", None, disk_id).await.expect("first page");
+        first
+            .validate("bucket", None, disk_id)
+            .expect("an empty page with a cursor validates");
+        assert!(first.roots.is_empty(), "a fully stale page has no roots");
+        let cursor = first.next_marker.clone().expect("more markers follow a fully stale page");
+        assert_eq!(cursor, format!("{:064x}", MULTIPART_INDEX_PAGE_SIZE - 1));
+
+        let second = disk
+            .multipart_index
+            .page("bucket", Some(&cursor), disk_id)
+            .await
+            .expect("second page");
+        second
+            .validate("bucket", Some(&cursor), disk_id)
+            .expect("stale first and cursor names validate");
+        assert_eq!(second.roots.len(), MULTIPART_INDEX_PAGE_SIZE - 2);
+        assert_eq!(second.roots.first(), Some(&format!("{:064x}", MULTIPART_INDEX_PAGE_SIZE + 1)));
+        let cursor = second
+            .next_marker
+            .clone()
+            .expect("the cursor is the stale last examined name");
+        assert_eq!(cursor, format!("{:064x}", MULTIPART_INDEX_PAGE_SIZE * 2 - 1));
+        assert!(second.roots.last() < Some(&cursor));
+
+        let third = disk
+            .multipart_index
+            .page("bucket", Some(&cursor), disk_id)
+            .await
+            .expect("third page");
+        third.validate("bucket", Some(&cursor), disk_id).expect("last page validates");
+        assert_eq!(third.roots.len(), 2);
+        assert!(third.next_marker.is_none());
+        let remaining = std::fs::read_dir(&directory).expect("marker directory").count();
+        assert_eq!(remaining, total - MULTIPART_INDEX_PAGE_SIZE - 2, "every stale marker is pruned");
     }
 
     #[tokio::test]
@@ -1015,7 +1290,6 @@ mod tests {
                 std::fs::create_dir_all(root.path().join(RUSTFS_META_MULTIPART_BUCKET).join(&name)).expect("staging root");
             }
         }
-        age_marker(&directory.join(&stale));
         let disk_id = Uuid::new_v4();
         let first = disk.multipart_index.page("bucket", None, disk_id).await.expect("first page");
         first

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

@@ -813,7 +813,8 @@ impl SourceStore {
         Ok(())
     }
     pub async fn replace_head(&self, bytes: &[u8]) -> Result<()> {
-        self.multipart_index
+        let index_pin = self
+            .multipart_index
             .record_head(&format!("{}/xl.meta", self.path), bytes)
             .await?;
         invalidate_active_multipart_source_head(&self.source);
@@ -822,6 +823,7 @@ impl SourceStore {
         let root = self.publication_root.clone();
         let bytes = bytes.to_vec();
         os::run_blocking_namespace_operation(self.lease.clone(), move || {
+            let _index_pin = index_pin;
             os::replace_multipart_source_metadata(&path, &bytes, &base, &root)
         })
         .await

crates/gateway-adapter/Cargo.toml

@@ -0,0 +1,48 @@
+# 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.
+
+[package]
+name = "rustfs-gateway-adapter"
+version.workspace = true
+edition.workspace = true
+license.workspace = true
+repository.workspace = true
+rust-version.workspace = true
+homepage.workspace = true
+description = "RustFS implementations of the rustfs-gateway extension traits (credentials, policy facts, virtual hosts, bucket owner and request id)."
+keywords = ["rustfs", "gateway", "s3", "iam"]
+categories = ["web-programming", "authentication"]
+authors.workspace = true
+publish = false
+
+[package.metadata.gateway]
+# Documentary only: this crate is a ring-2 consumer of the gateway (it depends on
+# RustFS business crates). The RustFS repository does not run the gateway's ring guard.
+ring = 2
+
+[lints]
+workspace = true
+
+[dependencies]
+http = { workspace = true }
+rustfs-gateway = { workspace = true }
+rustfs-credentials = { workspace = true }
+rustfs-iam = { workspace = true }
+rustfs-policy = { workspace = true }
+serde_json = { workspace = true }
+
+[dev-dependencies]
+rustfs-gateway-http = { workspace = true }
+time = { workspace = true }
+tokio = { workspace = true, features = ["macros", "rt"] }

crates/gateway-adapter/src/bucket_owner.rs

@@ -0,0 +1,215 @@
+// 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.
+
+//! The gateway's [`BucketOwnerSource`]: the one owner RustFS reports, for a bucket that exists.
+//!
+//! Responsible for: confirming through a [`BucketLookup`] that a bucket exists, answering the
+//! single owner id the deployment reports ([`RustfsBucketOwners::new`]) for it, and turning every
+//! outcome that is not a usable owner into the gateway's one refusal, [`BucketOwnerError`].
+//! NOT responsible for: choosing the owner id (the binary passes the constant it already reports
+//! as the owner in listings), comparing it with `x-amz-expected-bucket-owner` (the gateway does,
+//! byte for byte), or authorizing anything.
+//! Upstream: the gateway's expected-owner check, which consults this source only when a request
+//! carries the header, after authorization allowed it. Downstream: the lookup the binary supplies
+//! over the ecstore bucket metadata system.
+//!
+//! # What an owner is in RustFS
+//!
+//! RustFS stores no per-bucket owner: bucket ACL writes are no-ops and nothing writes the ACL
+//! document, so the only owner RustFS reports anywhere (ListBuckets, ACL and multipart listings)
+//! is one constant. The source answers exactly that constant. A caller therefore passes the
+//! gateway's check only by presenting the owner id RustFS lists, which separates no tenants; it is
+//! a compatibility answer, and the constant's own documentation says it must not decide
+//! authorization.
+//!
+//! # Difference from legacy (ledger entry R-071)
+//!
+//! Legacy RustFS ignores `x-amz-expected-bucket-owner` entirely. On the gateway stack a request
+//! carrying it is served only when the value equals the reported owner id; otherwise it is `403
+//! AccessDenied`. A missing bucket and a backend failure with the header present are also `403`
+//! at the pinned gateway, where legacy answers `404` and `5xx`: [`BucketOwnerError`] has one value,
+//! so it cannot express either. [`RustfsBucketOwners::resolve`] keeps the two apart at this
+//! adapter's boundary so the gateway can surface them once it grows the classes.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{BoxFuture, BucketName, BucketOwnerError, BucketOwnerSource};
+
+/// Why a bucket yielded no owner.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum OwnerLookupError {
+    /// The bucket does not exist.
+    NoSuchBucket,
+    /// The metadata backend failed, could not be reached, or the owner is not usable (an empty
+    /// owner id is a misconfiguration), so the bucket's state is unknown.
+    Backend,
+}
+
+/// The narrow slice of RustFS bucket metadata an owner answer needs: whether a bucket exists.
+///
+/// Exists so the source can be tested against a fake. The binary implements it over the ecstore
+/// bucket metadata system, mapping `BucketNotFound` to [`OwnerLookupError::NoSuchBucket`] and every
+/// other error to [`OwnerLookupError::Backend`].
+pub trait BucketLookup: Send + Sync + 'static {
+    /// `Ok(())` when `bucket` exists.
+    fn check<'a>(&'a self, bucket: &'a str) -> BoxFuture<'a, Result<(), OwnerLookupError>>;
+}
+
+/// A [`BucketOwnerSource`] answering one owner id for every bucket that exists.
+pub struct RustfsBucketOwners<L> {
+    lookup: L,
+    owner: Arc<str>,
+}
+
+impl<L: BucketLookup> RustfsBucketOwners<L> {
+    /// A source answering `owner` for every bucket `lookup` finds.
+    ///
+    /// An empty `owner` is no owner: every answer is then [`OwnerLookupError::Backend`].
+    #[must_use]
+    pub fn new(lookup: L, owner: impl Into<Arc<str>>) -> Self {
+        Self {
+            lookup,
+            owner: owner.into(),
+        }
+    }
+
+    /// The owner of `bucket`, or why there is none. Missing and failing stay distinct here.
+    ///
+    /// # Errors
+    ///
+    /// [`OwnerLookupError::NoSuchBucket`] when the bucket does not exist;
+    /// [`OwnerLookupError::Backend`] on any lookup failure or when the configured owner is empty.
+    pub async fn resolve(&self, bucket: &str) -> Result<Arc<str>, OwnerLookupError> {
+        self.lookup.check(bucket).await?;
+        if self.owner.is_empty() {
+            return Err(OwnerLookupError::Backend);
+        }
+        Ok(Arc::clone(&self.owner))
+    }
+}
+
+impl<L: BucketLookup> BucketOwnerSource for RustfsBucketOwners<L> {
+    fn owner<'a>(&'a self, bucket: &'a BucketName) -> BoxFuture<'a, Result<Arc<str>, BucketOwnerError>> {
+        Box::pin(async move {
+            // The pinned gateway collapses both classes (R-071), so both are the one refusal.
+            self.resolve(bucket.as_str())
+                .await
+                .map_err(|_: OwnerLookupError| BucketOwnerError::unavailable())
+        })
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use std::sync::Mutex;
+
+    /// Stands for the owner id RustFS reports; the binary passes its own constant.
+    const OWNER: &str = "reported-owner-id";
+
+    struct Fake {
+        answer: Result<(), OwnerLookupError>,
+        asked: Mutex<Vec<String>>,
+    }
+
+    impl Fake {
+        fn answering(answer: Result<(), OwnerLookupError>) -> Self {
+            Self {
+                answer,
+                asked: Mutex::new(Vec::new()),
+            }
+        }
+    }
+
+    impl BucketLookup for Fake {
+        fn check<'a>(&'a self, bucket: &'a str) -> BoxFuture<'a, Result<(), OwnerLookupError>> {
+            self.asked.lock().unwrap().push(bucket.to_owned());
+            let answer = self.answer;
+            Box::pin(async move { answer })
+        }
+    }
+
+    fn source(answer: Result<(), OwnerLookupError>) -> RustfsBucketOwners<Fake> {
+        RustfsBucketOwners::new(Fake::answering(answer), OWNER)
+    }
+
+    async fn gateway_answer(answer: Result<(), OwnerLookupError>) -> Result<Arc<str>, BucketOwnerError> {
+        source(answer).owner(&BucketName::new("photos").expect("a bucket name")).await
+    }
+
+    /// Positive: an existing bucket's owner is the configured id, byte for byte.
+    #[tokio::test]
+    async fn an_existing_bucket_has_the_configured_owner() {
+        assert_eq!(gateway_answer(Ok(())).await.as_deref(), Ok(OWNER));
+    }
+
+    /// Positive: every bucket gets the same owner, since RustFS stores none per bucket.
+    #[tokio::test]
+    async fn every_bucket_has_the_same_owner() {
+        let source = source(Ok(()));
+        assert_eq!(source.resolve("photos").await.as_deref(), Ok(OWNER));
+        assert_eq!(source.resolve("backups").await.as_deref(), Ok(OWNER));
+    }
+
+    /// Positive: the lookup is asked about exactly the bucket the gateway named.
+    #[tokio::test]
+    async fn the_lookup_is_asked_about_the_named_bucket() {
+        let source = source(Ok(()));
+        source
+            .owner(&BucketName::new("photos").expect("a bucket name"))
+            .await
+            .expect("an owner");
+        assert_eq!(*source.lookup.asked.lock().unwrap(), vec!["photos".to_owned()]);
+    }
+
+    /// Negative: a missing bucket is reported as missing at the adapter boundary.
+    #[tokio::test]
+    async fn a_missing_bucket_is_reported_as_missing() {
+        assert_eq!(
+            source(Err(OwnerLookupError::NoSuchBucket)).resolve("photos").await,
+            Err(OwnerLookupError::NoSuchBucket)
+        );
+    }
+
+    /// Negative: a failing backend is reported as a failure, never as an owner and never as missing.
+    #[tokio::test]
+    async fn a_backend_failure_is_reported_as_a_failure() {
+        assert_eq!(
+            source(Err(OwnerLookupError::Backend)).resolve("photos").await,
+            Err(OwnerLookupError::Backend)
+        );
+    }
+
+    /// Negative: the pinned gateway sees one refusal for both, and never an owner.
+    #[tokio::test]
+    async fn the_gateway_is_refused_in_both_cases() {
+        assert_eq!(
+            gateway_answer(Err(OwnerLookupError::NoSuchBucket)).await,
+            Err(BucketOwnerError::unavailable())
+        );
+        assert_eq!(gateway_answer(Err(OwnerLookupError::Backend)).await, Err(BucketOwnerError::unavailable()));
+    }
+
+    /// Negative: an empty configured owner is no owner; it must not equal an empty assertion.
+    #[tokio::test]
+    async fn an_empty_owner_is_refused() {
+        let source = RustfsBucketOwners::new(Fake::answering(Ok(())), "");
+        assert_eq!(source.resolve("photos").await, Err(OwnerLookupError::Backend));
+        assert_eq!(
+            source.owner(&BucketName::new("photos").expect("a bucket name")).await,
+            Err(BucketOwnerError::unavailable())
+        );
+    }
+}

crates/gateway-adapter/src/credentials.rs

@@ -0,0 +1,446 @@
+// 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.
+
+//! The gateway's [`CredentialProvider`], answered by RustFS IAM.
+//!
+//! Responsible for: resolving one access key id to the secret the gateway verifies a signature
+//! with, and classifying every other outcome as either a business result
+//! ([`CredentialLookup::NotFound`], eligible for the gateway's negative cache) or an operational
+//! fault ([`ProviderError`], counted and never cached).
+//! NOT responsible for: authorization of any kind, session-token claim validation, or deciding
+//! what the client is answered with. The gateway answers every unusable credential identically.
+//!
+//! # Secret handling
+//!
+//! Secrets travel only inside the gateway's [`Credentials`], whose `Debug` redacts them. No type
+//! in this module derives or implements `Debug`, and no error produced here carries text: the
+//! IAM error is classified and dropped. A secret that fails the crate's effective-length rule
+//! (`crate::MIN_SECRET_KEY_LEN` bytes once trailing NULs are removed) is never handed to the
+//! gateway, whatever the lookup says about its validity.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{BoxFuture, CredentialLookup, CredentialProvider, Credentials, ProviderError, SessionBinding};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::store::Store;
+use rustfs_iam::sys::IamSys;
+use rustfs_policy::auth::{ACCOUNT_OFF, UserIdentity};
+
+use crate::root::{RootSource, is_usable_secret};
+
+/// The opaque issuer recorded in the [`rustfs_gateway::SessionBinding`] of every RustFS STS
+/// credential.
+pub const STS_SESSION_ISSUER: &str = "rustfs-sts";
+
+/// Access keys with this prefix belong to Keystone token authentication, which never verifies an
+/// AWS signature.
+const KEYSTONE_ACCESS_KEY_PREFIX: &str = "keystone:";
+
+/// The narrow slice of IAM that a credential lookup needs.
+///
+/// Exists so the adapter can be tested against a fake: IAM exports no in-memory backend.
+/// Implemented for [`IamSys`], whose `check_key` loads a user from storage on a cache miss.
+pub trait IamKeyLookup: Send + Sync + 'static {
+    /// Whether the IAM store has finished loading.
+    fn is_ready(&self) -> bool;
+
+    /// Resolves an access key, with [`IamSys::check_key`]'s contract: `Ok((None, _))` for an
+    /// unknown key, `Ok((Some(identity), valid))` otherwise, where `valid` is false for a
+    /// disabled, expired, or malformed credential.
+    fn check_key<'a>(&'a self, access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>>;
+}
+
+impl<T: Store> IamKeyLookup for IamSys<T> {
+    fn is_ready(&self) -> bool {
+        IamSys::is_ready(self)
+    }
+
+    fn check_key<'a>(&'a self, access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+        Box::pin(IamSys::check_key(self, access_key))
+    }
+}
+
+/// A [`CredentialProvider`] over RustFS IAM and the live root credentials.
+///
+/// Built only by [`crate::IamAdapters::new`], so it always reads the same [`RootSource`] as the
+/// policy source it was assembled with.
+pub struct IamCredentials {
+    iam: Arc<dyn IamKeyLookup>,
+    root: Arc<dyn RootSource>,
+}
+
+impl IamCredentials {
+    pub(crate) fn new(iam: Arc<dyn IamKeyLookup>, root: Arc<dyn RootSource>) -> Self {
+        Self { iam, root }
+    }
+
+    async fn resolve(&self, access_key: &str) -> Result<CredentialLookup, ProviderError> {
+        // Deliberate change from the legacy s3s path (`rustfs/src/auth.rs` `get_secret_key`),
+        // which answered a `keystone:` key with an empty secret. Keystone authenticates by token,
+        // so there is no AWS secret to verify a signature against: the key is unknown here.
+        if access_key.starts_with(KEYSTONE_ACCESS_KEY_PREFIX) {
+            return Ok(CredentialLookup::NotFound);
+        }
+
+        // Legacy answers every key with an internal error while the root credentials are not
+        // initialized; without them no key can be classified as root or not.
+        let Some(root) = self.root.current() else {
+            return Err(ProviderError::Unavailable);
+        };
+        // Byte-exact, as the trait requires: the access key id is signed material.
+        if root.access_key() == access_key {
+            return Ok(CredentialLookup::Found(root.credentials().clone_credentials()));
+        }
+
+        if !self.iam.is_ready() {
+            return Err(ProviderError::Unavailable);
+        }
+
+        match self.iam.check_key(access_key).await {
+            // IAM refuses such a secret through `valid`, but this provider does not rely on every
+            // lookup doing so: a short secret is never usable, so it is not a credential at all.
+            Ok((Some(identity), _)) if !is_usable_secret(identity.credentials.secret_key.as_bytes()) => {
+                Ok(CredentialLookup::NotFound)
+            }
+            Ok((Some(identity), valid)) => to_gateway_credentials(access_key, &identity, valid).map(CredentialLookup::Found),
+            Ok((None, _)) => Ok(CredentialLookup::NotFound),
+            Err(error) => classify_iam_error(&error),
+        }
+    }
+}
+
+/// Classifies an IAM failure without keeping any of its text.
+fn classify_iam_error(error: &IamError) -> Result<CredentialLookup, ProviderError> {
+    match error {
+        IamError::NoSuchUser(_)
+        | IamError::NoSuchServiceAccount(_)
+        | IamError::NoSuchTempAccount(_)
+        | IamError::NoSuchAccount(_) => Ok(CredentialLookup::NotFound),
+        IamError::IamSysNotInitialized => Err(ProviderError::Unavailable),
+        _ => Err(ProviderError::Backend),
+    }
+}
+
+/// Builds the gateway credential for one IAM identity. No authorization happens here: a
+/// disabled or invalid principal still resolves, marked disabled, and the gateway refuses it
+/// exactly as it refuses an unknown key.
+fn to_gateway_credentials(access_key: &str, identity: &UserIdentity, valid: bool) -> Result<Credentials, ProviderError> {
+    let cred = &identity.credentials;
+    let mut credentials = Credentials::new(access_key, cred.secret_key.as_bytes()).map_err(|_| ProviderError::Backend)?;
+
+    // A service account carries a claims token internally, but its clients never present it, so
+    // it authenticates as a long-term credential (as `check_session_token` treats it in legacy).
+    if !cred.session_token.is_empty() && !cred.is_service_account() {
+        // A temporary credential without an expiry cannot become a gateway session, which always
+        // ends. Refusing it as a fault is the fail-closed answer; a never-ending session is not.
+        let expiration = cred.expiration.ok_or(ProviderError::Backend)?;
+        let binding = SessionBinding::new(STS_SESSION_ISSUER, expiration.unix_timestamp()).map_err(|_| ProviderError::Backend)?;
+        credentials = credentials
+            .with_session(&cred.session_token, binding)
+            .map_err(|_| ProviderError::Backend)?;
+    }
+
+    if !valid || cred.status == ACCOUNT_OFF {
+        credentials = credentials.disable();
+    }
+    Ok(credentials)
+}
+
+impl CredentialProvider for IamCredentials {
+    fn lookup<'a>(&'a self, access_key_id: &'a str) -> BoxFuture<'a, Result<CredentialLookup, ProviderError>> {
+        Box::pin(self.resolve(access_key_id))
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use crate::root::{RootCell, RootCredentials};
+    use rustfs_credentials::Credentials as IamCredential;
+    use rustfs_gateway::RequestNow;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use time::OffsetDateTime;
+
+    const ACCESS_KEY: &str = "AKIDUSER0001";
+    const SECRET: &str = "user-secret-key-0001";
+
+    type Answer = Box<dyn Fn() -> Result<(Option<UserIdentity>, bool), IamError> + Send + Sync>;
+
+    struct FakeIam {
+        ready: bool,
+        answer: Answer,
+        calls: AtomicUsize,
+    }
+
+    impl IamKeyLookup for FakeIam {
+        fn is_ready(&self) -> bool {
+            self.ready
+        }
+
+        fn check_key<'a>(&'a self, _access_key: &'a str) -> BoxFuture<'a, Result<(Option<UserIdentity>, bool), IamError>> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            let answer = (self.answer)();
+            Box::pin(async move { answer })
+        }
+    }
+
+    fn fake(
+        ready: bool,
+        answer: impl Fn() -> Result<(Option<UserIdentity>, bool), IamError> + Send + Sync + 'static,
+    ) -> Arc<FakeIam> {
+        Arc::new(FakeIam {
+            ready,
+            answer: Box::new(answer),
+            calls: AtomicUsize::new(0),
+        })
+    }
+
+    fn user(status: &str) -> IamCredential {
+        IamCredential {
+            access_key: ACCESS_KEY.to_owned(),
+            secret_key: SECRET.to_owned(),
+            status: status.to_owned(),
+            ..Default::default()
+        }
+    }
+
+    fn found(cred: IamCredential, valid: bool) -> Result<(Option<UserIdentity>, bool), IamError> {
+        Ok((Some(UserIdentity::new(cred)), valid))
+    }
+
+    fn root_cell() -> Arc<RootCell> {
+        let root = RootCredentials::new("rootadmin", "root-secret-key", "sts-signing-key").expect("valid root credentials");
+        Arc::new(RootCell::new(Some(root)))
+    }
+
+    async fn lookup(iam: Arc<FakeIam>, access_key: &str) -> Result<CredentialLookup, ProviderError> {
+        IamCredentials::new(iam, root_cell()).lookup(access_key).await
+    }
+
+    fn expect_found(result: Result<CredentialLookup, ProviderError>) -> Credentials {
+        match result {
+            Ok(CredentialLookup::Found(credentials)) => credentials,
+            Ok(CredentialLookup::NotFound) => panic!("expected Found, got NotFound"),
+            Err(error) => panic!("expected Found, got {error:?}"),
+        }
+    }
+
+    fn is_not_found(result: &Result<CredentialLookup, ProviderError>) -> bool {
+        matches!(result, Ok(CredentialLookup::NotFound))
+    }
+
+    /// Positive — an enabled user resolves to exactly its secret, as a long-term credential.
+    #[tokio::test]
+    async fn credentials_found_user_resolves_to_its_secret() {
+        let iam = fake(true, || found(user("on"), true));
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        assert_eq!(credentials.secret().expose(), SECRET.as_bytes());
+        assert_eq!(credentials.identity().access_key_id(), ACCESS_KEY);
+        assert!(!credentials.is_disabled());
+        assert!(credentials.session_binding().is_none());
+    }
+
+    /// Negative — a key IAM does not know is a business result, not a fault.
+    #[tokio::test]
+    async fn credentials_unknown_key_is_not_found() {
+        let iam = fake(true, || Ok((None, false)));
+        assert!(is_not_found(&lookup(iam, "AKIDUNKNOWN").await));
+    }
+
+    /// Negative — every "no such principal" error IAM can surface is `NotFound`. Answering these
+    /// as `Backend` would keep them out of the negative cache and count every guessed key as a
+    /// backend fault.
+    #[tokio::test]
+    async fn credentials_no_such_principal_errors_are_not_found() {
+        let errors: [fn() -> IamError; 4] = [
+            || IamError::NoSuchUser(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchServiceAccount(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchTempAccount(ACCESS_KEY.to_owned()),
+            || IamError::NoSuchAccount(ACCESS_KEY.to_owned()),
+        ];
+        for error in errors {
+            let iam = fake(true, move || Err(error()));
+            let result = lookup(iam, ACCESS_KEY).await;
+            assert!(is_not_found(&result), "{:?} must be NotFound", error());
+        }
+    }
+
+    /// Negative — any other IAM failure is a backend fault, never `NotFound` and never `Found`.
+    #[tokio::test]
+    async fn credentials_backend_error_is_backend() {
+        let iam = fake(true, || Err(IamError::StringError("storage read failed".to_owned())));
+        assert!(matches!(lookup(iam, ACCESS_KEY).await, Err(ProviderError::Backend)));
+    }
+
+    /// Negative — a `keystone:` key never reaches IAM and never resolves to a secret.
+    ///
+    /// Deliberate change from the legacy s3s path, which answered such a key with an empty
+    /// secret: Keystone authenticates by token, so there is no AWS secret to verify against.
+    #[tokio::test]
+    async fn credentials_keystone_prefix_is_not_found_without_iam() {
+        let iam = fake(true, || found(user("on"), true));
+        let result = lookup(iam.clone(), "keystone:project-user").await;
+        assert!(is_not_found(&result));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    /// Negative — a switched-off or otherwise invalid user resolves, but disabled, and the same
+    /// user enabled does not.
+    #[tokio::test]
+    async fn credentials_disabled_user_is_disabled() {
+        let off = expect_found(lookup(fake(true, || found(user("off"), false)), ACCESS_KEY).await);
+        assert!(off.is_disabled());
+        assert!(off.admit(None, RequestNow::from_unix_seconds(0)).is_err());
+
+        let invalid = expect_found(lookup(fake(true, || found(user("on"), false)), ACCESS_KEY).await);
+        assert!(invalid.is_disabled());
+
+        let on = expect_found(lookup(fake(true, || found(user("on"), true)), ACCESS_KEY).await);
+        assert!(!on.is_disabled());
+    }
+
+    /// Positive — an STS credential carries its token and its lifetime as one session binding.
+    #[tokio::test]
+    async fn credentials_sts_carries_session_binding() {
+        let expires_at = 4_000_000_000_i64;
+        let iam = fake(true, move || {
+            let mut cred = user("on");
+            cred.session_token = "sts-session-token".to_owned();
+            cred.parent_user = "parent".to_owned();
+            cred.expiration = Some(OffsetDateTime::from_unix_timestamp(expires_at).expect("valid timestamp"));
+            found(cred, true)
+        });
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        let binding = credentials.session_binding().expect("an STS credential has a session");
+        assert_eq!(binding.expires_at_unix_seconds(), expires_at);
+        assert_eq!(binding.issuer(), STS_SESSION_ISSUER);
+        let now = RequestNow::from_unix_seconds(expires_at - 1);
+        assert!(credentials.admit(Some(b"sts-session-token"), now).is_ok());
+        assert!(credentials.admit(None, now).is_err());
+    }
+
+    /// Negative — an STS credential with no expiry cannot be expressed as a session the gateway
+    /// would ever expire, so it is a backend fault rather than a session that never ends.
+    #[tokio::test]
+    async fn credentials_sts_without_expiry_is_backend() {
+        let iam = fake(true, || {
+            let mut cred = user("on");
+            cred.session_token = "sts-session-token".to_owned();
+            found(cred, true)
+        });
+        assert!(matches!(lookup(iam, ACCESS_KEY).await, Err(ProviderError::Backend)));
+    }
+
+    /// Positive — a service account carries a token internally but authenticates as a long-term
+    /// credential: clients never present that token.
+    #[tokio::test]
+    async fn credentials_service_account_is_long_term() {
+        let iam = fake(true, || {
+            let mut cred = user("on");
+            cred.session_token = "service-account-claims-token".to_owned();
+            cred.parent_user = "parent".to_owned();
+            cred.claims = Some([(rustfs_credentials::IAM_POLICY_CLAIM_NAME_SA.to_owned(), "inherited-policy".into())].into());
+            found(cred, true)
+        });
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        assert!(credentials.session_binding().is_none());
+        assert!(credentials.admit(None, RequestNow::from_unix_seconds(0)).is_ok());
+    }
+
+    /// Negative — before IAM is ready a lookup is `Unavailable`, not `NotFound`: a key must not
+    /// be negatively cached because the store had not loaded yet.
+    #[tokio::test]
+    async fn credentials_iam_not_ready_is_unavailable() {
+        let iam = fake(false, || found(user("on"), true));
+        assert!(matches!(lookup(iam.clone(), ACCESS_KEY).await, Err(ProviderError::Unavailable)));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+
+        let uninitialized = fake(true, || Err(IamError::IamSysNotInitialized));
+        assert!(matches!(lookup(uninitialized, ACCESS_KEY).await, Err(ProviderError::Unavailable)));
+    }
+
+    /// Positive — the injected root credentials resolve without IAM, even before it is ready, and
+    /// nothing else is answered with the root secret.
+    #[tokio::test]
+    async fn credentials_root_resolves_without_iam() {
+        let iam = fake(false, || found(user("on"), true));
+        let provider = IamCredentials::new(iam.clone(), root_cell());
+        let root = expect_found(provider.lookup("rootadmin").await);
+        assert_eq!(root.secret().expose(), b"root-secret-key");
+        assert!(matches!(provider.lookup("ROOTADMIN").await, Err(ProviderError::Unavailable)));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    /// Negative — a lookup that calls an identity valid but holds an empty or short secret is not
+    /// trusted: such a secret is not a credential, so the key is `NotFound`, never `Found`.
+    #[tokio::test]
+    async fn credentials_short_secret_is_never_found() {
+        for weak in ["", "1234567", "\0\0\0\0\0\0\0\0", "abc\0\0\0\0\0"] {
+            let iam = fake(true, move || {
+                let mut cred = user("on");
+                cred.secret_key = weak.to_owned();
+                found(cred, true)
+            });
+            let result = lookup(iam, ACCESS_KEY).await;
+            assert!(is_not_found(&result), "{weak:?} must be NotFound");
+        }
+        for usable in ["12345678", "abc\0defg"] {
+            let iam = fake(true, move || {
+                let mut cred = user("on");
+                cred.secret_key = usable.to_owned();
+                found(cred, true)
+            });
+            let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+            assert_eq!(credentials.secret().expose(), usable.as_bytes());
+        }
+    }
+
+    /// Negative — a switched-off account is disabled even when the lookup calls it valid.
+    #[tokio::test]
+    async fn credentials_off_status_alone_disables() {
+        let off = expect_found(lookup(fake(true, || found(user("off"), true)), ACCESS_KEY).await);
+        assert!(off.is_disabled());
+        assert!(off.admit(None, RequestNow::from_unix_seconds(0)).is_err());
+    }
+
+    /// Negative — an STS credential admits only its own session's token: the token of another
+    /// session, or none, is refused.
+    #[tokio::test]
+    async fn credentials_sts_refuses_another_sessions_token() {
+        let expires_at = 4_000_000_000_i64;
+        let iam = fake(true, move || {
+            let mut cred = user("on");
+            cred.session_token = "session-token-a".to_owned();
+            cred.parent_user = "parent".to_owned();
+            cred.expiration = Some(OffsetDateTime::from_unix_timestamp(expires_at).expect("valid timestamp"));
+            found(cred, true)
+        });
+        let credentials = expect_found(lookup(iam, ACCESS_KEY).await);
+        let now = RequestNow::from_unix_seconds(expires_at - 1);
+        assert!(credentials.admit(Some(b"session-token-b"), now).is_err());
+        assert!(credentials.admit(Some(b"session-token-a"), now).is_ok());
+    }
+
+    /// Negative — without root credentials no key can be classified, so every lookup is
+    /// `Unavailable` and IAM is not asked.
+    #[tokio::test]
+    async fn credentials_without_root_are_unavailable() {
+        let iam = fake(true, || found(user("on"), true));
+        let provider = IamCredentials::new(iam.clone(), Arc::new(RootCell::new(None)));
+        assert!(matches!(provider.lookup(ACCESS_KEY).await, Err(ProviderError::Unavailable)));
+        assert!(matches!(provider.lookup("rootadmin").await, Err(ProviderError::Unavailable)));
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+}

crates/gateway-adapter/src/host.rs

@@ -0,0 +1,306 @@
+// 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.
+
+//! The gateway's [`HostResolver`], built from RustFS's configured domain list.
+//!
+//! Responsible for: turning the `RUSTFS_SERVER_DOMAINS` list into the gateway's legacy RustFS
+//! virtual-host resolver.
+//! NOT responsible for: parsing a `Host` header, matching a domain, validating bucket names (all
+//! done by [`LegacyRustfsVirtualHosts`]), reading the environment, or deciding that a console
+//! deployment serves no domains (the caller passes an empty list then, as `s3_host_domains` in
+//! `rustfs/src/server/http.rs` returns `None`). An empty list needs no branch here:
+//! [`LegacyRustfsVirtualHosts`] built from no domain reads every host path-style, exactly as the
+//! gateway's `PathStyleOnly` does.
+//! Upstream: the RustFS configuration (`Config::server_domains`). Downstream: the gateway's
+//! `ServiceBuilder`, which holds the resolver as `Arc<dyn HostResolver>`.
+//!
+//! # What "the same as `MultiDomain`" means
+//!
+//! RustFS today hands the same list to the s3s framework's multi-domain host, which ignores ports
+//! on both sides when matching a domain, takes the whole prefix in front of a matched domain as
+//! the bucket, compares a domain exactly (so a host is matched on the case it arrives in), and,
+//! for a host no domain matches, falls back to CNAME-style: a host that is itself a bucket name
+//! (lower-cased) is that bucket. [`LegacyRustfsVirtualHosts`] reproduces exactly that.
+//!
+//! What reads **path-style**: an empty domain list; a configured domain itself, with or without a
+//! port; an IP address, or a socket address (`10.0.0.1`, `[::1]:9000`); a host that is a valid
+//! domain but, lower-cased, is no bucket name (it carries a port, is over 63 bytes, or starts
+//! with `xn--`). What reads as a **bucket**: anything under a configured domain, and any other
+//! bucket-shaped host, which is a bucket of its own (`localhost`, `badexample.com`, `EXAMPLE.COM`).
+//! What is **refused** before routing: a host that is no domain (`under_score.org`, a bare
+//! `[::1]`, a trailing dot), and a bucket under a domain that the bucket rules refuse
+//! (`PHOTOS.example.com`, `ab.example.com`). The tests below pin each of those readings.
+
+use std::sync::Arc;
+
+use rustfs_gateway::{HostResolver, LegacyDomainError, LegacyRustfsVirtualHosts};
+
+/// Builds the host resolver for RustFS's configured domain list.
+///
+/// An empty list reads every host path-style. Otherwise the list is read as legacy RustFS reads
+/// it: a later entry whose name without its port repeats an earlier one is dropped, and a list
+/// that legacy RustFS refuses at start-up is refused here.
+///
+/// # Errors
+///
+/// [`LegacyDomainError`] for an entry that is not a domain, or for two entries where one equals or
+/// is a subdomain of the other.
+pub fn rustfs_host_resolver<I, S>(domains: I) -> Result<Arc<dyn HostResolver>, LegacyDomainError>
+where
+    I: IntoIterator<Item = S>,
+    S: AsRef<str>,
+{
+    Ok(Arc::new(LegacyRustfsVirtualHosts::new(domains)?))
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use http::Method;
+    use rustfs_gateway::{Addressing, HostQuery, HostRefusal, TargetKind};
+
+    /// What the resolver reads `GET path` with `Host: host` as: the host-named bucket (`None` for
+    /// path style), the routed target, and the refusal, if any.
+    fn read(resolver: &dyn HostResolver, path: &str, host: &str) -> (Option<String>, TargetKind, Option<HostRefusal>) {
+        let method = Method::GET;
+        let request = http::Request::builder()
+            .method(method.clone())
+            .uri(path)
+            .header("host", host)
+            .body(())
+            .expect("a valid request");
+        let effective = rustfs_gateway_http::effective_host(&request).expect("an acceptable host");
+        let query = HostQuery {
+            host: &effective,
+            path,
+            method: &method,
+        };
+        let resolved = resolver.resolve(&query);
+        let bucket = match &resolved.addressing {
+            Addressing::VirtualHosted { bucket, .. } => Some(bucket.as_str().to_owned()),
+            Addressing::Path => None,
+        };
+        (bucket, resolved.target, resolver.refusal(&query))
+    }
+
+    fn with_domains(domains: &[&str]) -> Arc<dyn HostResolver> {
+        rustfs_host_resolver(domains).expect("a list RustFS accepts")
+    }
+
+    fn bucket(domains: &[&str], host: &str) -> Option<String> {
+        read(&*with_domains(domains), "/key", host).0
+    }
+
+    /// Positive: a host under a configured domain names the bucket in front of it.
+    #[test]
+    fn host_under_domain_names_the_bucket() {
+        assert_eq!(
+            read(&*with_domains(&["example.com"]), "/key", "photos.example.com"),
+            (Some("photos".to_owned()), TargetKind::Object, None)
+        );
+        assert_eq!(read(&*with_domains(&["example.com"]), "/", "photos.example.com").1, TargetKind::Bucket);
+    }
+
+    /// Positive: the port on the host, or on the configured domain, does not take part in matching.
+    #[test]
+    fn ports_are_ignored_on_both_sides() {
+        assert_eq!(bucket(&["example.com"], "photos.example.com:9000").as_deref(), Some("photos"));
+        assert_eq!(bucket(&["example.com:9000"], "photos.example.com").as_deref(), Some("photos"));
+        assert_eq!(bucket(&["example.com:9000"], "photos.example.com:1234").as_deref(), Some("photos"));
+    }
+
+    /// Positive: the whole prefix in front of the domain is the bucket, dots kept.
+    #[test]
+    fn whole_prefix_is_the_bucket() {
+        assert_eq!(bucket(&["example.com"], "my.dotted.bkt.example.com").as_deref(), Some("my.dotted.bkt"));
+    }
+
+    /// Positive: with several domains, each one resolves its own subdomains.
+    #[test]
+    fn every_configured_domain_resolves() {
+        let domains = ["example.com", "s3.local"];
+        assert_eq!(bucket(&domains, "abc.example.com").as_deref(), Some("abc"));
+        assert_eq!(bucket(&domains, "xyz.s3.local").as_deref(), Some("xyz"));
+    }
+
+    /// Positive: the domain itself is the endpoint, so the bucket comes from the path.
+    #[test]
+    fn the_domain_itself_is_path_style() {
+        assert_eq!(bucket(&["example.com"], "example.com"), None);
+        assert_eq!(bucket(&["example.com"], "example.com:9000"), None);
+    }
+
+    /// Negative: no configured domain means path style only, for every host shape.
+    #[test]
+    fn no_domain_is_path_style_only() {
+        for host in ["photos.example.com", "localhost", "localhost:9000", "10.0.0.1"] {
+            assert_eq!(bucket(&[], host), None, "{host}");
+        }
+    }
+
+    /// Negative: an address, and a host carrying a port that no domain matches, are path style.
+    #[test]
+    fn addresses_and_unmatched_ports_are_path_style() {
+        for host in ["10.0.0.1", "10.0.0.1:9000", "localhost:9000", "other.org:9000"] {
+            assert_eq!(bucket(&["example.com"], host), None, "{host}");
+        }
+    }
+
+    /// Negative: a host that only ends in the same letters as a domain is not under it. It is not
+    /// stripped to `bad`; the only reading it gets is the CNAME-style one, the whole host.
+    #[test]
+    fn suffix_lookalike_is_not_under_the_domain() {
+        assert_eq!(bucket(&["example.com"], "badexample.com").as_deref(), Some("badexample.com"));
+        assert_eq!(bucket(&["example.com"], "example.com.evil.org").as_deref(), Some("example.com.evil.org"));
+    }
+
+    /// Negative: matching a configured domain is exact on case, as the legacy multi-domain host does it, so a
+    /// mixed-case host is not stripped to its prefix; it is the lower-cased whole host.
+    #[test]
+    fn mixed_case_host_is_not_stripped_to_a_prefix() {
+        assert_eq!(bucket(&["example.com"], "Photos.Example.COM").as_deref(), Some("photos.example.com"));
+    }
+
+    /// Negative: a host that is no domain at all is refused, not routed.
+    #[test]
+    fn a_host_that_is_no_domain_is_refused() {
+        let resolver = with_domains(&["example.com"]);
+        assert_eq!(read(&*resolver, "/k", "under_score.org").2, Some(HostRefusal::UnusableHost));
+    }
+
+    /// Negative: a bucket the host names that bucket rules refuse is refused.
+    #[test]
+    fn a_host_named_bucket_the_rules_refuse_is_refused() {
+        let resolver = with_domains(&["example.com"]);
+        assert_eq!(read(&*resolver, "/k", "ab.example.com").2, Some(HostRefusal::RefusedBucket));
+        assert_eq!(read(&*resolver, "/k", "bad_host.example.com").2, Some(HostRefusal::RefusedBucket));
+    }
+
+    /// Negative: a list legacy RustFS refuses at start-up is refused.
+    #[test]
+    fn domain_lists_legacy_rustfs_refuses_are_refused() {
+        for list in [
+            vec!["example.com", "a.example.com"],
+            vec!["example.com", "com"],
+            vec!["bad_domain"],
+            vec![""],
+            vec!["example.com:notaport"],
+        ] {
+            assert!(rustfs_host_resolver(&list).is_err(), "{list:?}");
+        }
+    }
+
+    /// Positive: a later repeat of a name, whatever its port, is dropped rather than refused, as
+    /// `s3_host_domains` drops it before the legacy multi-domain host sees the list.
+    #[test]
+    fn a_repeat_of_a_domain_is_dropped() {
+        assert_eq!(
+            bucket(&["example.com:9411", "example.com:9000"], "xyz.example.com").as_deref(),
+            Some("xyz")
+        );
+    }
+
+    /// The refusal alone, for hosts whose bucket reading is not the point.
+    fn refusal(domains: &[&str], host: &str) -> Option<HostRefusal> {
+        read(&*with_domains(domains), "/k", host).2
+    }
+
+    /// Positive: a bucket-shaped host no domain matches is a bucket of its own, `localhost` and
+    /// an upper-case spelling of the domain among them.
+    #[test]
+    fn bucket_shaped_hosts_outside_every_domain_are_their_own_bucket() {
+        assert_eq!(bucket(&["example.com"], "localhost").as_deref(), Some("localhost"));
+        assert_eq!(bucket(&["example.com"], "EXAMPLE.COM").as_deref(), Some("example.com"));
+        assert_eq!(bucket(&["example.com"], "cdn.other.org").as_deref(), Some("cdn.other.org"));
+    }
+
+    /// Negative: a bucket the rules refuse under a domain is refused, and reads no bucket.
+    #[test]
+    fn an_upper_case_bucket_under_the_domain_is_refused() {
+        assert_eq!(bucket(&["example.com"], "PHOTOS.example.com"), None);
+        assert_eq!(refusal(&["example.com"], "PHOTOS.example.com"), Some(HostRefusal::RefusedBucket));
+    }
+
+    /// Negative: an IPv6 literal. The bare form is no domain and is refused; with a port it is a
+    /// socket address and reads path-style.
+    #[test]
+    fn ipv6_literals_follow_the_address_rules() {
+        assert_eq!(refusal(&["example.com"], "[::1]"), Some(HostRefusal::UnusableHost));
+        assert_eq!(bucket(&["example.com"], "[::1]:9000"), None);
+        assert_eq!(refusal(&["example.com"], "[::1]:9000"), None);
+    }
+
+    /// Negative: a trailing root dot makes the host no domain, so it is refused, not stripped.
+    #[test]
+    fn a_trailing_dot_is_refused() {
+        assert_eq!(refusal(&["example.com"], "photos.example.com."), Some(HostRefusal::UnusableHost));
+        assert_eq!(bucket(&["example.com"], "photos.example.com."), None);
+    }
+
+    /// Negative: an empty label never reaches the resolver; the gateway's wire layer refuses the
+    /// host first (legacy answers 400 as well).
+    #[test]
+    fn an_empty_label_is_refused_before_resolution() {
+        for host in ["a..b.example.com", ".example.com"] {
+            let request = http::Request::builder()
+                .uri("/k")
+                .header("host", host)
+                .body(())
+                .expect("a valid request");
+            assert!(rustfs_gateway_http::effective_host(&request).is_err(), "{host}");
+        }
+    }
+
+    /// Negative: a punycode label is no bucket under a domain; outside every domain the host is
+    /// a valid domain that is no bucket name, so it reads path-style.
+    #[test]
+    fn a_punycode_label_is_never_a_bucket() {
+        assert_eq!(refusal(&["example.com"], "xn--bcher-kva.example.com"), Some(HostRefusal::RefusedBucket));
+        assert_eq!(bucket(&["example.com"], "xn--bcher-kva.de"), None);
+        assert_eq!(refusal(&["example.com"], "xn--bcher-kva.de"), None);
+    }
+
+    /// Negative: a label over 63 bytes is no bucket, under a domain or outside one.
+    #[test]
+    fn a_label_over_63_bytes_is_never_a_bucket() {
+        let long = "a".repeat(64);
+        let under = format!("{long}.example.com");
+        assert_eq!(refusal(&["example.com"], &under), Some(HostRefusal::RefusedBucket));
+        let outside = format!("{long}.org");
+        assert_eq!(bucket(&["example.com"], &outside), None);
+        assert_eq!(refusal(&["example.com"], &outside), None);
+        let at_limit = "a".repeat(63);
+        assert_eq!(bucket(&["example.com"], &format!("{at_limit}.example.com")), Some(at_limit));
+    }
+
+    /// Positive and negative: two configured domains where one merely ends in the other's letters
+    /// both work, and each domain itself is path-style.
+    #[test]
+    fn a_domain_and_its_suffix_lookalike_are_both_served() {
+        let domains = ["example.com", "badexample.com"];
+        assert_eq!(bucket(&domains, "photos.example.com").as_deref(), Some("photos"));
+        assert_eq!(bucket(&domains, "photos.badexample.com").as_deref(), Some("photos"));
+        assert_eq!(bucket(&domains, "badexample.com"), None);
+        assert_eq!(bucket(&domains, "bad.example.com").as_deref(), Some("bad"));
+    }
+
+    /// Negative: a configured domain is matched exactly, case included.
+    #[test]
+    fn a_configured_domain_is_matched_on_its_case() {
+        assert_eq!(bucket(&["Example.COM"], "photos.Example.COM").as_deref(), Some("photos"));
+        assert_eq!(bucket(&["Example.COM"], "photos.example.com").as_deref(), Some("photos.example.com"));
+        assert_eq!(bucket(&["Example.COM"], "Example.COM"), None);
+    }
+}

crates/gateway-adapter/src/lib.rs

@@ -0,0 +1,105 @@
+// 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.
+
+//! RustFS implementations of the `rustfs-gateway` extension traits.
+//!
+//! This is the gateway's ring 2: it may depend on RustFS business crates, and nothing in the
+//! gateway depends on it. Each module adapts one extension point to the RustFS subsystem that
+//! owns the answer.
+
+pub mod bucket_owner;
+pub mod credentials;
+pub mod host;
+pub mod policy;
+pub mod root;
+pub mod trace;
+
+use std::sync::Arc;
+
+use rustfs_gateway::{Authenticator, CredentialProvider, ServiceBuilder};
+
+pub use bucket_owner::{BucketLookup, OwnerLookupError, RustfsBucketOwners};
+pub use credentials::{IamCredentials, IamKeyLookup, STS_SESSION_ISSUER};
+pub use host::rustfs_host_resolver;
+pub use policy::{IamPolicySource, IamPrincipalLookup, PrincipalKind, RustfsPrincipal};
+pub use root::{MIN_SECRET_KEY_LEN, RootCredentials, RootCredentialsError, RootSource};
+pub use trace::{AttachOutcome, RequestIdCarrier, attach_host_request_id, host_request_id};
+
+/// The IAM-backed credential provider and policy source, assembled together.
+///
+/// Both halves read the same IAM handle and the same live [`RootSource`], so a key cannot be root
+/// to one and an ordinary user to the other. Neither half can be taken out on its own: the fields
+/// are private, and [`IamAdapters::install`] consumes the pair and hands both to one builder.
+///
+/// Mixing the halves of two assemblies does not compile:
+///
+/// ```compile_fail
+/// fn mix(a: rustfs_gateway_adapter::IamAdapters, b: rustfs_gateway_adapter::IamAdapters) {
+///     let _mixed = (a.credentials, b.policy);
+/// }
+/// ```
+///
+/// Nor does building a half outside the pair:
+///
+/// ```compile_fail
+/// use std::sync::Arc;
+/// use rustfs_gateway_adapter::{IamPolicySource, IamPrincipalLookup, RootSource};
+/// fn build(iam: Arc<dyn IamPrincipalLookup>, root: Arc<dyn RootSource>) {
+///     let _alone = IamPolicySource::new(iam, root);
+/// }
+/// ```
+///
+/// Installing the pair is the one supported path:
+///
+/// ```no_run
+/// use std::sync::Arc;
+/// use rustfs_gateway::{RegionSet, ServiceBuilder, SigV4Authenticator};
+/// use rustfs_gateway_adapter::{IamAdapters, IamPrincipalLookup, RootSource};
+/// fn assemble(iam: Arc<dyn IamPrincipalLookup>, root: Arc<dyn RootSource>, regions: RegionSet) -> ServiceBuilder {
+///     IamAdapters::new(iam, root).install(ServiceBuilder::new(), |provider| SigV4Authenticator::new(provider, regions))
+/// }
+/// ```
+pub struct IamAdapters {
+    credentials: IamCredentials,
+    policy: IamPolicySource,
+}
+
+impl IamAdapters {
+    /// Assembles both adapters over one IAM handle and one root source. The binary passes a
+    /// source that reads its live root credentials on every call (see [`RootSource`]).
+    #[must_use]
+    pub fn new(iam: Arc<dyn IamPrincipalLookup>, root: Arc<dyn RootSource>) -> Self {
+        Self {
+            credentials: IamCredentials::new(iam.clone(), root.clone()),
+            policy: IamPolicySource::new(iam, root),
+        }
+    }
+
+    /// Installs both halves into `builder`: the credential provider through the authenticator
+    /// `authenticator` builds over it (for example `SigV4Authenticator::new(provider, regions)`
+    /// with the RustFS profile applied), and the policy source.
+    ///
+    /// Consumes the pair and gives both halves to this one builder. The `authenticator` closure
+    /// receives the provider handle and may keep a clone of it; nothing here prevents that, so
+    /// the closure must not hand it to a second builder.
+    #[must_use]
+    pub fn install<A: Authenticator>(
+        self,
+        builder: ServiceBuilder,
+        authenticator: impl FnOnce(Arc<dyn CredentialProvider>) -> A,
+    ) -> ServiceBuilder {
+        let authenticator = authenticator(Arc::new(self.credentials));
+        builder.authenticator(authenticator).policy_source(self.policy)
+    }
+}

crates/gateway-adapter/src/policy.rs

@@ -0,0 +1,779 @@
+// 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.
+
+//! The gateway's [`PolicySource`], answered by RustFS IAM.
+//!
+//! Responsible for: the one reading of the caller's IAM facts a request is judged against — which
+//! principal kind it is, the user it acts for, its groups, its token claims, and whether it owns
+//! the deployment — packed as a [`RustfsPrincipal`] into the gateway's [`PolicySnapshot`].
+//! NOT responsible for: evaluating any policy, choosing which policy documents apply (that stays
+//! in `IamSys::is_allowed`, which owns the STS, OIDC, session-policy and service-account rules), or
+//! verifying the signature or the session token (the gateway did both before calling this).
+//! Upstream: the gateway's verified [`Identity`]. Downstream: the RustFS authorizer, which reads
+//! the payload back with `PolicySnapshot::get::<RustfsPrincipal>()`.
+//!
+//! # Failure classes
+//!
+//! Every refusal is a [`PolicyError`] with a constant label and nothing from IAM or the request:
+//! IAM not ready, not initialized, or failing, or root credentials not initialized, is
+//! [`PolicyError::unavailable`]; a principal IAM no longer knows is [`UNKNOWN_PRINCIPAL`]; a
+//! disabled one, or one whose secret fails the effective-length rule, is
+//! [`DISABLED_PRINCIPAL`]; an identity
+//! whose credential kind disagrees with IAM's record is [`SESSION_MISMATCH`]; a token whose claims
+//! cannot be read is [`PolicyError::malformed`]. None of them is ever an empty snapshot: an empty
+//! snapshot is what an anonymous caller gets, and answering a fault with it would judge an
+//! authenticated caller as anonymous.
+//!
+//! # Secret handling
+//!
+//! [`RustfsPrincipal`] holds no secret key and no session token, and implements no `Debug`, so a
+//! snapshot formatted into a log cannot print its claims either. The root credentials and the STS
+//! token-signing key are read per request from the shared [`RootSource`] and never stored here.
+
+use std::collections::HashMap;
+use std::sync::Arc;
+
+use rustfs_credentials::Credentials as IamCredential;
+use rustfs_gateway::{BoxFuture, Identity, PolicyError, PolicySnapshot, PolicySource};
+use rustfs_iam::error::Error as IamError;
+use rustfs_iam::store::Store;
+use rustfs_iam::sys::{
+    IamSys, SESSION_POLICY_NAME, get_claims_from_token_with_secret, get_claims_from_token_with_secret_allow_missing_exp,
+    is_rustfs_oidc_claims,
+};
+use rustfs_policy::auth::ACCOUNT_OFF;
+use serde_json::Value;
+
+use crate::credentials::IamKeyLookup;
+use crate::root::{RootCredentials, RootSource, is_usable_secret};
+
+/// [`PolicyError`] label: IAM does not know the authenticated principal (any longer).
+pub const UNKNOWN_PRINCIPAL: &str = "unknown-principal";
+/// [`PolicyError`] label: IAM holds the principal disabled, expired, or otherwise invalid.
+pub const DISABLED_PRINCIPAL: &str = "disabled-principal";
+/// [`PolicyError`] label: the identity's credential kind (session or long-term) disagrees with
+/// what IAM holds for its access key.
+pub const SESSION_MISMATCH: &str = "session-mismatch";
+
+/// The narrow slice of IAM a policy reading needs, on top of [`IamKeyLookup`].
+///
+/// Implemented for [`IamSys`]; exists so the source can be tested against a fake.
+pub trait IamPrincipalLookup: IamKeyLookup {
+    /// The built-in IAM groups `user` belongs to now, as `IamSys::user_group_memberships`
+    /// answers. These may include groups granted after a credential was issued.
+    fn group_memberships(&self, user: &str) -> Vec<String>;
+}
+
+impl<T: Store> IamPrincipalLookup for IamSys<T> {
+    fn group_memberships(&self, user: &str) -> Vec<String> {
+        IamSys::user_group_memberships(self, user)
+    }
+}
+
+/// Which kind of credential the principal authenticated with.
+#[derive(Clone, Copy, PartialEq, Eq)]
+pub enum PrincipalKind {
+    /// The deployment's root credentials.
+    Root,
+    /// A long-term IAM user.
+    User,
+    /// A service account, acting for its parent user.
+    ServiceAccount,
+    /// An STS temporary credential, acting for its parent user.
+    Temporary,
+}
+
+/// The caller's IAM facts: the payload of every [`PolicySnapshot`] [`IamPolicySource`] produces.
+///
+/// Carries no secret key and no session token, and deliberately implements no `Debug`.
+pub struct RustfsPrincipal {
+    kind: PrincipalKind,
+    account: String,
+    username: String,
+    parent_user: String,
+    credential_groups: Option<Vec<String>>,
+    group_memberships: Vec<String>,
+    claims: Option<HashMap<String, Value>>,
+    is_owner: bool,
+}
+
+impl RustfsPrincipal {
+    /// The kind of credential the caller authenticated with.
+    #[must_use]
+    pub fn kind(&self) -> PrincipalKind {
+        self.kind
+    }
+
+    /// The access key the caller signed with (IAM `Args::account`).
+    #[must_use]
+    pub fn account(&self) -> &str {
+        &self.account
+    }
+
+    /// The user the caller acts as: the parent user for an STS credential or a service account,
+    /// the access key otherwise.
+    #[must_use]
+    pub fn username(&self) -> &str {
+        &self.username
+    }
+
+    /// The parent user IAM records for the credential; empty for root and long-term users.
+    #[must_use]
+    pub fn parent_user(&self) -> &str {
+        &self.parent_user
+    }
+
+    /// The groups recorded on the credential itself (IAM `Args::groups`).
+    #[must_use]
+    pub fn credential_groups(&self) -> &Option<Vec<String>> {
+        &self.credential_groups
+    }
+
+    /// The groups IAM currently lists for [`Self::username`], read once with the snapshot.
+    #[must_use]
+    pub fn group_memberships(&self) -> &[String] {
+        &self.group_memberships
+    }
+
+    /// The claims the credential's token carries; `None` when it has none.
+    #[must_use]
+    pub fn claims(&self) -> Option<&HashMap<String, Value>> {
+        self.claims.as_ref()
+    }
+
+    /// Whether the caller owns the deployment (IAM `Args::is_owner`): root, or a credential of
+    /// root that no session policy or external OIDC identity narrows.
+    #[must_use]
+    pub fn is_owner(&self) -> bool {
+        self.is_owner
+    }
+}
+
+/// A [`PolicySource`] over RustFS IAM and the live root credentials.
+///
+/// Built only by [`crate::IamAdapters::new`], so it always reads the same [`RootSource`] as the
+/// credential provider it was assembled with.
+pub struct IamPolicySource {
+    iam: Arc<dyn IamPrincipalLookup>,
+    root: Arc<dyn RootSource>,
+}
+
+impl IamPolicySource {
+    pub(crate) fn new(iam: Arc<dyn IamPrincipalLookup>, root: Arc<dyn RootSource>) -> Self {
+        Self { iam, root }
+    }
+
+    async fn read(&self, identity: Option<&Identity>) -> Result<PolicySnapshot, PolicyError> {
+        let Some(identity) = identity else {
+            return Ok(PolicySnapshot::empty());
+        };
+        let principal = self.principal(identity).await?;
+        Ok(PolicySnapshot::of(Arc::new(principal)))
+    }
+
+    async fn principal(&self, identity: &Identity) -> Result<RustfsPrincipal, PolicyError> {
+        let access_key = identity.access_key_id();
+        let has_session = identity.session().is_some();
+        // Without root credentials neither ownership nor STS claims can be decided.
+        let Some(root) = self.root.current() else {
+            return Err(PolicyError::unavailable());
+        };
+        let root_access_key = root.access_key();
+
+        if access_key == root_access_key {
+            // Root has no session token, so a session identity for its key is not one IAM issued.
+            if has_session {
+                return Err(PolicyError::because(SESSION_MISMATCH));
+            }
+            return Ok(RustfsPrincipal {
+                kind: PrincipalKind::Root,
+                account: access_key.to_owned(),
+                username: access_key.to_owned(),
+                parent_user: String::new(),
+                credential_groups: None,
+                group_memberships: Vec::new(),
+                claims: None,
+                is_owner: true,
+            });
+        }
+
+        if !self.iam.is_ready() {
+            return Err(PolicyError::unavailable());
+        }
+        let cred = match self.iam.check_key(access_key).await {
+            Ok((Some(identity), true)) => identity.credentials,
+            Ok((Some(_), false)) => return Err(PolicyError::because(DISABLED_PRINCIPAL)),
+            Ok((None, _)) => return Err(PolicyError::because(UNKNOWN_PRINCIPAL)),
+            Err(error) => return Err(classify_iam_error(&error)),
+        };
+        if cred.status == ACCOUNT_OFF || !is_usable_secret(cred.secret_key.as_bytes()) {
+            return Err(PolicyError::because(DISABLED_PRINCIPAL));
+        }
+
+        let kind = if cred.is_service_account() {
+            PrincipalKind::ServiceAccount
+        } else if !cred.session_token.is_empty() {
+            PrincipalKind::Temporary
+        } else {
+            PrincipalKind::User
+        };
+        // The gateway admitted this identity against the credential IAM returned, so the kinds
+        // agree unless IAM's record changed in between; refuse rather than judge a stale reading.
+        if has_session != (kind == PrincipalKind::Temporary) {
+            return Err(PolicyError::because(SESSION_MISMATCH));
+        }
+
+        let claims = read_claims(kind, &cred, &root)?;
+        let username = match kind {
+            PrincipalKind::Temporary | PrincipalKind::ServiceAccount => cred.parent_user.clone(),
+            PrincipalKind::Root | PrincipalKind::User => cred.access_key.clone(),
+        };
+        let membership_subject = if username.is_empty() { access_key } else { username.as_str() };
+        let group_memberships = self.iam.group_memberships(membership_subject);
+        let is_owner = acts_for_root(&cred, claims.as_ref(), root_access_key);
+
+        Ok(RustfsPrincipal {
+            kind,
+            account: access_key.to_owned(),
+            username,
+            parent_user: cred.parent_user,
+            credential_groups: cred.groups,
+            group_memberships,
+            claims,
+            is_owner,
+        })
+    }
+}
+
+/// The claims of the credential's own token. The gateway admitted the presented token only when it
+/// equals the stored one, so reading the stored token reads the presented one.
+fn read_claims(
+    kind: PrincipalKind,
+    cred: &IamCredential,
+    root: &RootCredentials,
+) -> Result<Option<HashMap<String, Value>>, PolicyError> {
+    // As in legacy `check_claims_from_token_with_context`: no token, no claims.
+    if cred.session_token.is_empty() {
+        return Ok(None);
+    }
+    let claims = match kind {
+        PrincipalKind::ServiceAccount => {
+            get_claims_from_token_with_secret_allow_missing_exp(&cred.session_token, &cred.secret_key)
+                .map_err(|_| PolicyError::malformed())?
+        }
+        PrincipalKind::Temporary => get_claims_from_token_with_secret(&cred.session_token, root.sts_signing_key())
+            .map_err(|_| PolicyError::malformed())?,
+        PrincipalKind::Root | PrincipalKind::User => return Ok(None),
+    };
+    Ok(if claims.is_empty() { None } else { Some(claims) })
+}
+
+/// Classifies an IAM failure without keeping any of its text.
+fn classify_iam_error(error: &IamError) -> PolicyError {
+    match error {
+        IamError::NoSuchUser(_)
+        | IamError::NoSuchServiceAccount(_)
+        | IamError::NoSuchTempAccount(_)
+        | IamError::NoSuchAccount(_) => PolicyError::because(UNKNOWN_PRINCIPAL),
+        _ => PolicyError::unavailable(),
+    }
+}
+
+/// Root ownership as the legacy edge decides it (`rustfs/src/auth.rs` `has_root_access`): the
+/// credential is root's or acts for root, and neither a session policy nor an external OIDC
+/// identity narrows it.
+fn acts_for_root(cred: &IamCredential, claims: Option<&HashMap<String, Value>>, root_access_key: &str) -> bool {
+    (cred.access_key == root_access_key || cred.parent_user == root_access_key)
+        && !claims.is_some_and(|claims| claims.contains_key(SESSION_POLICY_NAME) || is_rustfs_oidc_claims(claims))
+}
+
+impl PolicySource for IamPolicySource {
+    fn snapshot<'a>(&'a self, identity: Option<&'a Identity>) -> BoxFuture<'a, Result<PolicySnapshot, PolicyError>> {
+        Box::pin(self.read(identity))
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+    use crate::root::RootCell;
+    use rustfs_credentials::{Credentials as IamCredential, IAM_POLICY_CLAIM_NAME_SA};
+    use rustfs_gateway::SessionBinding;
+    use rustfs_iam::error::Error as IamError;
+    use rustfs_iam::sys::SESSION_POLICY_NAME;
+    use rustfs_iam::utils::generate_jwt;
+    use rustfs_policy::auth::UserIdentity;
+    use std::sync::atomic::{AtomicUsize, Ordering};
+    use time::OffsetDateTime;
+
+    const ROOT_ACCESS_KEY: &str = "rootadmin";
+    const ROOT_SECRET: &str = "root-secret-key-0001";
+    /// Deliberately not the root secret, so every STS test proves which key claims are read with.
+    const SIGNING_KEY: &str = "sts-signing-key-0001";
+    const ACCESS_KEY: &str = "AKIDUSER0001";
+    const SECRET: &str = "user-secret-key-0001";
+    const PARENT: &str = "alice";
+    const EXPIRES_AT: i64 = 4_000_000_000;
+
+    type KeyAnswer = Result<(Option<UserIdentity>, bool), IamError>;
+    type Answer = Box<dyn Fn() -> KeyAnswer + Send + Sync>;
+
+    struct FakeIam {
+        ready: bool,
+        answer: Answer,
+        calls: AtomicUsize,
+    }
+
+    impl IamKeyLookup for FakeIam {
+        fn is_ready(&self) -> bool {
+            self.ready
+        }
+
+        fn check_key<'a>(&'a self, _access_key: &'a str) -> BoxFuture<'a, KeyAnswer> {
+            self.calls.fetch_add(1, Ordering::SeqCst);
+            let answer = (self.answer)();
+            Box::pin(async move { answer })
+        }
+    }
+
+    impl IamPrincipalLookup for FakeIam {
+        /// Distinct groups per name, so a test can tell which name the source asked about.
+        fn group_memberships(&self, user: &str) -> Vec<String> {
+            match user {
+                PARENT => vec!["parent-group".to_owned()],
+                ACCESS_KEY => vec!["key-group".to_owned()],
+                ROOT_ACCESS_KEY => vec!["root-group".to_owned()],
+                _ => Vec::new(),
+            }
+        }
+    }
+
+    fn fake(ready: bool, answer: impl Fn() -> KeyAnswer + Send + Sync + 'static) -> Arc<FakeIam> {
+        Arc::new(FakeIam {
+            ready,
+            answer: Box::new(answer),
+            calls: AtomicUsize::new(0),
+        })
+    }
+
+    fn root() -> RootCredentials {
+        RootCredentials::new(ROOT_ACCESS_KEY, ROOT_SECRET, SIGNING_KEY).expect("valid root credentials")
+    }
+
+    fn source(iam: Arc<FakeIam>) -> IamPolicySource {
+        IamPolicySource::new(iam, Arc::new(RootCell::new(Some(root()))))
+    }
+
+    fn user() -> IamCredential {
+        IamCredential {
+            access_key: ACCESS_KEY.to_owned(),
+            secret_key: SECRET.to_owned(),
+            status: "on".to_owned(),
+            groups: Some(vec!["cred-group".to_owned()]),
+            ..Default::default()
+        }
+    }
+
+    fn sts_token(claims: &[(&str, Value)], secret: &str) -> String {
+        let mut map: HashMap<String, Value> = claims.iter().map(|(k, v)| ((*k).to_owned(), v.clone())).collect();
+        map.insert("exp".to_owned(), Value::from(EXPIRES_AT));
+        generate_jwt(&map, secret).expect("a token")
+    }
+
+    fn sts(parent: &str, token: String) -> IamCredential {
+        IamCredential {
+            session_token: token,
+            parent_user: parent.to_owned(),
+            expiration: Some(OffsetDateTime::from_unix_timestamp(EXPIRES_AT).expect("valid timestamp")),
+            ..user()
+        }
+    }
+
+    fn found(cred: IamCredential) -> KeyAnswer {
+        Ok((Some(UserIdentity::new(cred)), true))
+    }
+
+    fn long_term(access_key: &str) -> Identity {
+        Identity::new(access_key).expect("valid access key")
+    }
+
+    fn session(access_key: &str) -> Identity {
+        let binding = SessionBinding::new(crate::STS_SESSION_ISSUER, EXPIRES_AT).expect("valid binding");
+        long_term(access_key).with_session_binding(binding)
+    }
+
+    async fn principal_of(source: &IamPolicySource, identity: &Identity) -> PolicySnapshot {
+        source.snapshot(Some(identity)).await.expect("a snapshot")
+    }
+
+    fn payload(snapshot: &PolicySnapshot) -> &RustfsPrincipal {
+        snapshot.get::<RustfsPrincipal>().expect("the payload is a RustfsPrincipal")
+    }
+
+    fn refusal(result: Result<PolicySnapshot, PolicyError>) -> PolicyError {
+        match result {
+            Ok(snapshot) => panic!("expected a refusal, got a snapshot (empty: {})", snapshot.is_empty()),
+            Err(error) => error,
+        }
+    }
+
+    /// Compile-time: `RustfsPrincipal` has no `Debug`, so no formatting of a snapshot payload can
+    /// reach its claims. The two blanket impls make the path ambiguous, and so a compile error, as
+    /// soon as `RustfsPrincipal: Debug` holds.
+    #[test]
+    fn policy_principal_has_no_debug() {
+        trait AmbiguousIfDebug<A> {
+            fn some_item() {}
+        }
+        impl<T: ?Sized> AmbiguousIfDebug<()> for T {}
+        impl<T: ?Sized + core::fmt::Debug> AmbiguousIfDebug<u8> for T {}
+        let _ = <RustfsPrincipal as AmbiguousIfDebug<_>>::some_item;
+    }
+
+    /// Positive — a long-term user's facts: itself as username, both group sources, no claims,
+    /// not the owner.
+    #[tokio::test]
+    async fn policy_user_snapshot_carries_iam_facts() {
+        let iam = fake(true, || found(user()));
+        let snapshot = principal_of(&source(iam), &long_term(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::User);
+        assert_eq!(principal.account(), ACCESS_KEY);
+        assert_eq!(principal.username(), ACCESS_KEY);
+        assert_eq!(principal.parent_user(), "");
+        assert_eq!(principal.credential_groups(), &Some(vec!["cred-group".to_owned()]));
+        assert_eq!(principal.group_memberships(), ["key-group".to_owned()]);
+        assert!(principal.claims().is_none());
+        assert!(!principal.is_owner());
+    }
+
+    /// Positive — an STS credential acts for its parent user, and its claims are the ones its
+    /// token carries, read with the root secret that signed it.
+    #[tokio::test]
+    async fn policy_sts_snapshot_carries_parent_user_and_claims() {
+        let token = sts_token(&[("sub", Value::from("alice-session"))], SIGNING_KEY);
+        let iam = fake(true, move || found(sts(PARENT, token.clone())));
+        let snapshot = principal_of(&source(iam), &session(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::Temporary);
+        assert_eq!(principal.account(), ACCESS_KEY);
+        assert_eq!(principal.parent_user(), PARENT);
+        assert_eq!(principal.username(), PARENT);
+        let claims = principal.claims().expect("an STS principal has claims");
+        assert_eq!(claims.get("sub"), Some(&Value::from("alice-session")));
+        assert_eq!(principal.group_memberships(), ["parent-group".to_owned()]);
+        assert!(!principal.is_owner());
+    }
+
+    /// Positive — a service account's claims are read with its own secret, and it acts for its
+    /// parent.
+    #[tokio::test]
+    async fn policy_service_account_snapshot_reads_its_own_token() {
+        let mut token_claims: HashMap<String, Value> = HashMap::new();
+        token_claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"));
+        token_claims.insert("parent".to_owned(), Value::from(PARENT));
+        let token = generate_jwt(&token_claims, SECRET).expect("a token");
+        let iam = fake(true, move || {
+            let mut cred = user();
+            cred.session_token = token.clone();
+            cred.parent_user = PARENT.to_owned();
+            cred.claims = Some([(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"))].into());
+            found(cred)
+        });
+        let snapshot = principal_of(&source(iam), &long_term(ACCESS_KEY)).await;
+        let principal = payload(&snapshot);
+        assert!(principal.kind() == PrincipalKind::ServiceAccount);
+        assert_eq!(principal.username(), PARENT);
+        assert_eq!(principal.claims().and_then(|c| c.get("parent")), Some(&Value::from(PARENT)));
+        assert_eq!(principal.group_memberships(), ["parent-group".to_owned()]);
+        assert!(!principal.is_owner());
+    }
+
+    /// Positive and negative — root is the owner without asking IAM, even before IAM is ready;
+    /// an STS credential of root is the owner too, but not once a session policy narrows it.
+    #[tokio::test]
+    async fn policy_owner_is_root_or_unrestricted_root_session() {
+        let not_ready = fake(false, || found(user()));
+        let snapshot = principal_of(&source(not_ready.clone()), &long_term(ROOT_ACCESS_KEY)).await;
+        let root = payload(&snapshot);
+        assert!(root.kind() == PrincipalKind::Root);
+        assert!(root.is_owner());
+        assert_eq!(not_ready.calls.load(Ordering::SeqCst), 0);
+
+        let plain = sts_token(&[], SIGNING_KEY);
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, plain.clone())));
+        assert!(payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+
+        let narrowed = sts_token(&[(SESSION_POLICY_NAME, Value::from("e30"))], SIGNING_KEY);
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, narrowed.clone())));
+        assert!(!payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+    }
+
+    /// Negative — an anonymous caller is an empty snapshot, and IAM is not asked.
+    #[tokio::test]
+    async fn policy_anonymous_is_empty_snapshot() {
+        let iam = fake(true, || found(user()));
+        let snapshot = source(iam.clone()).snapshot(None).await.expect("an anonymous snapshot");
+        assert!(snapshot.is_empty());
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    /// Negative — IAM that is not ready, not initialized, or failing is `unavailable`: never an
+    /// empty snapshot, which an authorizer would judge as an anonymous caller.
+    #[tokio::test]
+    async fn policy_iam_unavailable_is_unavailable() {
+        let not_ready = fake(false, || found(user()));
+        let error = refusal(source(not_ready.clone()).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::unavailable());
+        assert_eq!(not_ready.calls.load(Ordering::SeqCst), 0);
+
+        let failures: [fn() -> IamError; 2] = [
+            || IamError::IamSysNotInitialized,
+            || IamError::StringError("storage read failed".to_owned()),
+        ];
+        for failure in failures {
+            let iam = fake(true, move || Err(failure()));
+            let error = refusal(source(iam).snapshot(Some(&long_term(ACCESS_KEY))).await);
+            assert_eq!(error, PolicyError::unavailable(), "{:?}", failure());
+        }
+    }
+
+    /// Negative — a principal IAM no longer knows is refused as unknown, not as unavailable and
+    /// not as an empty snapshot.
+    #[tokio::test]
+    async fn policy_unknown_principal_is_refused() {
+        let answers: [fn() -> KeyAnswer; 2] = [|| Ok((None, false)), || Err(IamError::NoSuchUser(ACCESS_KEY.to_owned()))];
+        for answer in answers {
+            let error = refusal(source(fake(true, answer)).snapshot(Some(&long_term(ACCESS_KEY))).await);
+            assert_eq!(error, PolicyError::because(UNKNOWN_PRINCIPAL));
+        }
+    }
+
+    /// Negative — a disabled or invalid principal is refused rather than given a snapshot.
+    #[tokio::test]
+    async fn policy_disabled_principal_is_refused() {
+        let invalid = fake(true, || Ok((Some(UserIdentity::new(user())), false)));
+        let error = refusal(source(invalid).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(DISABLED_PRINCIPAL));
+
+        let off = fake(true, || {
+            let mut cred = user();
+            cred.status = "off".to_owned();
+            found(cred)
+        });
+        let error = refusal(source(off).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(DISABLED_PRINCIPAL));
+    }
+
+    /// Negative — an STS token not signed by root yields no claims at all: it is `malformed`.
+    #[tokio::test]
+    async fn policy_sts_token_with_foreign_signature_is_malformed() {
+        let forged = sts_token(&[("sub", Value::from("alice-session"))], "not-the-root-secret");
+        let iam = fake(true, move || found(sts(PARENT, forged.clone())));
+        let error = refusal(source(iam).snapshot(Some(&session(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::malformed());
+    }
+
+    /// Negative — the identity's credential kind must match what IAM holds: a session identity for
+    /// a long-term key, or a long-term identity for an STS key, is refused.
+    #[tokio::test]
+    async fn policy_session_kind_mismatch_is_refused() {
+        let token = sts_token(&[], SIGNING_KEY);
+        let iam = fake(true, move || found(sts(PARENT, token.clone())));
+        let error = refusal(source(iam).snapshot(Some(&long_term(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+
+        let iam = fake(true, || found(user()));
+        let error = refusal(source(iam).snapshot(Some(&session(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+
+        let iam = fake(false, || found(user()));
+        let error = refusal(source(iam).snapshot(Some(&session(ROOT_ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::because(SESSION_MISMATCH));
+    }
+
+    fn service_account(parent: &str, extra_claims: &[(&str, Value)]) -> KeyAnswer {
+        let mut token_claims: HashMap<String, Value> = extra_claims.iter().map(|(k, v)| ((*k).to_owned(), v.clone())).collect();
+        token_claims.insert(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"));
+        let mut cred = user();
+        cred.session_token = generate_jwt(&token_claims, SECRET).expect("a token");
+        cred.parent_user = parent.to_owned();
+        cred.claims = Some([(IAM_POLICY_CLAIM_NAME_SA.to_owned(), Value::from("inherited-policy"))].into());
+        found(cred)
+    }
+
+    /// Negative — a valid-flagged identity whose secret is empty or short gets no snapshot.
+    #[tokio::test]
+    async fn policy_short_secret_principal_is_refused() {
+        for weak in ["", "1234567", "\0\0\0\0\0\0\0\0", "abc\0\0\0\0\0"] {
+            let iam = fake(true, move || {
+                let mut cred = user();
+                cred.secret_key = weak.to_owned();
+                found(cred)
+            });
+            let error = refusal(source(iam).snapshot(Some(&long_term(ACCESS_KEY))).await);
+            assert_eq!(error, PolicyError::because(DISABLED_PRINCIPAL), "{weak:?}");
+        }
+        for usable in ["12345678", "abc\0defg"] {
+            let iam = fake(true, move || {
+                let mut cred = user();
+                cred.secret_key = usable.to_owned();
+                found(cred)
+            });
+            assert!(payload(&principal_of(&source(iam), &long_term(ACCESS_KEY)).await).kind() == PrincipalKind::User);
+        }
+    }
+
+    /// Negative — a root session issued through external OIDC is not the owner, though its parent
+    /// is root; the same session without the OIDC issuer is.
+    #[tokio::test]
+    async fn policy_oidc_root_session_is_not_owner() {
+        let oidc = sts_token(
+            &[
+                ("iss", Value::from("rustfs-oidc")),
+                ("oidc_provider", Value::from("corp-idp")),
+                ("sub", Value::from("alice@corp")),
+            ],
+            SIGNING_KEY,
+        );
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, oidc.clone())));
+        assert!(!payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+
+        let local = sts_token(
+            &[
+                ("iss", Value::from("rustfs-sts")),
+                ("oidc_provider", Value::from("corp-idp")),
+                ("sub", Value::from("alice@corp")),
+            ],
+            SIGNING_KEY,
+        );
+        let iam = fake(true, move || found(sts(ROOT_ACCESS_KEY, local.clone())));
+        assert!(payload(&principal_of(&source(iam), &session(ACCESS_KEY)).await).is_owner());
+    }
+
+    /// Positive and negative — a service account of root is the owner; one narrowed by a session
+    /// policy is not, and neither is a service account of an ordinary user.
+    #[tokio::test]
+    async fn policy_service_account_of_root_is_owner() {
+        let iam = fake(true, || service_account(ROOT_ACCESS_KEY, &[]));
+        let principal = principal_of(&source(iam), &long_term(ACCESS_KEY)).await;
+        assert!(payload(&principal).kind() == PrincipalKind::ServiceAccount);
+        assert!(payload(&principal).is_owner());
+
+        let iam = fake(true, || service_account(ROOT_ACCESS_KEY, &[(SESSION_POLICY_NAME, Value::from("e30"))]));
+        assert!(!payload(&principal_of(&source(iam), &long_term(ACCESS_KEY)).await).is_owner());
+
+        let iam = fake(true, || service_account(PARENT, &[]));
+        assert!(!payload(&principal_of(&source(iam), &long_term(ACCESS_KEY)).await).is_owner());
+    }
+
+    /// Negative — STS claims are read with the STS signing key, not with the root secret: a token
+    /// signed with the root secret while the signing key differs is `malformed`.
+    #[tokio::test]
+    async fn policy_sts_claims_use_the_signing_key() {
+        let root_signed = sts_token(&[("sub", Value::from("alice-session"))], ROOT_SECRET);
+        let iam = fake(true, move || found(sts(PARENT, root_signed.clone())));
+        let error = refusal(source(iam).snapshot(Some(&session(ACCESS_KEY))).await);
+        assert_eq!(error, PolicyError::malformed());
+    }
+
+    /// Negative — without root credentials no principal can be judged: `unavailable`, never an
+    /// empty snapshot, and IAM is not asked.
+    #[tokio::test]
+    async fn policy_without_root_is_unavailable() {
+        let iam = fake(true, || found(user()));
+        let source = IamPolicySource::new(iam.clone(), Arc::new(RootCell::new(None)));
+        for identity in [long_term(ACCESS_KEY), long_term(ROOT_ACCESS_KEY)] {
+            assert_eq!(refusal(source.snapshot(Some(&identity)).await), PolicyError::unavailable());
+        }
+        assert_eq!(iam.calls.load(Ordering::SeqCst), 0);
+    }
+
+    type LiveRoot = Arc<std::sync::Mutex<Option<(String, String)>>>;
+
+    /// The primary path: a source that reads a live accessor on every call, as the binary's does,
+    /// with nothing cached in between.
+    fn live_root_source(live: &LiveRoot) -> Arc<dyn RootSource> {
+        let accessor = live.clone();
+        Arc::new(move || {
+            let guard = accessor.lock().ok()?;
+            let (access_key, secret_key) = guard.as_ref()?;
+            RootCredentials::new(access_key, secret_key, SIGNING_KEY).ok().map(Arc::new)
+        })
+    }
+
+    fn set_live(live: &LiveRoot, value: Option<(&str, &str)>) {
+        *live.lock().expect("live root lock") = value.map(|(ak, sk)| (ak.to_owned(), sk.to_owned()));
+    }
+
+    /// Positive and negative — the two adapters read one live root accessor: a root change is seen
+    /// by both on the next request with no second step, the old root key stops being root to
+    /// both, and an uninitialized or unusable root makes both unavailable.
+    #[tokio::test]
+    async fn adapters_share_one_live_root() {
+        use rustfs_gateway::{CredentialLookup, CredentialProvider, ProviderError};
+
+        let live: LiveRoot = Arc::default();
+        set_live(&live, Some((ROOT_ACCESS_KEY, ROOT_SECRET)));
+        let iam = fake(true, || Ok((None, false)));
+        let adapters = crate::IamAdapters::new(iam, live_root_source(&live));
+
+        let found = adapters.credentials.lookup(ROOT_ACCESS_KEY).await;
+        assert!(matches!(found, Ok(CredentialLookup::Found(_))));
+        let snapshot = principal_of(&adapters.policy, &long_term(ROOT_ACCESS_KEY)).await;
+        assert!(payload(&snapshot).is_owner());
+
+        set_live(&live, Some(("newroot", "new-root-secret-01")));
+        match adapters.credentials.lookup("newroot").await {
+            Ok(CredentialLookup::Found(credentials)) => assert_eq!(credentials.secret().expose(), b"new-root-secret-01"),
+            _ => panic!("the new root key must resolve to the new secret"),
+        }
+        assert!(matches!(
+            adapters.credentials.lookup(ROOT_ACCESS_KEY).await,
+            Ok(CredentialLookup::NotFound)
+        ));
+        assert!(payload(&principal_of(&adapters.policy, &long_term("newroot")).await).is_owner());
+        let old = refusal(adapters.policy.snapshot(Some(&long_term(ROOT_ACCESS_KEY))).await);
+        assert_eq!(old, PolicyError::because(UNKNOWN_PRINCIPAL));
+
+        for unusable in [None, Some(("newroot", "\0\0\0\0\0\0\0\0"))] {
+            set_live(&live, unusable);
+            assert!(matches!(adapters.credentials.lookup("newroot").await, Err(ProviderError::Unavailable)));
+            assert_eq!(
+                refusal(adapters.policy.snapshot(Some(&long_term("newroot"))).await),
+                PolicyError::unavailable()
+            );
+        }
+    }
+
+    /// Positive — `install` builds the authenticator exactly once, over a provider that reads the
+    /// pair's root source.
+    #[tokio::test]
+    async fn adapters_install_hands_over_the_paired_provider_once() {
+        use rustfs_gateway::{CredentialLookup, CredentialProvider, RegionSet, ServiceBuilder, SigV4Authenticator};
+
+        let live: LiveRoot = Arc::default();
+        set_live(&live, Some((ROOT_ACCESS_KEY, ROOT_SECRET)));
+        let adapters = crate::IamAdapters::new(fake(true, || Ok((None, false))), live_root_source(&live));
+        let mut handed = Vec::new();
+        let _builder = adapters.install(ServiceBuilder::new(), |provider| {
+            handed.push(provider.clone());
+            SigV4Authenticator::new(provider, RegionSet::new(["us-east-1"]).expect("a region"))
+        });
+        assert_eq!(handed.len(), 1);
+        let found = handed[0].lookup(ROOT_ACCESS_KEY).await;
+        assert!(matches!(found, Ok(CredentialLookup::Found(_))));
+    }
+}

crates/gateway-adapter/src/root.rs

@@ -0,0 +1,216 @@
+// 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.
+
+//! The deployment's root credentials, as every adapter in this crate reads them.
+//!
+//! Responsible for: [`RootCredentials`], the validated root key pair plus the key STS session
+//! tokens are signed with; [`RootSource`], the one live handle both the credential provider and
+//! the policy source read on every request; and the effective-secret-length rule every secret in
+//! this crate is held to.
+//! NOT responsible for: where the root keys come from or when they change. The binary implements
+//! [`RootSource`] directly over its live root accessor — a closure is enough — so there is no
+//! second copy of root to keep in step.
+//! Upstream: the binary's runtime credentials. Downstream: `crate::credentials`, `crate::policy`.
+//!
+//! # Why one source
+//!
+//! Legacy reads the current root credentials on every request. Two copies taken at assembly could
+//! disagree, or go stale after a root change, and a key one adapter still treats as root would be
+//! answered with the root secret or judged as the owner. Both adapters therefore hold the same
+//! `Arc<dyn RootSource>`, handed to them together by [`crate::IamAdapters::new`].
+
+use std::sync::Arc;
+
+use rustfs_gateway::Credentials;
+
+/// The shortest effective secret key, in bytes, that any credential in this crate may verify a
+/// signature with. The same bound IAM applies to its own credentials; a signature over a shorter
+/// (or empty) key is one anybody can compute.
+pub const MIN_SECRET_KEY_LEN: usize = 8;
+
+/// The number of bytes of `secret` a signature actually depends on: its length after trailing NUL
+/// bytes are removed.
+///
+/// HMAC pads a key shorter than its block with zero bytes, and SigV4's first key is `"AWS4"`
+/// followed by the secret, so trailing NULs add nothing: eight NUL bytes sign exactly like the
+/// empty secret, and `"abc"` followed by five NULs signs like `"abc"`. A secret of NULs only has
+/// an effective length of zero. An interior NUL is followed by real key bytes, which change the
+/// padded key, so it counts like any other byte.
+pub(crate) fn effective_secret_len(secret: &[u8]) -> usize {
+    secret.iter().rposition(|byte| *byte != 0).map_or(0, |last| last + 1)
+}
+
+/// Whether `secret` may verify a signature: its [`effective_secret_len`] is at least
+/// [`MIN_SECRET_KEY_LEN`]. The one rule for root, IAM credentials and the STS signing key.
+pub(crate) fn is_usable_secret(secret: &[u8]) -> bool {
+    effective_secret_len(secret) >= MIN_SECRET_KEY_LEN
+}
+
+/// Why [`RootCredentials::new`] refused its input. Carries no input text.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum RootCredentialsError {
+    /// The access key id is not one the gateway can hold.
+    InvalidAccessKey,
+    /// The root secret key fails the effective-length rule.
+    WeakSecretKey,
+    /// The STS token-signing key fails the effective-length rule.
+    WeakSigningKey,
+}
+
+impl core::fmt::Display for RootCredentialsError {
+    fn fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result {
+        f.write_str(match self {
+            Self::InvalidAccessKey => "root access key is not usable",
+            Self::WeakSecretKey => "root secret key is shorter than the minimum length",
+            Self::WeakSigningKey => "STS token-signing key is shorter than the minimum length",
+        })
+    }
+}
+
+impl std::error::Error for RootCredentialsError {}
+
+/// The root key pair and the STS token-signing key, validated together.
+///
+/// Implements no `Debug`; the secrets are readable only inside this crate.
+pub struct RootCredentials {
+    credentials: Credentials,
+    sts_signing_key: Box<str>,
+}
+
+impl RootCredentials {
+    /// Validates the root credentials and the key IAM signs STS session tokens with (today the
+    /// root secret, `rustfs_iam::manager::get_token_signing_key`; pass it separately so a future
+    /// key separation changes no adapter).
+    ///
+    /// # Errors
+    ///
+    /// [`RootCredentialsError`] when either secret fails the effective-length rule (fewer than
+    /// [`MIN_SECRET_KEY_LEN`] bytes once trailing NUL bytes are removed) or the access key id is not
+    /// one the gateway can hold.
+    pub fn new(access_key: &str, secret_key: &str, sts_signing_key: &str) -> Result<Self, RootCredentialsError> {
+        if !is_usable_secret(secret_key.as_bytes()) {
+            return Err(RootCredentialsError::WeakSecretKey);
+        }
+        if !is_usable_secret(sts_signing_key.as_bytes()) {
+            return Err(RootCredentialsError::WeakSigningKey);
+        }
+        let credentials =
+            Credentials::new(access_key, secret_key.as_bytes()).map_err(|_| RootCredentialsError::InvalidAccessKey)?;
+        Ok(Self {
+            credentials,
+            sts_signing_key: sts_signing_key.into(),
+        })
+    }
+
+    /// The root access key id, a public identifier.
+    #[must_use]
+    pub fn access_key(&self) -> &str {
+        self.credentials.identity().access_key_id()
+    }
+
+    pub(crate) fn credentials(&self) -> &Credentials {
+        &self.credentials
+    }
+
+    pub(crate) fn sts_signing_key(&self) -> &str {
+        &self.sts_signing_key
+    }
+}
+
+/// The live root credentials, read on every request by both adapters.
+///
+/// The binary implements it directly over RustFS's live root accessor, so a root change needs no
+/// second step. Any `Fn() -> Option<Arc<RootCredentials>>` is a source; for example, over the
+/// server context:
+///
+/// ```ignore
+/// let root = move || {
+///     let cred = ctx.action_credentials().get()?;
+///     let signing_key = ctx.token_signing_key()?;
+///     RootCredentials::new(&cred.access_key, &cred.secret_key, &signing_key).ok().map(Arc::new)
+/// };
+/// let adapters = IamAdapters::new(iam, Arc::new(root));
+/// ```
+///
+/// Root credentials that fail validation read as `None`, which both adapters answer as
+/// unavailable.
+pub trait RootSource: Send + Sync + 'static {
+    /// The current root credentials; `None` while they are not initialized, which both adapters
+    /// answer as unavailable rather than guessing.
+    fn current(&self) -> Option<Arc<RootCredentials>>;
+}
+
+impl<F> RootSource for F
+where
+    F: Fn() -> Option<Arc<RootCredentials>> + Send + Sync + 'static,
+{
+    fn current(&self) -> Option<Arc<RootCredentials>> {
+        self()
+    }
+}
+
+/// Test helper: a [`RootSource`] holding fixed root credentials. Production code reads the live
+/// accessor instead, so that there is no copy of root to keep in step.
+#[cfg(test)]
+pub(crate) struct RootCell(Option<Arc<RootCredentials>>);
+
+#[cfg(test)]
+impl RootCell {
+    pub(crate) fn new(root: Option<RootCredentials>) -> Self {
+        Self(root.map(Arc::new))
+    }
+}
+
+#[cfg(test)]
+impl RootSource for RootCell {
+    fn current(&self) -> Option<Arc<RootCredentials>> {
+        self.0.clone()
+    }
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    /// Negative — an empty or short root secret is refused at construction, so no source can ever
+    /// hand out a root key anybody could sign with; eight bytes is accepted.
+    #[test]
+    fn root_short_secret_is_refused() {
+        for weak in ["", "1234567", "\0\0\0\0\0\0\0\0", "abc\0\0\0\0\0", "abcdefg\0"] {
+            let refused = RootCredentials::new("rootadmin", weak, "signing-key-0001");
+            assert!(matches!(refused, Err(RootCredentialsError::WeakSecretKey)), "{} bytes", weak.len());
+        }
+        assert!(RootCredentials::new("rootadmin", "12345678", "signing-key-0001").is_ok());
+    }
+
+    /// Positive — an interior NUL is followed by real key bytes and counts: `"abc\0defg"` has an
+    /// effective length of eight and is accepted, as is the same secret with trailing NULs.
+    #[test]
+    fn root_interior_nul_counts_toward_length() {
+        for secret in ["abc\0defg", "abc\0defg\0\0"] {
+            assert!(RootCredentials::new("rootadmin", secret, "signing-key-0001").is_ok(), "{secret:?}");
+            assert!(RootCredentials::new("rootadmin", "root-secret-key-0001", secret).is_ok(), "{secret:?}");
+        }
+    }
+
+    /// Negative — an empty or short STS signing key is refused at construction.
+    #[test]
+    fn root_short_signing_key_is_refused() {
+        for weak in ["", "1234567", "\0\0\0\0\0\0\0\0", "abc\0\0\0\0\0"] {
+            let refused = RootCredentials::new("rootadmin", "root-secret-key-0001", weak);
+            assert!(matches!(refused, Err(RootCredentialsError::WeakSigningKey)), "{} bytes", weak.len());
+        }
+    }
+}

crates/gateway-adapter/src/trace.rs

@@ -0,0 +1,230 @@
+// 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.
+
+//! The request id RustFS's outer layer minted, handed to the gateway as its own when that is safe.
+//!
+//! Responsible for: reading the id from the context the outer layer put in the request's
+//! extensions ([`RequestIdCarrier`]) and, when it is inside the gateway's identifier alphabet and
+//! length, inserting it as the gateway's [`HostRequestId`].
+//! NOT responsible for: minting any id, refusing any request (nothing here can fail a request),
+//! choosing the response headers (the gateway writes `x-amz-request-id` from the id it was
+//! handed, and RustFS's outer layer overwrites that header on S3 paths), or the layer that mints
+//! (`rustfs/src/server/layer.rs`). It never reads a header itself.
+//! Upstream: RustFS's `RequestContext`, which the binary makes a [`RequestIdCarrier`]. Downstream:
+//! the gateway service, which reads the [`HostRequestId`] extension once, at the top.
+//!
+//! # Whose id it is
+//!
+//! On S3 paths the outer layer mints a server-owned UUID. On admin, STS and console paths the
+//! context copies the caller's `x-request-id` (or `x-amz-request-id`), as legacy does, so there
+//! the id can be caller-chosen. Either way the id is attached only when it is inert, that is 1 to
+//! 64 bytes of ASCII letters, digits, `-`, `.`, `_` and `~`, which is safe in a header, an XML
+//! document and a log line. Anything else is not attached and the request is **forwarded
+//! unchanged**: the gateway then mints its own id. No request is ever refused because of its id.
+//!
+//! # Known difference when nothing is attached
+//!
+//! The gateway answers with the id it minted (error body `<RequestId>`, its events and audit
+//! records) while, on S3 paths, the outer layer overwrites the response header with RustFS's id.
+//! So for such a request the header and the error body name different ids. That needs a caller-
+//! chosen id outside the alphabet, or a request that never passed the outer layer.
+//!
+//! # Why this is not a `TraceSource`
+//!
+//! The gateway's `TraceSource::mint` takes no request, so a source cannot read the id RustFS
+//! already minted; the gateway's documented path for a host that identifies its own requests is
+//! the [`HostRequestId`] extension, and this module fills it.
+//!
+//! # Why a carrier trait
+//!
+//! `RequestContext` lives in the `rustfs` binary crate, which a library crate cannot depend on.
+//! The binary implements [`RequestIdCarrier`] for it, and this module looks it up in the
+//! extensions by that type.
+
+use rustfs_gateway::HostRequestId;
+
+/// A type in the request extensions that holds the id the outer layer minted or propagated.
+pub trait RequestIdCarrier: Send + Sync + 'static {
+    /// The canonical request id: the value RustFS logs and the response should carry.
+    fn request_id(&self) -> &str;
+}
+
+/// What [`attach_host_request_id`] did. Ignoring it is harmless: every outcome forwards the request.
+#[derive(Clone, Copy, Debug, PartialEq, Eq)]
+pub enum AttachOutcome {
+    /// The carrier's id was inserted as the request's [`HostRequestId`].
+    Attached,
+    /// The request carries no context from the outer layer; nothing was inserted.
+    NoContext,
+    /// The carrier's id is empty, too long, or outside the gateway's alphabet; nothing was
+    /// inserted and the id was not repaired.
+    UnusableId,
+}
+
+/// The gateway's [`HostRequestId`] for the id `carrier` holds, or `None` when it is not inert.
+#[must_use]
+pub fn host_request_id<C: RequestIdCarrier + ?Sized>(carrier: &C) -> Option<HostRequestId> {
+    HostRequestId::new(carrier.request_id()).ok()
+}
+
+/// Reads the `C` the outer layer put in `request`'s extensions and, when its id is inert, inserts
+/// it as the request's [`HostRequestId`].
+///
+/// Never refuses the request: the caller forwards it whatever the outcome.
+pub fn attach_host_request_id<C: RequestIdCarrier, B>(request: &mut http::Request<B>) -> AttachOutcome {
+    let Some(carrier) = request.extensions().get::<C>() else {
+        return AttachOutcome::NoContext;
+    };
+    let Some(id) = host_request_id(carrier) else {
+        return AttachOutcome::UnusableId;
+    };
+    request.extensions_mut().insert(id);
+    AttachOutcome::Attached
+}
+
+#[cfg(test)]
+#[allow(clippy::expect_used, clippy::panic, clippy::unwrap_used)]
+mod tests {
+    use super::*;
+
+    #[derive(Clone)]
+    struct Ctx(String);
+
+    impl RequestIdCarrier for Ctx {
+        fn request_id(&self) -> &str {
+            &self.0
+        }
+    }
+
+    const UUID: &str = "7c9e6679-7425-40de-944b-e07fc1f90ae7";
+
+    fn request_with(id: &str) -> http::Request<()> {
+        let mut request = http::Request::new(());
+        request.extensions_mut().insert(Ctx(id.to_owned()));
+        request
+    }
+
+    fn handed_over(request: &http::Request<()>) -> Option<String> {
+        request
+            .extensions()
+            .get::<HostRequestId>()
+            .map(|id| id.request_id().as_str().to_owned())
+    }
+
+    fn attach(request: &mut http::Request<()>) -> AttachOutcome {
+        attach_host_request_id::<Ctx, _>(request)
+    }
+
+    /// Positive: the id RustFS minted is the id the gateway is handed, byte for byte.
+    #[test]
+    fn the_minted_id_is_handed_over_unchanged() {
+        let mut request = request_with(UUID);
+        assert_eq!(attach(&mut request), AttachOutcome::Attached);
+        assert_eq!(handed_over(&request).as_deref(), Some(UUID));
+    }
+
+    /// Positive: every shape RustFS mints is accepted, including its two fallbacks.
+    #[test]
+    fn every_shape_rustfs_mints_is_accepted() {
+        for id in [UUID, "trace-0123456789abcdef", "req-0123456789abcdef"] {
+            let mut request = request_with(id);
+            assert_eq!(attach(&mut request), AttachOutcome::Attached, "{id}");
+            assert_eq!(handed_over(&request).as_deref(), Some(id));
+        }
+    }
+
+    /// Positive: a caller-supplied id that is inert (admin and STS paths copy it) is attached, and
+    /// the longest allowed one is accepted.
+    #[test]
+    fn an_inert_caller_supplied_id_is_attached() {
+        let longest = "a".repeat(64);
+        for id in ["abc", "my.trace_id~1-2", longest.as_str()] {
+            let mut request = request_with(id);
+            assert_eq!(attach(&mut request), AttachOutcome::Attached, "{id}");
+            assert_eq!(handed_over(&request).as_deref(), Some(id));
+        }
+    }
+
+    /// Positive: two requests keep their own ids; nothing is shared or minted between them.
+    #[test]
+    fn each_request_keeps_its_own_id() {
+        let mut first = request_with("first-id");
+        let mut second = request_with("second-id");
+        attach(&mut first);
+        attach(&mut second);
+        assert_eq!(handed_over(&first).as_deref(), Some("first-id"));
+        assert_eq!(handed_over(&second).as_deref(), Some("second-id"));
+    }
+
+    /// Positive: with both a carrier and an `x-amz-request-id` header, the carrier's id wins.
+    #[test]
+    fn the_carrier_wins_over_a_request_id_header() {
+        let mut request = http::Request::builder()
+            .header("x-amz-request-id", "caller-chosen")
+            .header("x-request-id", "caller-chosen")
+            .body(())
+            .expect("a request");
+        request.extensions_mut().insert(Ctx(UUID.to_owned()));
+        assert_eq!(attach(&mut request), AttachOutcome::Attached);
+        assert_eq!(handed_over(&request).as_deref(), Some(UUID));
+    }
+
+    /// Negative: a request the outer layer never saw gets no id, and the request is not refused.
+    #[test]
+    fn a_missing_context_hands_over_nothing() {
+        let mut request = http::Request::new(());
+        assert_eq!(attach(&mut request), AttachOutcome::NoContext);
+        assert_eq!(handed_over(&request), None);
+    }
+
+    /// Negative: a header named like the id is not the id; only the context is read.
+    #[test]
+    fn a_request_id_header_alone_is_not_read() {
+        let mut request = http::Request::builder()
+            .header("x-amz-request-id", "caller-chosen")
+            .header("x-request-id", "caller-chosen")
+            .body(())
+            .expect("a request");
+        assert_eq!(attach(&mut request), AttachOutcome::NoContext);
+        assert_eq!(handed_over(&request), None);
+    }
+
+    /// Negative: an id outside the alphabet or the length is never attached or repaired, and the
+    /// outcome says so rather than failing the request.
+    #[test]
+    fn an_id_the_gateway_cannot_carry_is_not_attached() {
+        let too_long = "a".repeat(65);
+        for id in [
+            "",
+            "has space",
+            "line\r\nbreak",
+            "<RequestId>",
+            "quo\"te",
+            "Root=1-5759e988",
+            too_long.as_str(),
+        ] {
+            let mut request = request_with(id);
+            assert_eq!(attach(&mut request), AttachOutcome::UnusableId, "{id:?}");
+            assert_eq!(handed_over(&request), None, "{id:?}");
+        }
+    }
+
+    /// Negative: the unusable id is not repaired into a shorter or cleaned-up one.
+    #[test]
+    fn host_request_id_does_not_repair() {
+        assert!(host_request_id(&Ctx("has space".to_owned())).is_none());
+        assert!(host_request_id(&Ctx(format!("{UUID} "))).is_none());
+        assert!(host_request_id(&Ctx(UUID.to_owned())).is_some());
+    }
+}

scripts/README.md

@@ -169,7 +169,7 @@ their issue closes.
 |---|---|---|---|
 | `codemods/` | dev-tool | Scripted, idempotent source migrations for the gateway migration (rustfs/backlog#2734); `s3_error_to_s3_types.py` moves s3s error-type paths onto `rustfs-s3-types` per crate closure and refuses the s3s edge allowlist | `make script-tests` (`--self-test`); usage in each script's docstring |
 | `fuzz/` | ci-gate | Unified cargo-fuzz runner and helpers for the `fuzz/` sub-workspace | fuzz.yml; `fuzz/README.md` |
-| `release/` | ci-gate | Release creation and DEB/RPM version-normalization helpers | build.yml; package.yml |
+| `release/` | ci-gate | Release creation and DEB/RPM version-normalization helpers | build.yml; package.yml; manual-main-packages.yml |
 | `s3-tests/` | ci-gate | ceph/s3-tests compatibility harness (allow-lists, patches, report tooling) | ci.yml; e2e-s3tests.yml; `scripts/s3-tests/README.md` |
 | `security/` | ci-gate | Workflow-pin enforcement and release supply-chain asset generation | audit.yml; build.yml |
 | `table-catalog/` | dev-tool | S3-Tables / pyiceberg validation suite | `docs/architecture/s3-tables-support-matrix.md` |

scripts/download_rc.sh

@@ -13,7 +13,7 @@ OUTPUT="$4"
 
 case "$ARCH" in
   amd64|arm64) ;;
-  *) echo "Unsupported rc architecture: $ARCH" >&2; exit 1 ;;
+  *) echo "Unsupported rc architecture: '$ARCH' (expected amd64 or arm64; an empty value means the caller's TARGETARCH build argument was not set)" >&2; exit 1 ;;
 esac
 case "$LIBC" in
   musl) LIBC_SUFFIX="" ;;

scripts/release/package_manual_artifacts.sh

@@ -0,0 +1,406 @@
+#!/usr/bin/env bash
+# Copyright 2026 RustFS Team
+# SPDX-License-Identifier: Apache-2.0
+# Manual zip/deb/rpm packaging for one Linux target.
+# Version rules and deb/rpm metadata come from package_versions.sh.
+# deb/rpm payloads are built by build_deb.sh and build_rpm.sh.
+# Dependency strings match scripts/release/deliver.py package().
+
+set -euo pipefail
+
+repo_root=$(cd "$(dirname "${BASH_SOURCE[0]}")/../.." && pwd)
+helper="${repo_root}/scripts/release/package_versions.sh"
+
+die() {
+  printf 'package_manual_artifacts: %s\n' "$1" >&2
+  exit 1
+}
+
+usage() {
+  cat <<'EOF' >&2
+usage: package_manual_artifacts.sh matrix <x86|arm|all>
+       package_manual_artifacts.sh validate --version VER --arch ARCH --libc LIBC
+       package_manual_artifacts.sh package --version VER --arch ARCH --libc LIBC --binary PATH --output DIR [--zip-extra FILE]...
+EOF
+  exit 2
+}
+
+require_value() {
+  local flag=$1
+  # An empty value is a version/path error, not a missing argument. Flag-shaped
+  # values mean the next option was consumed by mistake.
+  if [[ $# -lt 2 || ${2:-} == --* ]]; then
+    die "missing value for ${flag}"
+  fi
+}
+
+resolve_version() {
+  local version=$1 arch=$2 libc=$3
+  local release_err prerelease_err release_out prerelease_out
+  release_err=$(mktemp)
+  prerelease_err=$(mktemp)
+  if release_out=$(bash "$helper" release "$version" "" "$arch" "$libc" 2>"$release_err"); then
+    printf 'release\n%s\n' "$release_out"
+    rm -f "$release_err" "$prerelease_err"
+    return 0
+  fi
+  if prerelease_out=$(bash "$helper" prerelease "$version" "" "$arch" "$libc" 2>"$prerelease_err"); then
+    printf 'prerelease\n%s\n' "$prerelease_out"
+    rm -f "$release_err" "$prerelease_err"
+    return 0
+  fi
+  printf 'invalid package version: %s\n' "$version" >&2
+  printf 'not a release: %s\n' "$(<"$release_err")" >&2
+  printf 'not a prerelease: %s\n' "$(<"$prerelease_err")" >&2
+  rm -f "$release_err" "$prerelease_err"
+  return 1
+}
+
+load_versions() {
+  local resolved line key value
+  resolved=$(resolve_version "$version" "$arch" "$libc") || exit 1
+  build_type=${resolved%%$'\n'*}
+  local rest=${resolved#*$'\n'}
+  deb_version=''
+  rpm_version=''
+  rpm_release=''
+  deb_file=''
+  rpm_file=''
+  while IFS= read -r line || [[ -n "$line" ]]; do
+    [[ -n "$line" ]] || continue
+    key=${line%%=*}
+    value=${line#*=}
+    case "$key" in
+      deb_version) deb_version=$value ;;
+      rpm_version) rpm_version=$value ;;
+      rpm_release) rpm_release=$value ;;
+      deb_file) deb_file=$value ;;
+      rpm_file) rpm_file=$value ;;
+      *) die "unexpected package_versions field: ${key}" ;;
+    esac
+  done <<< "$rest"
+  [[ -n "$deb_version" && -n "$rpm_version" && -n "$rpm_release" && -n "$deb_file" && -n "$rpm_file" ]] ||
+    die "incomplete package version output"
+  local stem=${deb_file%.deb}
+  [[ "$stem" == "${rpm_file%.rpm}" ]] || die "deb and rpm file names use different stems"
+  [[ "$stem" == "rustfs-linux-${arch}-${libc}-v${version}" ]] ||
+    die "package stem does not match rustfs-linux-${arch}-${libc}-v${version}"
+  zip_file="${stem}.zip"
+  case "$build_type" in
+    release | prerelease) ;;
+    *) die "unsupported build type: ${build_type}" ;;
+  esac
+}
+
+emit_resolved() {
+  printf 'build_type=%s\n' "$build_type"
+  printf 'deb_version=%s\n' "$deb_version"
+  printf 'rpm_version=%s\n' "$rpm_version"
+  printf 'rpm_release=%s\n' "$rpm_release"
+  printf 'deb_file=%s\n' "$deb_file"
+  printf 'rpm_file=%s\n' "$rpm_file"
+  printf 'zip_file=%s\n' "$zip_file"
+}
+
+emit_matrix() {
+  local choice=$1
+  command -v python3 >/dev/null 2>&1 || die "python3 is required"
+  python3 - "$choice" <<'PY'
+import json
+import sys
+
+choice = sys.argv[1]
+# Linux rows match the release matrix in .github/workflows/build.yml.
+# arch, libc, and build_os are the packaging names for those same rows.
+rows = [
+    {"target_id": "linux-x86_64-musl", "os": "sm-standard-4", "target": "x86_64-unknown-linux-musl", "cross": False, "platform": "linux", "rustflags": "", "features": "", "glibc": "", "arch": "x86_64", "libc": "musl", "build_os": "linux-x86_64"},
+    {"target_id": "linux-aarch64-musl", "os": "ubicloud-standard-16-arm", "target": "aarch64-unknown-linux-musl", "cross": False, "platform": "linux", "rustflags": "", "features": "", "glibc": "", "arch": "aarch64", "libc": "musl", "build_os": "linux-aarch64"},
+    {"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", "arch": "x86_64", "libc": "gnu", "build_os": "linux-x86_64"},
+    {"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", "arch": "aarch64", "libc": "gnu", "build_os": "linux-aarch64"},
+]
+if choice == "all":
+    selected = rows
+elif choice == "x86":
+    selected = [row for row in rows if row["arch"] == "x86_64"]
+elif choice == "arm":
+    selected = [row for row in rows if row["arch"] == "aarch64"]
+else:
+    raise SystemExit(
+        "package_manual_artifacts: unsupported architecture "
+        f"(expected x86, arm, or all): {choice}"
+    )
+if len(selected) == 0:
+    raise SystemExit("package_manual_artifacts: no targets selected")
+json.dump({"include": selected}, sys.stdout, separators=(",", ":"))
+sys.stdout.write("\n")
+PY
+}
+
+parse_identity() {
+  version=''
+  arch=''
+  libc=''
+  local saw_version=0 saw_arch=0 saw_libc=0
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --version)
+        require_value "$@"
+        saw_version=1
+        version=$2
+        shift 2
+        ;;
+      --arch)
+        require_value "$@"
+        saw_arch=1
+        arch=$2
+        shift 2
+        ;;
+      --libc)
+        require_value "$@"
+        saw_libc=1
+        libc=$2
+        shift 2
+        ;;
+      *)
+        die "unknown argument: $1"
+        ;;
+    esac
+  done
+  [[ "$saw_version" == 1 && "$saw_arch" == 1 && "$saw_libc" == 1 ]] ||
+    die "validate requires --version, --arch, and --libc"
+}
+
+abspath() {
+  local path=$1
+  local dir base
+  dir=$(cd "$(dirname "$path")" && pwd)
+  base=$(basename "$path")
+  printf '%s/%s' "$dir" "$base"
+}
+
+sha256_file() {
+  sha256sum "$1" | awk 'NR==1 { print $1 }'
+}
+
+package_binary() {
+  local binary='' output=''
+  local -a zip_extras=()
+  local saw_version=0 saw_arch=0 saw_libc=0
+  version=''
+  arch=''
+  libc=''
+  while [[ $# -gt 0 ]]; do
+    case "$1" in
+      --version)
+        require_value "$@"
+        saw_version=1
+        version=$2
+        shift 2
+        ;;
+      --arch)
+        require_value "$@"
+        saw_arch=1
+        arch=$2
+        shift 2
+        ;;
+      --libc)
+        require_value "$@"
+        saw_libc=1
+        libc=$2
+        shift 2
+        ;;
+      --binary)
+        require_value "$@"
+        binary=$2
+        shift 2
+        ;;
+      --output)
+        require_value "$@"
+        output=$2
+        shift 2
+        ;;
+      --zip-extra)
+        require_value "$@"
+        zip_extras+=("$2")
+        shift 2
+        ;;
+      *)
+        die "unknown argument: $1"
+        ;;
+    esac
+  done
+  [[ "$saw_version" == 1 && "$saw_arch" == 1 && "$saw_libc" == 1 && -n "$binary" && -n "$output" ]] ||
+    die "package requires --version, --arch, --libc, --binary, and --output"
+  [[ -f "$binary" && ! -L "$binary" ]] || die "binary must be a regular file"
+  load_versions
+
+  local -a extra_names=()
+  local extra base seen=' '
+  local has_json=0 has_sha=0
+  for extra in "${zip_extras[@]}"; do
+    base=$(basename "$extra")
+    case "$base" in
+      rustfs.cpu-symbol-catalog.json) has_json=1 ;;
+      rustfs.cpu-symbol-catalog.sha256) has_sha=1 ;;
+      *) die "unsupported zip member: ${base}" ;;
+    esac
+    [[ "$seen" != *" ${base} "* ]] || die "duplicate zip member: ${base}"
+    seen+=" ${base} "
+    [[ -f "$extra" && ! -L "$extra" ]] || die "zip extra must be a regular file: ${base}"
+    extra_names+=("$base")
+  done
+  [[ "$has_json" == "$has_sha" ]] ||
+    die "cpu symbol catalog json and sha256 must be packaged together"
+
+  local tool
+  for tool in zip unzip sha256sum dpkg-deb fakeroot rpm rpm2cpio cpio fpm; do
+    command -v "$tool" >/dev/null 2>&1 || die "required tool not found: ${tool}"
+  done
+
+  binary=$(abspath "$binary")
+  local -a absolute_extras=()
+  for extra in "${zip_extras[@]}"; do
+    absolute_extras+=("$(abspath "$extra")")
+  done
+  mkdir -p "$output"
+  output=$(cd "$output" && pwd)
+  local produced
+  for produced in "$zip_file" "$deb_file" "$rpm_file"; do
+    [[ ! -e "$output/$produced" ]] || die "refusing to overwrite ${output}/${produced}"
+  done
+
+  # Globals so the EXIT trap can still see them after this function returns.
+  work=$(mktemp -d)
+  stage=$(mktemp -d)
+  cleanup() {
+    rm -rf "${work:-}" "${stage:-}"
+  }
+  trap cleanup EXIT
+
+  local zip_dir="$work/zip"
+  mkdir -p "$zip_dir"
+  cp "$binary" "$zip_dir/rustfs"
+  chmod 755 "$zip_dir/rustfs"
+  local index=0
+  for extra in "${absolute_extras[@]}"; do
+    cp "$extra" "$zip_dir/${extra_names[$index]}"
+    index=$((index + 1))
+  done
+  (
+    cd "$zip_dir"
+    zip "$stage/$zip_file" rustfs "${extra_names[@]}"
+  )
+
+  local members expected
+  members=$(unzip -Z1 "$stage/$zip_file" | sort)
+  expected=$(printf '%s\n' rustfs "${extra_names[@]}" | sort)
+  [[ "$members" == "$expected" ]] || die "zip members do not match the release archive"
+  [[ "$(unzip -p "$stage/$zip_file" rustfs | sha256sum | awk '{print $1}')" == "$(sha256_file "$binary")" ]] ||
+    die "zip binary differs from the packaged executable"
+
+  ln -s "$repo_root/scripts" "$work/scripts"
+  ln -s "$repo_root/deploy" "$work/deploy"
+  cp "$repo_root/LICENSE" "$repo_root/README.md" "$work/"
+  mkdir -p "$work/bin"
+  cp "$binary" "$work/bin/rustfs"
+  chmod 755 "$work/bin/rustfs"
+
+  local deb_arch deb_depends rpm_depends
+  case "$arch" in
+    x86_64) deb_arch=amd64 ;;
+    aarch64) deb_arch=arm64 ;;
+    *) die "unsupported architecture: ${arch}" ;;
+  esac
+  case "$libc" in
+    gnu)
+      deb_depends='libc6 (>= 2.31)'
+      rpm_depends='glibc >= 2.31'
+      ;;
+    musl)
+      deb_depends=''
+      rpm_depends=''
+      ;;
+    *) die "unsupported libc: ${libc}" ;;
+  esac
+
+  (
+    cd "$work"
+    GITHUB_OUTPUT="$work/outputs" \
+      DEB_VERSION="$deb_version" \
+      RPM_VERSION="$rpm_version" \
+      RPM_RELEASE="$rpm_release" \
+      DEB_FILE="$deb_file" \
+      RPM_FILE="$rpm_file" \
+      DEB_ARCH="$deb_arch" \
+      RPM_ARCH="$arch" \
+      DEB_DEPENDS="$deb_depends" \
+      RPM_DEPENDS="$rpm_depends" \
+      LIBC="$libc" \
+      bash "$repo_root/scripts/release/build_deb.sh"
+    GITHUB_OUTPUT="$work/outputs" \
+      DEB_VERSION="$deb_version" \
+      RPM_VERSION="$rpm_version" \
+      RPM_RELEASE="$rpm_release" \
+      DEB_FILE="$deb_file" \
+      RPM_FILE="$rpm_file" \
+      DEB_ARCH="$deb_arch" \
+      RPM_ARCH="$arch" \
+      DEB_DEPENDS="$deb_depends" \
+      RPM_DEPENDS="$rpm_depends" \
+      LIBC="$libc" \
+      bash "$repo_root/scripts/release/build_rpm.sh"
+  )
+  cp "$work/$deb_file" "$stage/$deb_file"
+  cp "$work/$rpm_file" "$stage/$rpm_file"
+
+  local unpacked expected_hash actual_hash
+  expected_hash=$(sha256_file "$binary")
+  unpacked="$work/unpacked-deb"
+  mkdir -p "$unpacked"
+  dpkg-deb --extract "$stage/$deb_file" "$unpacked"
+  actual_hash=$(sha256_file "$unpacked/usr/bin/rustfs")
+  [[ "$actual_hash" == "$expected_hash" ]] || die "deb binary differs from the packaged executable"
+  [[ "$(dpkg-deb -f "$stage/$deb_file" Package)" == rustfs ]] || die "deb package name mismatch"
+  [[ "$(dpkg-deb -f "$stage/$deb_file" Version)" == "$deb_version" ]] || die "deb version mismatch"
+  [[ "$(dpkg-deb -f "$stage/$deb_file" Architecture)" == "$deb_arch" ]] || die "deb architecture mismatch"
+
+  unpacked="$work/unpacked-rpm"
+  mkdir -p "$unpacked"
+  rpm2cpio "$stage/$rpm_file" | (cd "$unpacked" && cpio --extract --make-directories --quiet ./usr/bin/rustfs)
+  actual_hash=$(sha256_file "$unpacked/usr/bin/rustfs")
+  [[ "$actual_hash" == "$expected_hash" ]] || die "rpm binary differs from the packaged executable"
+  local rpm_metadata expected_metadata
+  rpm_metadata=$(rpm -qp --qf '%{NAME}\n%{VERSION}\n%{RELEASE}\n%{ARCH}\n' "$stage/$rpm_file")
+  expected_metadata=$(printf 'rustfs\n%s\n%s\n%s' "$rpm_version" "$rpm_release" "$arch")
+  [[ "$rpm_metadata" == "$expected_metadata" ]] || die "rpm metadata mismatch"
+
+  mv "$stage/$zip_file" "$output/$zip_file"
+  mv "$stage/$deb_file" "$output/$deb_file"
+  mv "$stage/$rpm_file" "$output/$rpm_file"
+  emit_resolved
+}
+
+main() {
+  local command=${1:-}
+  case "$command" in
+    matrix)
+      [[ $# -eq 2 ]] || usage
+      emit_matrix "$2"
+      ;;
+    validate)
+      shift
+      parse_identity "$@"
+      load_versions
+      emit_resolved
+      ;;
+    package)
+      shift
+      package_binary "$@"
+      ;;
+    *)
+      usage
+      ;;
+  esac
+}
+
+main "$@"

scripts/release/test_manual_main_packages.py

@@ -0,0 +1,226 @@
+#!/usr/bin/env python3
+# Copyright 2026 RustFS Team
+# SPDX-License-Identifier: Apache-2.0
+"""Contract tests for manual origin/main packaging."""
+from __future__ import annotations
+
+import json
+import os
+from pathlib import Path
+import re
+import shutil
+import subprocess
+import sys
+import tempfile
+import unittest
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPT = ROOT / "scripts" / "release" / "package_manual_artifacts.sh"
+WORKFLOW = ROOT / ".github" / "workflows" / "manual-main-packages.yml"
+sys.path.insert(0, str(ROOT / "scripts"))
+from ci_gate import workflow_document  # noqa: E402
+import validation_registry as registry  # noqa: E402
+
+
+def run_script(*args, check=False):
+    return subprocess.run(["bash", str(SCRIPT), *args], cwd=ROOT, text=True,
+                          capture_output=True, timeout=60, check=check)
+
+
+def trigger(text):
+    match = re.search(r"^on:\n(.*?)(?=^[^ #\n]|\Z)", text, re.M | re.S)
+    if match is None:
+        raise AssertionError("workflow has no event declaration")
+    return re.findall(r"^  (\w+):", match.group(1), re.M)
+
+
+class VersionAndMatrixTests(unittest.TestCase):
+    def test_release_and_prerelease_names_come_from_package_versions(self):
+        cases = {
+            ("1.2.3", "x86_64", "gnu"): """\
+build_type=release
+deb_version=1.2.3
+rpm_version=1.2.3
+rpm_release=1
+deb_file=rustfs-linux-x86_64-gnu-v1.2.3.deb
+rpm_file=rustfs-linux-x86_64-gnu-v1.2.3.rpm
+zip_file=rustfs-linux-x86_64-gnu-v1.2.3.zip
+""",
+            ("1.2.3-alpha.1", "aarch64", "gnu"): """\
+build_type=prerelease
+deb_version=1.2.3~alpha.1
+rpm_version=1.2.3~alpha.1
+rpm_release=1
+deb_file=rustfs-linux-aarch64-gnu-v1.2.3-alpha.1.deb
+rpm_file=rustfs-linux-aarch64-gnu-v1.2.3-alpha.1.rpm
+zip_file=rustfs-linux-aarch64-gnu-v1.2.3-alpha.1.zip
+""",
+            ("1.2.3-beta.2", "x86_64", "musl"): """\
+build_type=prerelease
+deb_version=1.2.3~beta.2
+rpm_version=1.2.3~beta.2
+rpm_release=1
+deb_file=rustfs-linux-x86_64-musl-v1.2.3-beta.2.deb
+rpm_file=rustfs-linux-x86_64-musl-v1.2.3-beta.2.rpm
+zip_file=rustfs-linux-x86_64-musl-v1.2.3-beta.2.zip
+""",
+            ("1.2.3-rc.4", "aarch64", "musl"): """\
+build_type=prerelease
+deb_version=1.2.3~rc.4
+rpm_version=1.2.3~rc.4
+rpm_release=1
+deb_file=rustfs-linux-aarch64-musl-v1.2.3-rc.4.deb
+rpm_file=rustfs-linux-aarch64-musl-v1.2.3-rc.4.rpm
+zip_file=rustfs-linux-aarch64-musl-v1.2.3-rc.4.zip
+""",
+        }
+        for (version, arch, libc), expected in cases.items():
+            with self.subTest(version=version, arch=arch, libc=libc):
+                result = run_script("validate", "--version", version, "--arch", arch, "--libc", libc)
+                self.assertEqual(result.returncode, 0, result.stderr)
+                self.assertEqual(result.stdout, expected)
+
+    def test_invalid_versions_fail_without_stdout(self):
+        for version in ("", "v1.2.3", "1.2", "01.2.3", "1.2.3-preview.1", "1.2.3-alpha",
+                        "1.2.3-rc.01", "1.2.3-rc.1-extra", "1.2.3 ", "1.2.3$(id)", "dev-abcdef"):
+            with self.subTest(version=version):
+                result = run_script("validate", "--version", version, "--arch", "x86_64", "--libc", "gnu")
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertIn("invalid package version", result.stderr)
+                self.assertIn("not a release:", result.stderr)
+                self.assertIn("not a prerelease:", result.stderr)
+
+    def test_invalid_version_does_not_create_packages(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            binary = root / "rustfs"
+            binary.write_bytes(b"not packaged")
+            output = root / "dist"
+            result = run_script("package", "--version", "v1.2.3", "--arch", "x86_64", "--libc", "gnu",
+                                "--binary", str(binary), "--output", str(output))
+            self.assertNotEqual(result.returncode, 0)
+            self.assertEqual(result.stdout, "")
+            self.assertFalse(output.exists())
+
+    def test_catalog_members_must_be_the_official_pair(self):
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            binary = root / "rustfs"
+            binary.write_bytes(b"not packaged")
+            catalog = root / "rustfs.cpu-symbol-catalog.json"
+            catalog.write_text("{}\n", encoding="utf-8")
+            result = run_script("package", "--version", "1.2.3", "--arch", "x86_64", "--libc", "gnu",
+                                "--binary", str(binary), "--output", str(root / "dist"),
+                                "--zip-extra", str(catalog))
+            self.assertNotEqual(result.returncode, 0)
+            self.assertIn("packaged together", result.stderr)
+            self.assertFalse((root / "dist").exists())
+            other = root / "notes.txt"
+            other.write_text("nope\n", encoding="utf-8")
+            rejected = run_script("package", "--version", "1.2.3", "--arch", "x86_64", "--libc", "musl",
+                                  "--binary", str(binary), "--output", str(root / "dist"),
+                                  "--zip-extra", str(other))
+            self.assertNotEqual(rejected.returncode, 0)
+            self.assertIn("unsupported zip member", rejected.stderr)
+
+    def test_architecture_matrix_matches_official_linux_rows(self):
+        source = (ROOT / ".github" / "workflows" / "build.yml").read_text(encoding="utf-8")
+        official = json.loads(re.search(r"all='(\{.*?\})'", source, re.S).group(1))["include"]
+        linux = [row for row in official if row["platform"] == "linux"]
+
+        def derived(row):
+            arch = "aarch64" if row["target"].startswith("aarch64-") else "x86_64"
+            libc = row["target_id"].rsplit("-", 1)[1]
+            return {**row, "arch": arch, "libc": libc, "build_os": f"linux-{arch}"}
+
+        expected = {
+            "all": [derived(row) for row in linux],
+            "x86": [derived(row) for row in linux if row["target"].startswith("x86_64-")],
+            "arm": [derived(row) for row in linux if row["target"].startswith("aarch64-")],
+        }
+        self.assertEqual([row["target_id"] for row in expected["x86"]],
+                         ["linux-x86_64-musl", "linux-x86_64-gnu"])
+        self.assertEqual([row["target_id"] for row in expected["arm"]],
+                         ["linux-aarch64-musl", "linux-aarch64-gnu"])
+        for choice, rows in expected.items():
+            with self.subTest(choice=choice):
+                result = run_script("matrix", choice, check=True)
+                self.assertEqual(json.loads(result.stdout), {"include": rows})
+        for choice in ("", "ALL", "linux", "x86_64", "x86,arm"):
+            with self.subTest(choice=choice):
+                result = run_script("matrix", choice)
+                self.assertNotEqual(result.returncode, 0)
+                self.assertEqual(result.stdout, "")
+                self.assertIn("unsupported architecture", result.stderr)
+
+
+class WorkflowContractTests(unittest.TestCase):
+    def test_dispatch_builds_origin_main_and_only_uploads_artifacts(self):
+        text = WORKFLOW.read_text(encoding="utf-8")
+        self.assertEqual(trigger(text), ["workflow_dispatch"])
+        document = workflow_document(text)
+        events = document.get("on", document.get(True))
+        self.assertEqual(set(events), {"workflow_dispatch"})
+        inputs = events["workflow_dispatch"]["inputs"]
+        self.assertEqual(inputs["architecture"]["type"], "choice")
+        self.assertEqual(inputs["architecture"]["options"], ["x86", "arm", "all"])
+        self.assertTrue(inputs["architecture"]["required"])
+        self.assertEqual(inputs["version"]["type"], "string")
+        self.assertTrue(inputs["version"]["required"])
+        self.assertEqual(document["permissions"], {"contents": "read"})
+        self.assertEqual(re.findall(r"^\s+ref:\s*(.+)$", text, re.M),
+                         ["main", "${{ needs.prepare.outputs.source_sha }}"])
+        self.assertIn("refs/heads/main", text)
+        self.assertNotIn("github.sha", text)
+        self.assertIn("package_manual_artifacts.sh validate", text)
+        self.assertIn("package_manual_artifacts.sh matrix", text)
+        self.assertIn("package_manual_artifacts.sh package", text)
+        self.assertIn("RUSTFS_BUILD_VERSION: ${{ needs.prepare.outputs.version }}", text)
+        self.assertIn("cargo zigbuild --release --locked", text)
+        self.assertIn("cargo build --release --locked", text)
+        self.assertIn("scripts/release/check_glibc.py", text)
+        self.assertIn('cache-save-if: \'false\'', text)
+        self.assertIn("actions/upload-artifact@", text)
+        self.assertIn("dist/*", text)
+        for forbidden in ("contents: write", "aws s3", "gh release", "softprops/action-gh-release",
+                          "docker push", "R2_ACCESS_KEY_ID", "workflow_call:"):
+            self.assertNotIn(forbidden, text)
+        self.assertEqual(text.count("persist-credentials: false"), text.count("actions/checkout@"))
+        item = registry.workflow_entry(".github/workflows/manual-main-packages.yml")
+        self.assertEqual(item["name"], "Manual main packages")
+        self.assertEqual(item["events"], ["workflow_dispatch"])
+        self.assertEqual(item["schedule"], [])
+        self.assertEqual(item["monitor"], "none")
+        self.assertNotEqual(item["origin"], "baseline")
+
+    def test_registry_accepts_the_added_workflow(self):
+        self.assertTrue(registry.check_wiring(registry.load_registry()))
+
+
+class PackageToolTests(unittest.TestCase):
+    def test_zip_deb_and_rpm_wrap_the_same_binary_when_tools_exist(self):
+        required = ("dpkg-deb", "fakeroot", "rpm", "rpm2cpio", "cpio", "fpm", "zip", "unzip")
+        missing = [tool for tool in required if shutil.which(tool) is None]
+        if missing or os.name != "posix":
+            self.skipTest("packaging tools not installed: " + ", ".join(missing))
+        with tempfile.TemporaryDirectory() as temp:
+            root = Path(temp)
+            payload = b"manual main package fixture\n"
+            binary = root / "rustfs"
+            binary.write_bytes(payload)
+            result = run_script("package", "--version", "9.9.9-rc.8", "--arch", "x86_64", "--libc", "gnu",
+                                "--binary", str(binary), "--output", str(root / "dist"))
+            self.assertEqual(result.returncode, 0, result.stderr)
+            dist = root / "dist"
+            self.assertEqual(sorted(path.name for path in dist.iterdir()), [
+                "rustfs-linux-x86_64-gnu-v9.9.9-rc.8.deb",
+                "rustfs-linux-x86_64-gnu-v9.9.9-rc.8.rpm",
+                "rustfs-linux-x86_64-gnu-v9.9.9-rc.8.zip",
+            ])
+            self.assertIn("deb_version=9.9.9~rc.8", result.stdout)
+            self.assertIn("zip_file=rustfs-linux-x86_64-gnu-v9.9.9-rc.8.zip", result.stdout)
+
+
+if __name__ == "__main__":
+    unittest.main()

scripts/test_release_packages.py

@@ -11,6 +11,7 @@ import subprocess
 import sys
 import tempfile
 import unittest
+import zipfile
 from unittest.mock import patch
 
 sys.path.insert(0, str(Path(__file__).resolve().parent / "release"))
@@ -63,6 +64,57 @@ class PackageIntegrationTests(unittest.TestCase):
                     deliver.package(fixtures.inputs(), "linux-x86_64-musl", root / "output")
             self.assertFalse(list((root / "output").glob("packages-*.json")))
 
+    def test_manual_packages_match_release_names_and_bytes(self):
+        script = Path(__file__).resolve().parent / "release" / "package_manual_artifacts.sh"
+        cases = (
+            ("9.9.9-rc.8", "x86_64", "gnu", True),
+            ("9.9.9", "aarch64", "musl", False),
+        )
+        for version, arch, libc, with_catalog in cases:
+            with self.subTest(version=version, arch=arch, libc=libc), tempfile.TemporaryDirectory() as temp:
+                root = Path(temp)
+                payload = f"manual-package:{version}:{arch}:{libc}\n".encode()
+                binary = root / "rustfs"
+                binary.write_bytes(payload)
+                command = ["bash", str(script), "package", "--version", version, "--arch", arch, "--libc", libc,
+                           "--binary", str(binary), "--output", str(root / "dist")]
+                if with_catalog:
+                    for name in ("rustfs.cpu-symbol-catalog.json", "rustfs.cpu-symbol-catalog.sha256"):
+                        (root / name).write_text(name + "\n", encoding="utf-8")
+                        command.extend(["--zip-extra", str(root / name)])
+                subprocess.run(command, check=True, text=True, timeout=180)
+                stem = f"rustfs-linux-{arch}-{libc}-v{version}"
+                dist = root / "dist"
+                self.assertEqual(sorted(path.name for path in dist.iterdir()),
+                                 [f"{stem}.deb", f"{stem}.rpm", f"{stem}.zip"])
+                expected_hash = hashlib.sha256(payload).hexdigest()
+                with zipfile.ZipFile(dist / f"{stem}.zip") as archive:
+                    names = sorted(archive.namelist())
+                    self.assertEqual(hashlib.sha256(archive.read("rustfs")).hexdigest(), expected_hash)
+                if with_catalog:
+                    self.assertEqual(names, ["rustfs", "rustfs.cpu-symbol-catalog.json", "rustfs.cpu-symbol-catalog.sha256"])
+                else:
+                    self.assertEqual(names, ["rustfs"])
+                deb = dist / f"{stem}.deb"
+                self.assertEqual(subprocess.check_output(["dpkg-deb", "-f", deb, "Version"], text=True).strip(),
+                                 version.replace("-", "~"))
+                self.assertEqual(subprocess.check_output(["dpkg-deb", "-f", deb, "Architecture"], text=True).strip(),
+                                 "amd64" if arch == "x86_64" else "arm64")
+                depends = subprocess.check_output(["dpkg-deb", "-f", deb, "Depends"], text=True).strip()
+                self.assertEqual(depends, "libc6 (>= 2.31)" if libc == "gnu" else "")
+                unpacked = root / "deb"
+                subprocess.run(["dpkg-deb", "--extract", str(deb), str(unpacked)], check=True)
+                self.assertEqual(hashlib.sha256((unpacked / "usr/bin/rustfs").read_bytes()).hexdigest(), expected_hash)
+                rpm = dist / f"{stem}.rpm"
+                identity = subprocess.check_output(
+                    ["rpm", "-qp", "--qf", "%{NAME} %{VERSION} %{RELEASE} %{ARCH}", str(rpm)], text=True)
+                self.assertEqual(identity, f"rustfs {version.replace('-', '~')} 1 {arch}")
+                requires = subprocess.check_output(["rpm", "-qp", "--requires", str(rpm)], text=True)
+                if libc == "gnu":
+                    self.assertIn("glibc >= 2.31", requires)
+                else:
+                    self.assertNotIn("glibc", requires)
+
 
 if __name__ == "__main__":
     unittest.main()