Merge branch 'main' of https://github.com/rustfs/rustfs into feat/odm-azure-gcs-sources

# Conflicts:
#	crates/ecstore/src/bucket/on_demand_migration/source_client.rs

.github/pull_request_template.md

@@ -10,16 +10,16 @@ Use N/A when there is no related issue.
 
 ## Summary of Changes
 <!--
-Describe the concrete problem and resulting behavior. For a behavior change, name the input or state that triggers it and the expected outcome. Explain any new dependency or abstraction that the change needs.
+Briefly explain what changed and why reviewers should accept it.
+Focus on behavior, compatibility, and review-relevant context.
 -->
 
 ## Verification
 <!--
-Give 1–3 concrete pieces of evidence for the changed behavior: the test or command, its observed result, and the regression it catches. For a bug fix, record a failing-before/passing-after check or explain why it was unavailable.
+List the commands or checks you ran, for example:
+- `make pre-commit`
 
-Identify the tested commit and any local changes. When testing a prebuilt binary or external service, include its source/version and artifact identity; a successful run against a different build is not evidence for this change.
-
-List relevant checks not run and the remaining risk. Use the validation tier in AGENTS.md; do not run broader checks solely to fill this section. For documentation-only changes, list the applicable documentation checks. Use N/A only when verification is not applicable.
+Use N/A only when verification is not applicable.
 -->
 
 ## Impact

.github/workflows/rustfs-heal-test.yml

@@ -54,25 +54,14 @@ env:
 jobs:
   heal-test:
     runs-on: smoke-testing
+    # Requirement: a failing suite must not fail the workflow; failures
+    # are filed to rustfs/backlog and the chain continues.
+    continue-on-error: true
     timeout-minutes: 480
     # Standalone manual run, or one link of the nightly functional chain
     # (storage -> heal -> pool). Pool expansion no longer re-runs heal.
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-heal-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -128,7 +117,7 @@ jobs:
           else
             ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
           fi
-          ./auto-testing/rustfs_heal_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
+          ./auto-testing/rustfs_heal_test.sh "${ARGS[@]}"
 
       - name: Preflight checks
         run: |
@@ -138,7 +127,7 @@ jobs:
           else
             ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
           fi
-          ./auto-testing/rustfs_heal_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
+          ./auto-testing/rustfs_heal_test.sh "${ARGS[@]}"
 
       - name: Run heal test (write -> outage -> heal -> verify)
         id: test
@@ -148,10 +137,13 @@ jobs:
             --endpoint "${{ env.RUSTFS_API_ENDPOINT }}" \
             --stop-node-gb "${{ inputs.stop_node_gb || '15' }}" \
             --warp-stop-gb "${{ inputs.warp_stop_gb || '40' }}" \
-            --log-file "${LOG_FILE}"
+            --log-file /tmp/rustfs-heal-test.log
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-heal-test.log
+          REPORT_FILE: /tmp/rustfs-heal-report.md
         run: |
           set -euo pipefail
           PACKAGE_URL='${{ inputs.package_url }}'
@@ -160,9 +152,8 @@ jobs:
           else
             PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
-          STEPS_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/steps.md"
-          CASE_RESULT=success
-          python3 - "${LOG_FILE}" "${STEPS_TABLE}" <<'PY' || CASE_RESULT=failure
+          STEPS_TABLE="/tmp/rustfs-heal-steps.md"
+          python3 - "${LOG_FILE}" "${STEPS_TABLE}" <<'PY'
           import re
           import sys
 
@@ -174,7 +165,6 @@ jobs:
 
           steps = {}
           order = []
-          status_rank = {'SKIP': 0, 'PASS': 1, 'FAIL': 2}
           version = None
           version_node = None
           verdict = None
@@ -188,15 +178,14 @@ jobs:
                           n, desc, status = m.group(1), m.group(2), m.group(3)
                           if n not in steps:
                               order.append(n)
-                          if n not in steps or status_rank[status] > status_rank[steps[n][1]]:
-                              steps[n] = (desc, status)
+                          steps[n] = (desc, status)  # later lines win (fail after pass)
                           continue
                       m = ver_re.match(line)
                       if m:
                           version, version_node = m.group(1), m.group(2)
                           continue
                       m = result_re.match(line)
-                      if m and verdict != 'FAIL':
+                      if m:
                           verdict, verdict_detail = m.group(1), m.group(2)
           except FileNotFoundError:
               pass
@@ -216,43 +205,30 @@ jobs:
                   out.write(f'| {n} | {desc} | {status} |\n')
               if not order:
                   out.write('| - | - | NOT RUN (no step result lines found) |\n')
-          complete = set(steps) == {str(n) for n in range(1, 8)}
-          sys.exit(0 if complete and verdict != 'FAIL' and all(status == 'PASS' for _, status in steps.values()) else 1)
           PY
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
           {
             echo "# RustFS heal test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${STEPS_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial step results and suite.log."
-            fi
+            cat "${STEPS_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-heal-report.md
           SUITE: heal
         run: |
           set -euo pipefail
@@ -279,10 +255,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'heal'
           SUITE_LABEL: 'Heal'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-heal-report.md'
+          LOG_FILE: '/tmp/rustfs-heal-test.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -310,16 +287,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -335,12 +310,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload test logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-heal-test-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-heal-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-heal-test*.log
+            /tmp/rustfs-warp.*.log
+          if-no-files-found: warn
 
       - name: Cleanup environment (after)
         if: ${{ always() && inputs.cleanup_after != 'false' }}

.github/workflows/rustfs-kms-test.yml

@@ -49,28 +49,10 @@ env:
 jobs:
   kms-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Checkout repository (for report parser)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-kms-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -127,6 +109,9 @@ jobs:
 
       - name: Run KMS suite
         id: test
+        continue-on-error: true
+        env:
+          LOG_FILE: /tmp/rustfs-kms.log
         run: |
           set -euo pipefail
           chmod +x auto-testing/rustfs-kms-test.sh
@@ -156,7 +141,10 @@ jobs:
           ./auto-testing/rustfs-kms-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-kms.log
+          REPORT_FILE: /tmp/rustfs-kms-report.md
         run: |
           set -euo pipefail
           PACKAGE_URL='${{ inputs.package_url }}'
@@ -168,43 +156,79 @@ jobs:
           else
             PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
-          CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
-          CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" || CASE_RESULT=failure
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
+          CASE_TABLE="/tmp/rustfs-kms-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS KMS test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${CASE_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial case results and suite.log."
-            fi
+            cat "${CASE_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-kms-report.md
           SUITE: kms
         run: |
           set -euo pipefail
@@ -231,10 +255,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'kms'
           SUITE_LABEL: 'KMS'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-kms-report.md'
+          LOG_FILE: '/tmp/rustfs-kms.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -262,16 +287,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -287,12 +310,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-kms-test-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-kms-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-kms.log
+            /tmp/rustfs-kms-report.md
+          if-no-files-found: warn
 
       - name: Cleanup environment (after)
         if: always()

.github/workflows/rustfs-performance-test.yml

@@ -76,33 +76,22 @@ env:
   # Package used by the nightly run (workflow_dispatch inputs are empty for
   # workflow_run events), i.e. the latest nightly deb published by nightly-gnu.yml.
   RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
+  # Fixed benchmark result directory so later steps can read summary.md
+  RUSTFS_RESULT_DIR: /tmp/rustfs-perf-results
   # Cross-repo token for uploading reports to rustfs/dashboard (set in repo settings)
   PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
 
 jobs:
   performance-test:
     runs-on: pf-testing
+    # Requirement: a failing benchmark must not fail the workflow;
+    # failures are filed to rustfs/backlog.
+    continue-on-error: true
     timeout-minutes: 900
     # Run on manual dispatch, or when the nightly build completed successfully.
     # Skipped when nightly failed.
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-performance-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'RUSTFS_RESULT_DIR=%s/results\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'VERSION_FILE=%s/version.txt\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -134,7 +123,7 @@ jobs:
         if: ${{ inputs.cleanup_before != 'false' }}
         run: |
           chmod +x auto-testing/rustfs_performance_test.sh
-          ./auto-testing/rustfs_performance_test.sh --step 1 -y --log-file "${LOG_FILE:-/dev/null}"
+          ./auto-testing/rustfs_performance_test.sh --step 1 -y
 
       - name: Install RustFS package & start cluster (4x4)
         run: |
@@ -144,7 +133,7 @@ jobs:
           else
             ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
           fi
-          ./auto-testing/rustfs_performance_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
+          ./auto-testing/rustfs_performance_test.sh "${ARGS[@]}"
 
       - name: Preflight checks
         run: |
@@ -154,7 +143,7 @@ jobs:
           else
             ARGS+=(--package-url "${{ env.RUSTFS_NIGHTLY_PACKAGE_URL }}")
           fi
-          ./auto-testing/rustfs_performance_test.sh "${ARGS[@]}" --log-file "${LOG_FILE}"
+          ./auto-testing/rustfs_performance_test.sh "${ARGS[@]}"
 
       - name: Run benchmark (GET/PUT/MIXED)
         id: benchmark
@@ -167,15 +156,17 @@ jobs:
             --step 5 -y \
             --warp-duration "${{ inputs.warp_duration || '5m' }}" \
             --warp-concurrency "${{ inputs.warp_concurrency || '64' }}" \
-            --log-file "${LOG_FILE}"
+            --log-file /tmp/rustfs-perf-test.log
 
       - name: Analyze results
         if: ${{ steps.benchmark.conclusion == 'success' }}
         run: |
-          ./auto-testing/rustfs_performance_test.sh --step 6 -y --log-file "${LOG_FILE:-/dev/null}"
+          ./auto-testing/rustfs_performance_test.sh --step 6 -y
 
       - name: Collect RustFS version info
         if: ${{ steps.benchmark.conclusion == 'success' }}
+        env:
+          VERSION_FILE: /tmp/rustfs-version.txt
         run: |
           set -euo pipefail
           read -r -a NODES <<< "${RUSTFS_NODES}"
@@ -195,6 +186,7 @@ jobs:
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
           RESULT_DIR: ${{ env.RUSTFS_RESULT_DIR }}
+          VERSION_FILE: /tmp/rustfs-version.txt
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -202,7 +194,7 @@ jobs:
             exit 0
           fi
           SUMMARY="${RESULT_DIR}/summary.md"
-          [ -s "${SUMMARY}" ] || { echo "summary.md not found at ${SUMMARY}"; exit 1; }
+          [ -f "${SUMMARY}" ] || { echo "summary.md not found at ${SUMMARY}"; exit 1; }
           DATE="$(date -u +%Y-%m-%d)"
           REPORT_PATH="reports/${DATE}.md"
           {
@@ -210,8 +202,6 @@ jobs:
             echo ""
             echo "- **Date**: ${DATE}"
             echo "- **Run**: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- **Attempt**: ${GITHUB_RUN_ATTEMPT}"
-            echo "- **Workflow Commit**: ${GITHUB_SHA}"
             echo "- **Trigger**: ${{ github.event_name }}"
             echo "- **Package**: ${{ inputs.package_url || 'nightly (R2 latest)' }}"
             echo ""
@@ -221,8 +211,8 @@ jobs:
             echo '```text'
             cat "${VERSION_FILE}"
             echo '```'
-          } > "${REPORT_FILE}"
-          CONTENT="$(python3 -c 'import base64,sys; print(base64.b64encode(open(sys.argv[1],"rb").read()).decode())' "${REPORT_FILE}")"
+          } > /tmp/rustfs-perf-report.md
+          CONTENT="$(python3 -c 'import base64; print(base64.b64encode(open("/tmp/rustfs-perf-report.md","rb").read()).decode())')"
           SHA="$(gh api "repos/rustfs/dashboard/contents/${REPORT_PATH}" -q '.sha' 2>/dev/null || true)"
           if [ -n "${SHA}" ]; then
             jq -n --arg msg "report: ${DATE}" --arg content "${CONTENT}" --arg sha "${SHA}" \
@@ -241,10 +231,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'performance'
           SUITE_LABEL: 'Performance'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-perf-report.md'
+          LOG_FILE: '/tmp/rustfs-perf-test.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -272,16 +263,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -297,17 +286,20 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload test logs & results
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-perf-test-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-perf-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-perf-test*.log
+            /tmp/rustfs-perf-results/**
+            /tmp/rustfs-version.txt
+          if-no-files-found: warn
 
       - name: Reset test environment (after)
         if: ${{ always() && inputs.cleanup_after != 'false' }}
         run: |
-          ./auto-testing/rustfs_performance_test.sh --step 7 -y --log-file "${LOG_FILE:-/dev/null}"
+          ./auto-testing/rustfs_performance_test.sh --step 7 -y
 
       - name: Notify on failure
         if: failure()

.github/workflows/rustfs-pool-expand-test.yml

@@ -76,6 +76,9 @@ jobs:
   pool-expansion-test:
     name: Pool expansion / decommission test
     runs-on: smoke-testing
+    # Requirement: a failing suite must not fail the workflow; failures
+    # are filed to rustfs/backlog and the chain continues.
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     env:

.github/workflows/rustfs-replication-test.yml

@@ -62,28 +62,12 @@ env:
 jobs:
   replication-test:
     runs-on: smoke-testing
+    # A failed replication run must not break the chain or the workflow: the
+    # failure is reported to rustfs/backlog instead (see the issue step).
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Checkout repository (for report parser)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-replication-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -132,6 +116,9 @@ jobs:
 
       - name: Run replication suite
         id: test
+        continue-on-error: true
+        env:
+          LOG_FILE: /tmp/rustfs-replication.log
         run: |
           set -euo pipefail
           chmod +x auto-testing/rustfs-replication-test.sh
@@ -154,7 +141,10 @@ jobs:
           ./auto-testing/rustfs-replication-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-replication.log
+          REPORT_FILE: /tmp/rustfs-replication-report.md
         run: |
           set -euo pipefail
           PACKAGE_URL='${{ inputs.package_url }}'
@@ -176,44 +166,80 @@ jobs:
               RUSTFS_VERSION_INFO="${DETECTED_VERSION}"
             fi
           fi
-          CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
-          CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" || CASE_RESULT=failure
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
+          CASE_TABLE="/tmp/rustfs-replication-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z0-9]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z0-9]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          try:
+              with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+                  for raw in fh:
+                      line = ansi.sub('', raw).strip()
+                      m = start_re.match(line)
+                      if m:
+                          case_id, name = m.group(1), m.group(2)
+                          if case_id not in index:
+                              index[case_id] = len(rows)
+                              rows.append([case_id, name, 'RUNNING'])
+                          continue
+                      m = done_re.match(line)
+                      if m:
+                          status, case_id = m.group(1), m.group(2)
+                          if case_id in index:
+                              rows[index[case_id]][2] = status
+                          else:
+                              rows.append([case_id, case_id, status])
+                              index[case_id] = len(rows) - 1
+          except FileNotFoundError:
+              rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+              counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+              out.write('## Case Summary\n\n')
+              out.write(f"- Total: {len(rows)}\\n")
+              out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+              out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+              out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+              out.write('\\n')
+              out.write('| Case | Name | Status |\\n')
+              out.write('| --- | --- | --- |\\n')
+              for case_id, name, status in rows:
+                  out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS replication test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
             echo "- RustFS Version: ${RUSTFS_VERSION_INFO}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${CASE_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial case results and suite.log."
-            fi
+            cat "${CASE_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-replication-report.md
           SUITE: replication
         run: |
           set -euo pipefail
@@ -240,10 +266,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'replication'
           SUITE_LABEL: 'Replication (bucket + site)'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-replication-report.md'
+          LOG_FILE: '/tmp/rustfs-replication.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -271,16 +298,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -296,12 +321,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-replication-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-replication-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-replication.log
+            /tmp/rustfs-replication-report.md
+          if-no-files-found: warn
 
       - name: Cleanup environment (after)
         if: always()

.github/workflows/rustfs-s3-compat-test.yml

@@ -37,28 +37,10 @@ env:
 jobs:
   s3-compat-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Checkout repository (for report parser)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-s3-compat-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -106,6 +88,9 @@ jobs:
 
       - name: Run S3 compatibility suite
         id: test
+        continue-on-error: true
+        env:
+          LOG_FILE: /tmp/rustfs-s3-compat.log
         run: |
           set -euo pipefail
           chmod +x auto-testing/rustfs-s3-compat-test.sh
@@ -122,7 +107,10 @@ jobs:
           ./auto-testing/rustfs-s3-compat-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-s3-compat.log
+          REPORT_FILE: /tmp/rustfs-s3-compat-report.md
         run: |
           set -euo pipefail
           PACKAGE_URL='${{ inputs.package_url }}'
@@ -144,44 +132,83 @@ jobs:
               RUSTFS_VERSION_INFO="${DETECTED_VERSION}"
             fi
           fi
-          CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
-          CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" || CASE_RESULT=failure
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
+          CASE_TABLE="/tmp/rustfs-s3-compat-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z0-9]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z0-9]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          current = None
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  current = case_id
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+                  current = None
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS S3 compatibility test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
             echo "- RustFS Version: ${RUSTFS_VERSION_INFO}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${CASE_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial case results and suite.log."
-            fi
+            cat "${CASE_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-s3-compat-report.md
           SUITE: s3
         run: |
           set -euo pipefail
@@ -208,10 +235,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 's3'
           SUITE_LABEL: 'S3 compatibility'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-s3-compat-report.md'
+          LOG_FILE: '/tmp/rustfs-s3-compat.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -239,16 +267,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -264,12 +290,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-s3-compat-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-s3-compat-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-s3-compat.log
+            /tmp/rustfs-s3-compat-report.md
+          if-no-files-found: warn
 
       - name: Cleanup environment (after)
         if: always()

.github/workflows/rustfs-storage-test.yml

@@ -46,28 +46,10 @@ env:
 jobs:
   storage-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 360
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Checkout repository (for report parser)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-storage-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -115,6 +97,9 @@ jobs:
 
       - name: Run storage engine suite
         id: test
+        continue-on-error: true
+        env:
+          LOG_FILE: /tmp/rustfs-storage.log
         run: |
           set -euo pipefail
           chmod +x auto-testing/rustfs-storage-test.sh
@@ -137,7 +122,10 @@ jobs:
           ./auto-testing/rustfs-storage-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-storage.log
+          REPORT_FILE: /tmp/rustfs-storage-report.md
         run: |
           set -euo pipefail
           PACKAGE_URL='${{ inputs.package_url }}'
@@ -159,44 +147,83 @@ jobs:
               RUSTFS_VERSION_INFO="${DETECTED_VERSION}"
             fi
           fi
-          CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
-          CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" || CASE_RESULT=failure
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
+          CASE_TABLE="/tmp/rustfs-storage-cases.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file = sys.argv[1], sys.argv[2]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z0-9]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z0-9]+-[0-9]+)\b')
+
+          rows = []
+          index = {}
+          current = None
+          try:
+            with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+              for raw in fh:
+                line = ansi.sub('', raw).strip()
+                m = start_re.match(line)
+                if m:
+                  case_id, name = m.group(1), m.group(2)
+                  current = case_id
+                  if case_id not in index:
+                    index[case_id] = len(rows)
+                    rows.append([case_id, name, 'RUNNING'])
+                  continue
+                m = done_re.match(line)
+                if m:
+                  status, case_id = m.group(1), m.group(2)
+                  if case_id in index:
+                    rows[index[case_id]][2] = status
+                  else:
+                    rows.append([case_id, case_id, status])
+                    index[case_id] = len(rows) - 1
+                  current = None
+          except FileNotFoundError:
+            rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+            counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+            out.write('## Case Summary\n\n')
+            out.write(f"- Total: {len(rows)}\\n")
+            out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+            out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+            out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+            out.write('\\n')
+            out.write('| Case | Name | Status |\\n')
+            out.write('| --- | --- | --- |\\n')
+            for case_id, name, status in rows:
+              out.write(f'| {case_id} | {name} | {status} |\\n')
+          PY
           {
             echo "# RustFS storage engine test report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- Package: ${PACKAGE_SOURCE}"
             echo "- RustFS Version: ${RUSTFS_VERSION_INFO}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${CASE_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial case results and suite.log."
-            fi
+            cat "${CASE_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-storage-report.md
           SUITE: storage
         run: |
           set -euo pipefail
@@ -223,10 +250,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'storage'
           SUITE_LABEL: 'Storage engine'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-storage-report.md'
+          LOG_FILE: '/tmp/rustfs-storage.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -254,16 +282,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -279,12 +305,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-storage-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-storage-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-storage.log
+            /tmp/rustfs-storage-report.md
+          if-no-files-found: warn
 
       - name: Cleanup environment (after)
         if: always()

.github/workflows/rustfs-tier-test.yml

@@ -61,6 +61,9 @@ env:
 jobs:
   tier-test:
     runs-on: smoke-testing
+    # Requirement: a failing suite must not fail the workflow; failures
+    # are filed to rustfs/backlog and the chain continues.
+    continue-on-error: true
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:

.github/workflows/rustfs-upgrade-test.yml

@@ -79,28 +79,10 @@ env:
 jobs:
   upgrade-test:
     runs-on: smoke-testing
+    continue-on-error: true
     timeout-minutes: 420
     if: ${{ github.event_name == 'workflow_dispatch' || github.event_name == 'repository_dispatch' }}
     steps:
-      - name: Checkout repository (for report parser)
-        uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7
-        with:
-          persist-credentials: false
-
-      - name: Initialize functional evidence
-        id: evidence
-        run: |
-          set -euo pipefail
-          umask 077
-          FUNCTIONAL_ARTIFACTS_DIR="${RUNNER_TEMP}/rustfs-upgrade-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
-          mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}"
-          {
-            printf 'FUNCTIONAL_ARTIFACTS_DIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'LOG_FILE=%s/suite.log\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'REPORT_FILE=%s/report.md\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-            printf 'TMPDIR=%s\n' "${FUNCTIONAL_ARTIFACTS_DIR}"
-          } >> "${GITHUB_ENV}"
-
       # auto-testing is private: clone it with the dedicated PF token (not
       # GITHUB_TOKEN) and retry transient GitHub/network failures.
       - name: Checkout auto-testing scripts (with retry)
@@ -160,7 +142,9 @@ jobs:
 
       - name: Run upgrade compatibility suite
         id: test
+        continue-on-error: true
         env:
+          LOG_FILE: /tmp/rustfs-upgrade.log
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
         run: |
           set -euo pipefail
@@ -218,7 +202,10 @@ jobs:
           ./auto-testing/rustfs-upgrade-test.sh "${ARGS[@]}"
 
       - name: Generate report
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
+        env:
+          LOG_FILE: /tmp/rustfs-upgrade.log
+          REPORT_FILE: /tmp/rustfs-upgrade-report.md
         run: |
           set -euo pipefail
           FROM_URL='${{ inputs.from_url }}'
@@ -239,47 +226,103 @@ jobs:
           else
             TO_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
           fi
-          CASE_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/cases.md"
-          MATRIX_TABLE="${FUNCTIONAL_ARTIFACTS_DIR}/matrix.md"
-          CASE_RESULT=success
-          python3 scripts/functional_case_report.py "${LOG_FILE}" "${CASE_TABLE}" "${MATRIX_TABLE}" || CASE_RESULT=failure
-          RESULT=failure
-          if [ '${{ steps.test.outcome }}' = 'success' ] && [ "${CASE_RESULT}" = 'success' ]; then
-            RESULT=success
-          fi
+          CASE_TABLE="/tmp/rustfs-upgrade-cases.md"
+          MATRIX_TABLE="/tmp/rustfs-upgrade-matrix.md"
+          python3 - "${LOG_FILE}" "${CASE_TABLE}" "${MATRIX_TABLE}" <<'PY'
+          import re
+          import sys
+
+          log_file, out_file, matrix_file = sys.argv[1], sys.argv[2], sys.argv[3]
+          ansi = re.compile(r'\x1b\[[0-9;]*m')
+          start_re = re.compile(r'^---\s+([A-Z]+-[0-9]+)\s+(.+?)\s+---$')
+          done_re = re.compile(r'^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z]+-[0-9]+)\b')
+          topo_re = re.compile(
+              r'^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$')
+
+          rows = []
+          index = {}
+          topo_rows = []
+          try:
+              with open(log_file, 'r', encoding='utf-8', errors='replace') as fh:
+                  for raw in fh:
+                      line = ansi.sub('', raw).strip()
+                      m = topo_re.match(line)
+                      if m:
+                          topo_rows.append(m.groups())
+                          continue
+                      m = start_re.match(line)
+                      if m:
+                          case_id, name = m.group(1), m.group(2)
+                          if case_id not in index:
+                              index[case_id] = len(rows)
+                              rows.append([case_id, name, 'RUNNING'])
+                          continue
+                      m = done_re.match(line)
+                      if m:
+                          status, case_id = m.group(1), m.group(2)
+                          if case_id in index:
+                              rows[index[case_id]][2] = status
+                          else:
+                              rows.append([case_id, case_id, status])
+                              index[case_id] = len(rows) - 1
+          except FileNotFoundError:
+              rows = []
+
+          counts = {'PASS': 0, 'FAIL': 0, 'UNSUPPORTED': 0, 'RUNNING': 0}
+          for _, _, status in rows:
+              counts[status] = counts.get(status, 0) + 1
+
+          with open(out_file, 'w', encoding='utf-8') as out:
+              out.write('## Case Summary\n\n')
+              out.write(f"- Total: {len(rows)}\\n")
+              out.write(f"- PASS: {counts.get('PASS', 0)}\\n")
+              out.write(f"- FAIL: {counts.get('FAIL', 0)}\\n")
+              out.write(f"- UNSUPPORTED: {counts.get('UNSUPPORTED', 0)}\\n")
+              out.write('\\n')
+              out.write('| Case | Name | Status |\\n')
+              out.write('| --- | --- | --- |\\n')
+              for case_id, name, status in rows:
+                  out.write(f'| {case_id} | {name} | {status} |\\n')
+
+          # Upgrade matrix: one row per topology/backend with the versions
+          # captured on the nodes (rustfs --version) and the aggregated
+          # result. The dashboard renders this table directly.
+          with open(matrix_file, 'w', encoding='utf-8') as out:
+              out.write('## Upgrade Matrix\n\n')
+              out.write('| Topology | KMS Backend | From Version | To Version | Result |\n')
+              out.write('| --- | --- | --- | --- | --- |\n')
+              for topo, backend, old_v, new_v, npass, nfail in topo_rows:
+                  result = 'PASS' if nfail == '0' else 'FAIL'
+                  out.write(f'| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) |\n')
+              if not topo_rows:
+                  out.write('| - | - | - | - | NOT RUN (suite failed before upgrade) |\n')
+          PY
           {
             echo "# RustFS upgrade compatibility report"
             echo ""
             echo "- Run: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${{ github.event_name }}"
             echo "- From: ${FROM_SOURCE}"
             echo "- To: ${TO_SOURCE}"
-            echo "- Test Step Outcome: ${RESULT}"
-            echo "- Suite Step Outcome: ${{ steps.test.outcome }}"
+            echo "- Test Step Outcome: ${{ steps.test.outcome }}"
             echo ""
-            if [ "${RESULT}" = "success" ]; then
-              cat "${MATRIX_TABLE}"
-              echo ""
-              cat "${CASE_TABLE}"
-              echo ""
-              echo "## Log tail"
-              echo '```text'
-              tail -n 200 "${LOG_FILE}"
-              echo '```'
-            else
-              echo "The suite or evidence validation failed. See this run's artifact for partial case results and suite.log."
-            fi
+            cat "${MATRIX_TABLE}" || true
+            echo ""
+            cat "${CASE_TABLE}" || true
+            echo ""
+            echo "## Log tail"
+            echo '```text'
+            tail -n 200 "${LOG_FILE}" || true
+            echo '```'
           } | tee "${REPORT_FILE}"
           cat "${REPORT_FILE}" >> "${GITHUB_STEP_SUMMARY}"
-          [ "${RESULT}" = "success" ]
 
       - name: Upload functional report to dashboard
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         continue-on-error: true
         env:
           GH_TOKEN: ${{ env.PF_TESTING_GH_TOKEN }}
+          REPORT_FILE: /tmp/rustfs-upgrade-report.md
           SUITE: upgrade
         run: |
           set -euo pipefail
@@ -306,10 +349,11 @@ jobs:
         continue-on-error: true
         env:
           GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
-          EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
           SUITE: 'upgrade'
           SUITE_LABEL: 'Upgrade compatibility'
           RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
+          REPORT_FILE: '/tmp/rustfs-upgrade-report.md'
+          LOG_FILE: '/tmp/rustfs-upgrade.log'
         run: |
           set -euo pipefail
           if [ -z "${GH_TOKEN:-}" ]; then
@@ -337,16 +381,14 @@ jobs:
             echo ""
             echo "- Suite: \`${SUITE}\`"
             echo "- Run: ${RUN_URL}"
-            echo "- Attempt: ${GITHUB_RUN_ATTEMPT}"
-            echo "- Workflow Commit: ${GITHUB_SHA}"
             echo "- Trigger: ${GITHUB_EVENT_NAME}"
             echo "- Date: $(date -u +%Y-%m-%d)"
             echo ""
             echo "## Report (errors and symptoms)"
             echo ""
-            if [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${REPORT_FILE}" ]; then
+            if [ -s "${REPORT_FILE}" ]; then
               redact < "${REPORT_FILE}"
-            elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
+            elif [ -s "${LOG_FILE:-}" ]; then
               echo "(report file missing; log tail below)"
               echo ""
               tail -n 200 "${LOG_FILE}" | redact
@@ -362,12 +404,14 @@ jobs:
           echo "filed backlog issue for suite ${SUITE}"
 
       - name: Upload report and logs
-        if: ${{ always() && steps.evidence.outcome == 'success' }}
+        if: always()
         uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
         with:
-          name: rustfs-upgrade-test-${{ github.run_id }}-${{ github.run_attempt }}
-          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/
-          if-no-files-found: error
+          name: rustfs-upgrade-test-${{ github.run_id }}
+          path: |
+            /tmp/rustfs-upgrade-report.md
+            /tmp/rustfs-upgrade.*/*
+          if-no-files-found: ignore
           retention-days: 3
 
       - name: Cleanup environment (after)

Cargo.lock

@@ -9792,6 +9792,7 @@ dependencies = [
  "path-absolutize",
  "pin-project-lite",
  "proptest",
+ "quick-xml",
  "rand 0.10.2",
  "ratelimit",
  "rcgen",

crates/ecstore/Cargo.toml

@@ -215,6 +215,7 @@ serde_urlencoded.workspace = true
 google-cloud-storage = { workspace = true }
 google-cloud-auth = { workspace = true }
 faster-hex = { workspace = true }
+quick-xml = { workspace = true }
 ratelimit = { workspace = true }
 aws-smithy-http-client = { workspace = true, default-features = false, features = ["rustls-aws-lc"] }
 

crates/ecstore/src/api/mod.rs

@@ -153,12 +153,13 @@ pub mod bucket {
             LastSourceError, LatencyBucketSnapshot, NEGATIVE_CACHE_MAX_ENTRIES, NegativeCache, OdmBucketSnapshot, OdmLookup,
             OdmOp, OdmOutcome, OdmStateError, OdmStats, OdmStatsSnapshot, OnDemandMigrationSys, PullError, PullFailureReason,
             PullFollower, PullLeader, PullOutcome, PullPath, PullResult, PullSlot, SOURCE_LATENCY_BUCKET_BOUNDS_MS,
-            SourceLatencySnapshot, source_client_spec,
+            SourceLatencySnapshot, source_backend_spec, source_client_spec,
         };
         pub use crate::bucket::on_demand_migration::{
-            ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
-            OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy,
-            SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+            AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
+            ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig,
+            Provider, RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig,
+            ValidationContext,
         };
         pub use crate::bucket::on_demand_migration::{
             EnqueueOutcome, LocalObject, MAX_MULTIPART_PARTS, OdmWriteBack, PULL_MAX_RETRIES, PULL_RETRY_BASE_DELAYS,
@@ -184,9 +185,9 @@ pub mod bucket {
         }
         pub mod source_client {
             pub use crate::bucket::on_demand_migration::source_client::{
-                SourceClient, SourceClientSpec, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
-                SourceProbe, SourceProvider, SourceSse, SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value,
-                resolve_path_style,
+                AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError,
+                SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage, SourceProbe, SourceProvider, SourceSse,
+                SourceTimeouts, USER_AGENT_SUFFIX, is_multipart_etag, range_header_value, resolve_path_style,
             };
         }
     }

crates/ecstore/src/bucket/on_demand_migration/backend_contract.rs

@@ -0,0 +1,172 @@
+// 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.
+
+//! One contract every [`SourceBackend`] implementation must satisfy.
+//!
+//! The migration pipeline talks to a source only through the trait, so a new
+//! provider is correct exactly when it answers the same questions the same way:
+//! the same head fields, the same range semantics, the same page shape, the
+//! same error classes. Each backend supplies a fixture that answers this fixed
+//! corpus in its own dialect and then runs [`assert_backend_contract`], so a
+//! provider-specific mapping bug shows up as a contract failure rather than as
+//! a surprise in the pull pipeline.
+//!
+//! Backends differ in two documented ways, declared through
+//! [`BackendCapabilities`]: whether the provider's ETag is a content digest,
+//! and whether the provider can resume a listing from a key.
+
+use super::source_client::{SourceBackend, SourceError, SourceListRequest};
+use crate::storage_api_contracts::range::HTTPRangeSpec;
+use std::collections::HashMap;
+
+/// The single object every fixture serves.
+pub(super) const OBJECT_KEY: &str = "dir/a.txt";
+pub(super) const OBJECT_BODY: &[u8] = b"hello";
+/// MD5 of [`OBJECT_BODY`]; the ETag of the object on a digest provider.
+pub(super) const OBJECT_MD5: &str = "5d41402abc4b2a76b9719d911017c592";
+/// The second key the fixture's listing returns, on its second page.
+pub(super) const SECOND_KEY: &str = "dir/b.txt";
+pub(super) const COMMON_PREFIX: &str = "dir/sub/";
+pub(super) const LIST_CURSOR: &str = "cursor-1";
+/// A key the fixture answers with the provider's "no such object".
+pub(super) const MISSING_KEY: &str = "missing";
+/// A key the fixture answers with the provider's "not authorized".
+pub(super) const FORBIDDEN_KEY: &str = "secret";
+
+/// Where backends are allowed to differ.
+#[derive(Clone, Copy, Debug)]
+pub(super) struct BackendCapabilities {
+    /// The provider's ETag is an opaque token, not a digest of the bytes.
+    pub(super) etag_is_opaque: bool,
+    /// The provider can resume a listing from a key rather than only from an
+    /// opaque cursor.
+    pub(super) supports_start_after: bool,
+    /// The provider has an object-tagging concept at all. GCS does not, and
+    /// answers with an empty map instead of failing a pull.
+    pub(super) supports_tagging: bool,
+}
+
+/// Drives `backend` through the shared corpus. Fixtures are scripted in
+/// request order, so the call order here is part of the contract.
+pub(super) async fn assert_backend_contract(backend: &dyn SourceBackend, caps: BackendCapabilities) {
+    // 1. HEAD maps the object's shared fields.
+    let head = backend.head(OBJECT_KEY).await.expect("HEAD of the fixture object");
+    assert_eq!(head.size, OBJECT_BODY.len() as u64, "HEAD reports the object size");
+    assert_eq!(head.content_type.as_deref(), Some("text/plain"));
+    assert_eq!(
+        head.user_metadata,
+        HashMap::from([("owner".to_string(), "alice".to_string())]),
+        "user metadata is keyed without the provider prefix"
+    );
+    assert!(head.storage_class.is_some(), "the provider's tier is recorded");
+    assert!(head.last_modified.is_some(), "the provider's timestamp is parsed");
+    assert!(head.sse.is_none(), "the fixture object is not server-side encrypted");
+    assert!(!head.is_multipart_etag);
+    assert_eq!(head.etag_is_opaque, caps.etag_is_opaque);
+    match caps.etag_is_opaque {
+        false => assert_eq!(head.etag.as_deref(), Some(OBJECT_MD5), "a digest ETag is mapped verbatim"),
+        true => assert!(head.etag.is_some(), "an opaque ETag is still recorded"),
+    }
+
+    // 2. An unranged GET streams the whole object and reports no range.
+    let got = backend.get(OBJECT_KEY, None).await.expect("unranged GET");
+    assert_eq!(got.head.size, OBJECT_BODY.len() as u64);
+    assert!(got.content_range.is_none(), "an unranged GET has no content-range");
+    assert_eq!(got.head.etag_is_opaque, caps.etag_is_opaque, "GET and HEAD agree about the ETag");
+    let body = got.body.collect().await.expect("body streams").into_bytes();
+    assert_eq!(body.as_ref(), OBJECT_BODY);
+
+    // 3. A ranged GET returns exactly the requested interval, and `size` is
+    //    the length of the returned bytes rather than of the object.
+    let range = HTTPRangeSpec {
+        is_suffix_length: false,
+        start: 1,
+        end: 3,
+    };
+    let got = backend.get(OBJECT_KEY, Some(&range)).await.expect("ranged GET");
+    assert_eq!(got.head.size, 3, "a ranged GET reports the range length");
+    assert_eq!(got.content_range.as_deref(), Some("bytes 1-3/5"));
+    let body = got.body.collect().await.expect("body streams").into_bytes();
+    assert_eq!(body.as_ref(), &OBJECT_BODY[1..=3]);
+
+    // 4. A delimiter listing rolls prefixes up and hands back a cursor.
+    let page = backend
+        .list(&SourceListRequest {
+            prefix: Some("dir/"),
+            delimiter: Some("/"),
+            max_keys: 2,
+            ..Default::default()
+        })
+        .await
+        .expect("first listing page");
+    assert_eq!(page.objects.len(), 1, "the first page holds one object");
+    assert_eq!(page.objects[0].key, OBJECT_KEY, "listing keys are in the source namespace");
+    assert_eq!(page.objects[0].size, OBJECT_BODY.len() as u64);
+    assert!(page.objects[0].last_modified.is_some());
+    assert_eq!(page.common_prefixes, vec![COMMON_PREFIX.to_string()]);
+    assert!(page.is_truncated);
+    assert_eq!(page.next_continuation_token.as_deref(), Some(LIST_CURSOR));
+
+    // 5. The cursor is passed back verbatim and the last page ends the walk.
+    let page = backend
+        .list(&SourceListRequest {
+            prefix: Some("dir/"),
+            delimiter: Some("/"),
+            continuation_token: Some(LIST_CURSOR),
+            max_keys: 2,
+            ..Default::default()
+        })
+        .await
+        .expect("second listing page");
+    assert_eq!(page.objects.len(), 1);
+    assert_eq!(page.objects[0].key, SECOND_KEY);
+    assert!(!page.is_truncated);
+    assert!(page.next_continuation_token.is_none(), "a complete listing carries no cursor");
+
+    // 6. Tags come back as a flat map, empty on a provider without tags.
+    let tags = backend.tagging(OBJECT_KEY).await.expect("object tags");
+    match caps.supports_tagging {
+        true => assert_eq!(tags, HashMap::from([("env".to_string(), "prod".to_string())])),
+        false => assert!(tags.is_empty(), "a provider without tags reports none: {tags:?}"),
+    }
+
+    // 7. The probe confirms the bucket or container answers.
+    backend.probe().await.expect("probe of the fixture bucket");
+
+    // 8. A missing object is `NotFound`, and never retried.
+    let err = backend.head(MISSING_KEY).await.expect_err("a missing object must fail");
+    assert!(matches!(err, SourceError::NotFound), "{err:?}");
+    assert_eq!(err.class_label(), "not_found");
+    assert!(!err.is_retryable());
+
+    // 9. A denied object is `AccessDenied`, and never retried.
+    let err = backend.head(FORBIDDEN_KEY).await.expect_err("a denied object must fail");
+    assert!(matches!(err, SourceError::AccessDenied), "{err:?}");
+    assert_eq!(err.class_label(), "access_denied");
+    assert!(!err.is_retryable());
+
+    // 10. A provider without a key cursor must refuse one instead of listing
+    //     from the wrong position. This issues no request either way.
+    if !caps.supports_start_after {
+        let err = backend
+            .list(&SourceListRequest {
+                start_after: Some(OBJECT_KEY),
+                max_keys: 1,
+                ..Default::default()
+            })
+            .await
+            .expect_err("a backend without a key cursor must refuse start_after");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/config.rs

@@ -30,6 +30,10 @@ pub const ON_DEMAND_MIGRATION_CONFIG_VERSION: u32 = 1;
 const REDACTED: &str = "REDACTED";
 const AUTO_REGION: &str = "auto";
 const AUTO_REGION_FALLBACK: &str = "us-east-1";
+/// Public Azure Blob host suffix; the account name is the first label.
+pub const AZURE_BLOB_SUFFIX: &str = "blob.core.windows.net";
+/// Public Google Cloud Storage endpoint for the native provider.
+pub const GCS_DEFAULT_ENDPOINT: &str = "https://storage.googleapis.com";
 
 const KIB: u64 = 1024;
 const MIB: u64 = 1024 * KIB;
@@ -75,14 +79,25 @@ pub struct SourceConfig {
     pub bucket: String,
     #[serde(default)]
     pub path_style: PathStyle,
-    /// `None` means anonymous access to a public source bucket.
+    /// `None` means anonymous access to a public source bucket. Only the
+    /// SigV4 providers read it; `azure` and `gcs_native` carry their own
+    /// credentials in `azure` / `gcs`.
     #[serde(default)]
     pub credentials: Option<SourceCredentials>,
     #[serde(default)]
     pub tls: TlsConfig,
+    /// Required for [`Provider::Azure`] and rejected for every other
+    /// provider.
+    #[serde(default)]
+    pub azure: Option<AzureSourceConfig>,
+    /// Required for [`Provider::GcsNative`] and rejected for every other
+    /// provider. [`Provider::Gcs`] keeps using `credentials` because it
+    /// speaks the S3 interoperability API.
+    #[serde(default)]
+    pub gcs: Option<GcsSourceConfig>,
 }
 
-/// Source vendor family. `azure` is deliberately absent from this version.
+/// Source vendor family.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
 #[serde(rename_all = "lowercase")]
 pub enum Provider {
@@ -94,6 +109,12 @@ pub enum Provider {
     R2,
     /// GCS XML interoperability API with HMAC keys.
     Gcs,
+    /// Native Azure Blob service; parameters in `source.azure`.
+    Azure,
+    /// Native GCS JSON API with a service-account key; parameters in
+    /// `source.gcs`.
+    #[serde(rename = "gcs_native")]
+    GcsNative,
 }
 
 impl Provider {
@@ -105,13 +126,22 @@ impl Provider {
             Provider::Rustfs => "rustfs",
             Provider::R2 => "r2",
             Provider::Gcs => "gcs",
+            Provider::Azure => "azure",
+            Provider::GcsNative => "gcs_native",
         }
     }
 
+    /// Providers that do not speak S3 and therefore ignore `region`,
+    /// `path_style` and `credentials`.
+    pub fn is_native(&self) -> bool {
+        matches!(self, Provider::Azure | Provider::GcsNative)
+    }
+
     /// Providers whose SDKs accept `region = "auto"`; RustFS maps it to
-    /// `us-east-1` for signing.
+    /// `us-east-1` for signing. The native providers never sign with a
+    /// region, so they accept it as well.
     fn accepts_auto_region(&self) -> bool {
-        matches!(self, Provider::R2 | Provider::Minio | Provider::Rustfs)
+        matches!(self, Provider::R2 | Provider::Minio | Provider::Rustfs) || self.is_native()
     }
 }
 
@@ -164,6 +194,73 @@ impl fmt::Debug for SourceCredentials {
     }
 }
 
+/// Native Azure Blob source parameters. The container is `source.bucket`,
+/// so a config never carries two names for the same container. Exactly one
+/// of `account_key` and `sas_token` must be set: the account key signs with
+/// Shared Key, the SAS token is appended to every request URL.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct AzureSourceConfig {
+    /// Storage account name; also derives the default `blob.core.windows.net`
+    /// endpoint when `source.endpoint` is absent.
+    pub account: String,
+    /// Base64 shared key of the storage account.
+    #[serde(default)]
+    pub account_key: Option<String>,
+    /// SAS query string without the leading `?`.
+    #[serde(default)]
+    pub sas_token: Option<String>,
+}
+
+impl AzureSourceConfig {
+    /// A copy safe to return to admin clients or log: both secrets are
+    /// replaced by `REDACTED`, and whether each is set stays visible.
+    pub fn redacted(&self) -> Self {
+        Self {
+            account: self.account.clone(),
+            account_key: self.account_key.as_ref().map(|_| REDACTED.to_string()),
+            sas_token: self.sas_token.as_ref().map(|_| REDACTED.to_string()),
+        }
+    }
+}
+
+impl fmt::Debug for AzureSourceConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("AzureSourceConfig")
+            .field("account", &self.account)
+            .field("account_key", &self.account_key.as_ref().map(|_| REDACTED))
+            .field("sas_token", &self.sas_token.as_ref().map(|_| REDACTED))
+            .finish()
+    }
+}
+
+/// Native Google Cloud Storage source parameters. The bucket is
+/// `source.bucket`; only the service-account key lives here.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct GcsSourceConfig {
+    /// Service-account key JSON, verbatim as downloaded from Google Cloud.
+    pub service_account_json: String,
+}
+
+impl GcsSourceConfig {
+    /// A copy safe to return to admin clients or log: the whole key JSON is
+    /// a secret (it embeds the private key), so it is replaced wholesale.
+    pub fn redacted(&self) -> Self {
+        Self {
+            service_account_json: REDACTED.to_string(),
+        }
+    }
+}
+
+impl fmt::Debug for GcsSourceConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GcsSourceConfig")
+            .field("service_account_json", &REDACTED)
+            .finish()
+    }
+}
+
 #[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
 #[serde(deny_unknown_fields)]
 pub struct TlsConfig {
@@ -354,6 +451,14 @@ pub enum OnDemandMigrationConfigError {
     InvalidBucket(&'static str),
     #[error("source credentials field {0} must not be empty")]
     EmptyCredential(&'static str),
+    #[error("source.{0} is required for provider {1}")]
+    MissingProviderBlock(&'static str, Provider),
+    #[error("source.{0} is not valid for provider {1}")]
+    UnexpectedProviderBlock(&'static str, Provider),
+    /// Carries only the reason: the block holds account keys, SAS tokens and
+    /// service-account JSON, so no value of it is ever echoed.
+    #[error("source.{0} is invalid: {1}")]
+    InvalidProviderBlock(&'static str, &'static str),
     #[error("source tls.ca_cert_pem is not a PEM certificate")]
     InvalidCaCert,
     #[error("filter.{0} must be null or a non-empty string")]
@@ -388,6 +493,8 @@ impl OnDemandMigrationConfig {
     pub fn redacted(&self) -> Self {
         let mut copy = self.clone();
         copy.source.credentials = self.source.credentials.as_ref().map(SourceCredentials::redacted);
+        copy.source.azure = self.source.azure.as_ref().map(AzureSourceConfig::redacted);
+        copy.source.gcs = self.source.gcs.as_ref().map(GcsSourceConfig::redacted);
         copy
     }
 
@@ -433,6 +540,12 @@ impl SourceConfig {
         match (&self.endpoint, self.provider) {
             (Some(endpoint), _) => endpoint.clone(),
             (None, Provider::Aws) => format!("https://s3.{}.amazonaws.com", self.region),
+            (None, Provider::Azure) => self
+                .azure
+                .as_ref()
+                .map(|azure| format!("https://{}.{AZURE_BLOB_SUFFIX}", azure.account))
+                .unwrap_or_default(),
+            (None, Provider::GcsNative) => GCS_DEFAULT_ENDPOINT.to_string(),
             (None, _) => String::new(),
         }
     }
@@ -448,6 +561,8 @@ impl SourceConfig {
     }
 
     fn validate(&self) -> Result<(), OnDemandMigrationConfigError> {
+        self.validate_provider_block()?;
+
         if self.region.is_empty() {
             return Err(OnDemandMigrationConfigError::EmptyRegion);
         }
@@ -466,6 +581,9 @@ impl SourceConfig {
                     ));
                 }
             }
+            // Both native providers derive a fixed endpoint; Azure's is built
+            // from the account name, already checked by `validate_provider_block`.
+            None if self.provider.is_native() => {}
             None => return Err(OnDemandMigrationConfigError::MissingEndpoint(self.provider)),
         }
 
@@ -496,6 +614,84 @@ impl SourceConfig {
 
         Ok(())
     }
+
+    /// The provider-specific block must be present for exactly its own
+    /// provider: a stray `azure` block on an `s3` source would otherwise be
+    /// accepted, stored, and silently ignored by the client builder.
+    fn validate_provider_block(&self) -> Result<(), OnDemandMigrationConfigError> {
+        let missing = OnDemandMigrationConfigError::MissingProviderBlock;
+        let unexpected = OnDemandMigrationConfigError::UnexpectedProviderBlock;
+        let invalid = OnDemandMigrationConfigError::InvalidProviderBlock;
+
+        if self.provider != Provider::Azure && self.azure.is_some() {
+            return Err(unexpected("azure", self.provider));
+        }
+        if self.provider != Provider::GcsNative && self.gcs.is_some() {
+            return Err(unexpected("gcs", self.provider));
+        }
+
+        match self.provider {
+            Provider::Azure => {
+                let azure = self.azure.as_ref().ok_or(missing("azure", self.provider))?;
+                if azure.account.is_empty() {
+                    return Err(invalid("azure", "account must not be empty"));
+                }
+                // The account feeds a hostname when the endpoint is derived:
+                // keep it to label characters so it cannot rewrite the host.
+                if !azure.account.bytes().all(|b| b.is_ascii_alphanumeric() || b == b'-') {
+                    return Err(invalid("azure", "account contains characters outside [A-Za-z0-9-]"));
+                }
+                match (azure.account_key.as_deref(), azure.sas_token.as_deref()) {
+                    (Some(_), Some(_)) => return Err(invalid("azure", "account_key and sas_token are mutually exclusive")),
+                    (None, None) => return Err(invalid("azure", "one of account_key and sas_token is required")),
+                    (Some(key), None) => {
+                        if key.is_empty() {
+                            return Err(invalid("azure", "account_key must not be empty"));
+                        }
+                        // Decoded here so a mistyped key fails at the admin
+                        // boundary instead of on the first source request.
+                        if base64_simd::STANDARD.decode_to_vec(key.as_bytes()).is_err() {
+                            return Err(invalid("azure", "account_key is not base64"));
+                        }
+                    }
+                    (None, Some(sas)) => {
+                        if sas.is_empty() {
+                            return Err(invalid("azure", "sas_token must not be empty"));
+                        }
+                        if sas.starts_with('?') {
+                            return Err(invalid("azure", "sas_token must not start with '?'"));
+                        }
+                        if sas.chars().any(char::is_whitespace) {
+                            return Err(invalid("azure", "sas_token must not contain whitespace"));
+                        }
+                    }
+                }
+            }
+            Provider::GcsNative => {
+                let gcs = self.gcs.as_ref().ok_or(missing("gcs", self.provider))?;
+                let key: serde_json::Value = serde_json::from_str(&gcs.service_account_json)
+                    .map_err(|_| invalid("gcs", "service_account_json is not valid JSON"))?;
+                let Some(object) = key.as_object() else {
+                    return Err(invalid("gcs", "service_account_json is not a JSON object"));
+                };
+                if object.get("type").and_then(serde_json::Value::as_str) != Some("service_account") {
+                    return Err(invalid("gcs", "service_account_json is not a service_account key"));
+                }
+                for field in ["client_email", "private_key"] {
+                    if object
+                        .get(field)
+                        .and_then(serde_json::Value::as_str)
+                        .is_none_or(str::is_empty)
+                    {
+                        return Err(invalid("gcs", "service_account_json is missing client_email or private_key"));
+                    }
+                }
+            }
+            Provider::S3 | Provider::Aws | Provider::Minio | Provider::Rustfs | Provider::R2 | Provider::Gcs => {}
+        }
+
+        Ok(())
+    }
 }
 
 fn validate_endpoint(endpoint: &str) -> Result<(), OnDemandMigrationConfigError> {
@@ -699,7 +895,15 @@ mod tests {
             ),
             (
                 "provider enum",
-                r#"{"source":{"provider":"azure","endpoint":"https://h","region":"r","bucket":"b"}}"#,
+                r#"{"source":{"provider":"swift","endpoint":"https://h","region":"r","bucket":"b"}}"#,
+            ),
+            (
+                "azure block",
+                r#"{"source":{"provider":"azure","region":"auto","bucket":"b","azure":{"account":"acct","account_key":"a2V5","extra":1}}}"#,
+            ),
+            (
+                "gcs block",
+                r#"{"source":{"provider":"gcs_native","region":"auto","bucket":"b","gcs":{"service_account_json":"{}","extra":1}}}"#,
             ),
         ] {
             let err = OnDemandMigrationConfig::from_json(json.as_bytes()).expect_err(label);
@@ -820,9 +1024,201 @@ mod tests {
                 "{provider}"
             );
         }
+        // The native providers never sign with a region, so "auto" is the
+        // honest value to write for them.
+        for cfg in [azure_cfg(), gcs_native_cfg()] {
+            assert_eq!(cfg.source.region, "auto");
+            cfg.validate(empty_ctx())
+                .unwrap_or_else(|err| panic!("{}: {err}", cfg.source.provider));
+        }
         assert_eq!(sample().source.effective_region(), "us-west-1");
     }
 
+    const SERVICE_ACCOUNT_JSON: &str = r#"{"type":"service_account","project_id":"p","client_email":"a@b.iam.gserviceaccount.com","private_key":"-----BEGIN PRIVATE KEY-----\nx\n-----END PRIVATE KEY-----\n"}"#;
+
+    fn azure_cfg() -> OnDemandMigrationConfig {
+        let mut cfg = sample();
+        cfg.source.provider = Provider::Azure;
+        cfg.source.endpoint = None;
+        cfg.source.region = "auto".to_string();
+        cfg.source.credentials = None;
+        cfg.source.azure = Some(AzureSourceConfig {
+            account: "legacyaccount".to_string(),
+            account_key: Some("c2VjcmV0LWtleQ==".to_string()),
+            sas_token: None,
+        });
+        cfg
+    }
+
+    fn gcs_native_cfg() -> OnDemandMigrationConfig {
+        let mut cfg = sample();
+        cfg.source.provider = Provider::GcsNative;
+        cfg.source.endpoint = None;
+        cfg.source.region = "auto".to_string();
+        cfg.source.credentials = None;
+        cfg.source.gcs = Some(GcsSourceConfig {
+            service_account_json: SERVICE_ACCOUNT_JSON.to_string(),
+        });
+        cfg
+    }
+
+    #[test]
+    fn native_providers_derive_their_endpoint_and_round_trip_on_the_wire() {
+        let azure = azure_cfg();
+        assert_eq!(azure.source.effective_endpoint(), "https://legacyaccount.blob.core.windows.net");
+        let gcs = gcs_native_cfg();
+        assert_eq!(gcs.source.effective_endpoint(), "https://storage.googleapis.com");
+
+        for cfg in [azure_cfg(), gcs_native_cfg()] {
+            let json = cfg.to_json().expect("config must serialize");
+            assert_eq!(OnDemandMigrationConfig::from_json(&json).expect("config must parse"), cfg);
+        }
+        // The wire labels are part of the admin contract.
+        assert!(
+            String::from_utf8(azure_cfg().to_json().expect("json"))
+                .expect("utf8")
+                .contains(r#""provider":"azure""#)
+        );
+        assert!(
+            String::from_utf8(gcs_native_cfg().to_json().expect("json"))
+                .expect("utf8")
+                .contains(r#""provider":"gcs_native""#)
+        );
+    }
+
+    #[test]
+    fn an_explicit_endpoint_overrides_the_derived_native_one() {
+        // Azurite and fake-gcs-server are addressed this way.
+        let mut cfg = azure_cfg();
+        cfg.source.endpoint = Some("http://azurite.example.com:10000".to_string());
+        cfg.validate(empty_ctx()).expect("an explicit native endpoint is allowed");
+        assert_eq!(cfg.source.effective_endpoint(), "http://azurite.example.com:10000");
+
+        cfg.source.endpoint = Some("http://azurite.example.com:10000/devstoreaccount1".to_string());
+        assert!(
+            matches!(cfg.validate(empty_ctx()), Err(OnDemandMigrationConfigError::InvalidEndpoint(_))),
+            "a native endpoint is still an origin"
+        );
+    }
+
+    #[test]
+    fn a_provider_block_belongs_to_exactly_its_own_provider() {
+        let mut cfg = sample();
+        cfg.source.azure = azure_cfg().source.azure;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("azure", Provider::S3))
+        );
+
+        let mut cfg = sample();
+        cfg.source.gcs = gcs_native_cfg().source.gcs;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::UnexpectedProviderBlock("gcs", Provider::S3))
+        );
+
+        let mut cfg = azure_cfg();
+        cfg.source.azure = None;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::MissingProviderBlock("azure", Provider::Azure))
+        );
+
+        let mut cfg = gcs_native_cfg();
+        cfg.source.gcs = None;
+        assert_eq!(
+            cfg.validate(empty_ctx()),
+            Err(OnDemandMigrationConfigError::MissingProviderBlock("gcs", Provider::GcsNative))
+        );
+    }
+
+    #[test]
+    fn azure_block_rules() {
+        let with = |account: &str, key: Option<&str>, sas: Option<&str>| {
+            let mut cfg = azure_cfg();
+            cfg.source.azure = Some(AzureSourceConfig {
+                account: account.to_string(),
+                account_key: key.map(str::to_string),
+                sas_token: sas.map(str::to_string),
+            });
+            cfg.validate(empty_ctx())
+        };
+
+        with("legacyaccount", None, Some("sv=2021-08-06&sig=abc%3D")).expect("a SAS token is a complete credential");
+        with("legacyaccount", Some("c2VjcmV0LWtleQ=="), None).expect("an account key is a complete credential");
+
+        for (label, result) in [
+            ("empty account", with("", Some("c2VjcmV0LWtleQ=="), None)),
+            // The account becomes the first label of the derived hostname.
+            ("account with a dot", with("legacy.account", Some("c2VjcmV0LWtleQ=="), None)),
+            ("account with a slash", with("legacy/account", Some("c2VjcmV0LWtleQ=="), None)),
+            ("no credential", with("legacyaccount", None, None)),
+            ("both credentials", with("legacyaccount", Some("c2VjcmV0LWtleQ=="), Some("sv=1"))),
+            ("empty key", with("legacyaccount", Some(""), None)),
+            ("key that is not base64", with("legacyaccount", Some("not base64!"), None)),
+            ("empty sas", with("legacyaccount", None, Some(""))),
+            ("sas with a leading question mark", with("legacyaccount", None, Some("?sv=1"))),
+            ("sas with whitespace", with("legacyaccount", None, Some("sv=1 &sig=a"))),
+        ] {
+            assert!(
+                matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("azure", _))),
+                "{label}: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn gcs_native_block_requires_a_usable_service_account_key() {
+        let with = |json: &str| {
+            let mut cfg = gcs_native_cfg();
+            cfg.source.gcs = Some(GcsSourceConfig {
+                service_account_json: json.to_string(),
+            });
+            cfg.validate(empty_ctx())
+        };
+
+        with(SERVICE_ACCOUNT_JSON).expect("a service-account key is accepted");
+        for (label, json) in [
+            ("empty", ""),
+            ("not json", "not json"),
+            ("not an object", "[]"),
+            ("wrong type", r#"{"type":"authorized_user","client_email":"a@b","private_key":"k"}"#),
+            ("no private key", r#"{"type":"service_account","client_email":"a@b"}"#),
+            ("empty client email", r#"{"type":"service_account","client_email":"","private_key":"k"}"#),
+        ] {
+            let result = with(json);
+            assert!(
+                matches!(result, Err(OnDemandMigrationConfigError::InvalidProviderBlock("gcs", _))),
+                "{label}: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn native_secrets_never_survive_redaction_or_debug() {
+        let mut azure = azure_cfg();
+        azure.source.azure.as_mut().expect("block").sas_token = Some("sv=2021-08-06&sig=top-secret".to_string());
+        azure.source.azure.as_mut().expect("block").account_key = None;
+        let gcs = gcs_native_cfg();
+
+        for rendered in [
+            format!("{:?}", azure.redacted()),
+            format!("{azure:?}"),
+            String::from_utf8(azure.redacted().to_json().expect("json")).expect("utf8"),
+        ] {
+            assert!(!rendered.contains("top-secret"), "{rendered}");
+            assert!(rendered.contains("legacyaccount"), "the account name is not a secret: {rendered}");
+        }
+        for rendered in [
+            format!("{:?}", gcs.redacted()),
+            format!("{gcs:?}"),
+            String::from_utf8(gcs.redacted().to_json().expect("json")).expect("utf8"),
+        ] {
+            assert!(!rendered.contains("BEGIN PRIVATE KEY"), "{rendered}");
+            assert!(!rendered.contains("gserviceaccount"), "{rendered}");
+        }
+    }
+
     #[test]
     fn bucket_rules() {
         let mut cfg = sample();

crates/ecstore/src/bucket/on_demand_migration/gcs.rs

@@ -0,0 +1,506 @@
+// 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.
+
+//! Native Google Cloud Storage source backend.
+//!
+//! The `gcs` provider already reaches GCS through its S3 interoperability API,
+//! which needs an HMAC key pair. This backend is the other half: it authorizes
+//! with a service-account key, the credential most GCS projects actually issue,
+//! by minting OAuth tokens through the shared `google-cloud-auth` credential
+//! machinery the tier layer already uses.
+//!
+//! Two GCS surfaces are involved, each for the half it describes best. The read
+//! path uses the XML API (`/{bucket}/{object}`), whose responses carry
+//! `x-goog-meta-*` user metadata and the `x-goog-hash` digest in one round trip.
+//! Listing uses the JSON API (`objects.list`), whose `pageToken` maps directly
+//! onto the shared page cursor and whose `prefixes` are the delimiter roll-up.
+//! Both accept the same bearer token.
+//!
+//! Every call this backend makes needs only `storage.objects.get` and
+//! `storage.objects.list`, the two permissions of the `objectViewer` role, so a
+//! key scoped to exactly the migration's needs works.
+//!
+//! `x-goog-hash` carries a base64 MD5 for every non-composite object; it is
+//! converted to hex and becomes the head's ETag, so a pulled object is checked
+//! against the digest GCS itself computed. A composite object has no MD5, and
+//! its ETag is then marked opaque rather than checked.
+
+use super::native_http::{
+    NativeHeadFields, NativeHttp, base64_md5_to_hex, header, native_source_head, parse_http_timestamp, read_text, response_body,
+};
+use super::source_client::{
+    GcsSourceSpec, SourceBackend, SourceError, SourceGet, SourceHead, SourceListRequest, SourceObject, SourcePage,
+    SourceTimeouts, range_header_value,
+};
+use crate::bucket::remote_s3_client::RemoteS3ClientError;
+use crate::storage_api_contracts::range::HTTPRangeSpec;
+use google_cloud_auth::credentials::service_account::{AccessSpecifier, Builder as ServiceAccountBuilder};
+use google_cloud_auth::credentials::{CacheableResource, Credentials};
+use http::{HeaderMap, HeaderValue, Method};
+use serde::Deserialize;
+use std::collections::HashMap;
+use url::Url;
+
+/// Read-only object scope: this backend never writes to the source.
+const READ_ONLY_SCOPE: &str = "https://www.googleapis.com/auth/devstorage.read_only";
+const METADATA_PREFIX: &str = "x-goog-meta-";
+/// GCS reports its error code in the response body, not a header; the shared
+/// transport takes a header name, so it is given one that never matches and
+/// classification falls back to the status.
+const NO_ERROR_CODE_HEADER: &str = "x-goog-unused-error-code";
+/// One `objects.list` page is small; refuse an unbounded document.
+const MAX_JSON_BYTES: usize = 8 * 1024 * 1024;
+
+pub struct GcsNativeSourceBackend {
+    http: NativeHttp,
+    bucket: String,
+    credentials: Credentials,
+}
+
+impl GcsNativeSourceBackend {
+    pub fn new(
+        endpoint: &str,
+        bucket: &str,
+        spec: &GcsSourceSpec,
+        timeouts: SourceTimeouts,
+        skip_tls_verify: bool,
+        ca_cert_pem: Option<&str>,
+    ) -> Result<Self, RemoteS3ClientError> {
+        let key: serde_json::Value = serde_json::from_str(&spec.service_account_json)
+            .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not valid JSON"))?;
+        let credentials = ServiceAccountBuilder::new(key)
+            .with_access_specifier(AccessSpecifier::from_scopes([READ_ONLY_SCOPE]))
+            .build()
+            .map_err(|_| RemoteS3ClientError::Credentials("gcs service account key is not usable"))?;
+        Ok(Self {
+            http: NativeHttp::new(endpoint, timeouts, skip_tls_verify, ca_cert_pem)?,
+            bucket: bucket.to_string(),
+            credentials,
+        })
+    }
+
+    /// Authorization headers for one request. A credential failure is reported
+    /// as `AccessDenied` with no message: the renderer of a credential error
+    /// has the key material in scope, and the class is what callers act on.
+    async fn auth_headers(&self) -> Result<HeaderMap, SourceError> {
+        match self.credentials.headers(http::Extensions::new()).await {
+            Ok(CacheableResource::New { data, .. }) => Ok(data),
+            // Only returned when the caller passes an entity tag, which this
+            // backend never does; an empty set is still the honest answer.
+            Ok(CacheableResource::NotModified) => Ok(HeaderMap::new()),
+            Err(_) => Err(SourceError::AccessDenied),
+        }
+    }
+
+    /// XML API URL of one object; `/` in the key stay path separators.
+    fn object_url(&self, key: &str) -> Result<Url, SourceError> {
+        self.http.url(std::iter::once(self.bucket.as_str()).chain(key.split('/')))
+    }
+
+    /// JSON API URL of the bucket's object collection.
+    fn objects_url(&self) -> Result<Url, SourceError> {
+        self.http.url(["storage", "v1", "b", self.bucket.as_str(), "o"])
+    }
+
+    async fn request(&self, method: Method, url: Url, mut headers: HeaderMap) -> Result<reqwest::Request, SourceError> {
+        for (name, value) in self.auth_headers().await? {
+            if let Some(name) = name {
+                headers.insert(name, value);
+            }
+        }
+        let mut request = reqwest::Request::new(method, url);
+        *request.headers_mut() = headers;
+        Ok(request)
+    }
+
+    /// Shared mapping for the XML API's HEAD and GET responses.
+    fn head_from_response(headers: &HeaderMap) -> Result<SourceHead, SourceError> {
+        if header(headers, "x-goog-encryption-key-sha256").is_some() {
+            return Err(SourceError::Unsupported(
+                "source object uses a customer-supplied encryption key; customer-key sources are not supported".to_string(),
+            ));
+        }
+        // `x-goog-hash` lists digests as `name=base64`, comma separated, and may
+        // repeat across header lines. Only the MD5 describes the whole object.
+        let md5 = headers
+            .get_all("x-goog-hash")
+            .iter()
+            .filter_map(|value| value.to_str().ok())
+            .flat_map(|value| value.split(','))
+            .filter_map(|digest| digest.trim().strip_prefix("md5="))
+            .find_map(base64_md5_to_hex);
+
+        let (etag, etag_is_opaque) = match md5 {
+            Some(md5) => (Some(md5), false),
+            // A composite object has no MD5; its ETag describes the composition
+            // rather than the bytes, so it is provenance only.
+            None => (header(headers, "etag").map(str::to_string), true),
+        };
+        native_source_head(
+            headers,
+            METADATA_PREFIX,
+            NativeHeadFields {
+                etag,
+                etag_is_opaque,
+                version_id: header(headers, "x-goog-generation").map(str::to_string),
+                storage_class: header(headers, "x-goog-storage-class").map(str::to_string),
+            },
+        )
+    }
+}
+
+#[async_trait::async_trait]
+impl SourceBackend for GcsNativeSourceBackend {
+    async fn head(&self, key: &str) -> Result<SourceHead, SourceError> {
+        let request = self.request(Method::HEAD, self.object_url(key)?, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        Self::head_from_response(response.headers())
+    }
+
+    async fn get(&self, key: &str, range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+        let mut headers = HeaderMap::new();
+        if let Some(range) = range.map(range_header_value).transpose()? {
+            headers.insert(
+                http::header::RANGE,
+                HeaderValue::from_str(&range).map_err(|_| SourceError::Other("invalid range header".to_string()))?,
+            );
+        }
+        let request = self.request(Method::GET, self.object_url(key)?, headers).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let head = Self::head_from_response(response.headers())?;
+        let content_range = header(response.headers(), "content-range").map(str::to_string);
+        Ok(SourceGet {
+            head,
+            body: response_body(response),
+            content_range,
+        })
+    }
+
+    async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+        // `objects.list` offers `startOffset`, which is inclusive, so it cannot
+        // express "resume after this key" without silently repeating it.
+        if request.start_after.is_some() {
+            return Err(SourceError::Unsupported(
+                "gcs sources cannot resume a listing from a key; use the continuation token".to_string(),
+            ));
+        }
+        let mut url = self.objects_url()?;
+        {
+            let mut query = url.query_pairs_mut();
+            if let Some(prefix) = request.prefix.filter(|prefix| !prefix.is_empty()) {
+                query.append_pair("prefix", prefix);
+            }
+            if let Some(delimiter) = request.delimiter.filter(|delimiter| !delimiter.is_empty()) {
+                query.append_pair("delimiter", delimiter);
+            }
+            if let Some(token) = request.continuation_token.filter(|token| !token.is_empty()) {
+                query.append_pair("pageToken", token);
+            }
+            if request.max_keys > 0 {
+                query.append_pair("maxResults", &request.max_keys.to_string());
+            }
+        }
+
+        let request = self.request(Method::GET, url, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        let body = read_text(response, MAX_JSON_BYTES).await?;
+        parse_objects_list(&body)
+    }
+
+    /// GCS has no object tagging API; user metadata is already carried by the
+    /// head mapping. An empty map keeps `policy.copy_tags` from failing a pull
+    /// over a concept the provider does not have.
+    async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
+        Ok(HashMap::new())
+    }
+
+    /// A one-object listing, not `buckets.get`: the migration pipeline only
+    /// ever needs `storage.objects.list` and `storage.objects.get`, and a key
+    /// scoped to exactly those (the `objectViewer` role) cannot read the bucket
+    /// resource. Probing with `buckets.get` would reject a correct key.
+    async fn probe(&self) -> Result<(), SourceError> {
+        let mut url = self.objects_url()?;
+        url.query_pairs_mut().append_pair("maxResults", "1");
+        let request = self.request(Method::GET, url, HeaderMap::new()).await?;
+        let response = self.http.send(request, NO_ERROR_CODE_HEADER).await?;
+        read_text(response, MAX_JSON_BYTES)
+            .await
+            .and_then(|body| parse_objects_list(&body))?;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ObjectsList {
+    #[serde(default)]
+    items: Vec<ListedObject>,
+    #[serde(default)]
+    prefixes: Vec<String>,
+    #[serde(default)]
+    next_page_token: Option<String>,
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(rename_all = "camelCase")]
+struct ListedObject {
+    name: String,
+    /// GCS renders the size as a decimal string, not a JSON number.
+    #[serde(default)]
+    size: Option<String>,
+    #[serde(default)]
+    updated: Option<String>,
+    #[serde(default)]
+    md5_hash: Option<String>,
+    #[serde(default)]
+    etag: Option<String>,
+    #[serde(default)]
+    storage_class: Option<String>,
+}
+
+fn parse_objects_list(body: &str) -> Result<SourcePage, SourceError> {
+    let listing: ObjectsList =
+        serde_json::from_str(body).map_err(|err| SourceError::Other(format!("source listing is not valid JSON: {err}")))?;
+    let next_continuation_token = listing.next_page_token.filter(|token| !token.is_empty());
+    let objects = listing
+        .items
+        .into_iter()
+        .map(|item| {
+            let etag = item
+                .md5_hash
+                .as_deref()
+                .and_then(base64_md5_to_hex)
+                .or_else(|| item.etag.map(|etag| etag.trim_matches('"').to_string()))
+                .filter(|etag| !etag.is_empty());
+            SourceObject {
+                key: item.name,
+                etag,
+                size: item.size.and_then(|size| size.parse().ok()).unwrap_or(0),
+                last_modified: item.updated.as_deref().and_then(parse_http_timestamp),
+                storage_class: item.storage_class,
+                // GCS never encodes a part count in a digest or an ETag.
+                is_multipart_etag: false,
+            }
+        })
+        .collect();
+
+    Ok(SourcePage {
+        objects,
+        common_prefixes: listing.prefixes,
+        is_truncated: next_continuation_token.is_some(),
+        next_continuation_token,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, assert_backend_contract};
+    use crate::bucket::on_demand_migration::test_http_fixture::{ScriptedResponse, scripted_server};
+    use google_cloud_auth::credentials::anonymous::Builder as AnonymousBuilder;
+
+    const LIST_PAGE_ONE: &str = r#"{
+      "kind": "storage#objects",
+      "nextPageToken": "cursor-1",
+      "prefixes": ["dir/sub/"],
+      "items": [
+        {
+          "name": "dir/a.txt",
+          "size": "5",
+          "updated": "2015-10-21T07:28:00.000Z",
+          "md5Hash": "XUFAKrxLKna5cZ2REBfFkg==",
+          "etag": "CJizy9Wq0McCEAE=",
+          "storageClass": "STANDARD"
+        }
+      ]
+    }"#;
+
+    const LIST_PAGE_TWO: &str = r#"{
+      "kind": "storage#objects",
+      "items": [
+        {
+          "name": "dir/b.txt",
+          "size": "7",
+          "updated": "2015-10-21T07:28:00.000Z",
+          "etag": "\"CJizy9Wq0McCEAI=\""
+        }
+      ]
+    }"#;
+
+    fn backend(endpoint: &Url) -> GcsNativeSourceBackend {
+        GcsNativeSourceBackend {
+            http: NativeHttp::for_test(endpoint.clone()),
+            bucket: "legacy".to_string(),
+            // Anonymous credentials add no headers, so the fixture sees exactly
+            // the request this backend builds.
+            credentials: AnonymousBuilder::new().build(),
+        }
+    }
+
+    fn object_headers() -> Vec<(&'static str, String)> {
+        vec![
+            ("Content-Type", "text/plain".to_string()),
+            ("Last-Modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
+            ("ETag", "\"CJizy9Wq0McCEAE=\"".to_string()),
+            ("x-goog-hash", "crc32c=AAAAAA==,md5=XUFAKrxLKna5cZ2REBfFkg==".to_string()),
+            ("x-goog-meta-owner", "alice".to_string()),
+            ("x-goog-storage-class", "STANDARD".to_string()),
+            ("x-goog-generation", "1445412480000000".to_string()),
+        ]
+    }
+
+    #[test]
+    fn objects_list_maps_items_prefixes_and_the_page_token() {
+        let page = parse_objects_list(LIST_PAGE_ONE).expect("page should parse");
+        assert_eq!(page.common_prefixes, vec!["dir/sub/"]);
+        assert!(page.is_truncated);
+        assert_eq!(page.next_continuation_token.as_deref(), Some("cursor-1"));
+        assert_eq!(page.objects.len(), 1);
+        assert_eq!(page.objects[0].key, "dir/a.txt");
+        assert_eq!(page.objects[0].size, 5, "the string size is parsed");
+        assert_eq!(
+            page.objects[0].etag.as_deref(),
+            Some("5d41402abc4b2a76b9719d911017c592"),
+            "the base64 md5Hash becomes a hex ETag"
+        );
+        assert_eq!(page.objects[0].storage_class.as_deref(), Some("STANDARD"));
+        assert!(page.objects[0].last_modified.is_some(), "RFC 3339 `updated` is parsed");
+
+        let page = parse_objects_list(LIST_PAGE_TWO).expect("page should parse");
+        assert!(!page.is_truncated);
+        assert!(page.next_continuation_token.is_none());
+        assert_eq!(
+            page.objects[0].etag.as_deref(),
+            Some("CJizy9Wq0McCEAI="),
+            "without md5Hash the raw etag is carried"
+        );
+
+        assert!(parse_objects_list("not json").is_err());
+    }
+
+    #[tokio::test]
+    async fn head_prefers_the_goog_hash_md5_over_the_etag() {
+        let (endpoint, recorded) = scripted_server(vec![ScriptedResponse::new(200, object_headers(), String::new())]).await;
+        let head = backend(&endpoint).head("dir/a b.txt").await.expect("HEAD should map");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert_eq!(recorded[0].method, "HEAD");
+        assert_eq!(recorded[0].target, "/legacy/dir/a%20b.txt", "the XML API addresses the object by path");
+        assert_eq!(
+            head.etag.as_deref(),
+            Some("5d41402abc4b2a76b9719d911017c592"),
+            "the x-goog-hash md5 is the content digest"
+        );
+        assert!(!head.etag_is_opaque, "a GCS md5 may be checked against the pulled bytes");
+        assert_eq!(head.user_metadata, HashMap::from([("owner".to_string(), "alice".to_string())]));
+        assert_eq!(head.version_id.as_deref(), Some("1445412480000000"));
+        assert_eq!(head.storage_class.as_deref(), Some("STANDARD"));
+    }
+
+    #[tokio::test]
+    async fn a_composite_object_without_an_md5_keeps_an_opaque_etag() {
+        let headers = object_headers()
+            .into_iter()
+            .map(|(name, value)| {
+                if name == "x-goog-hash" {
+                    (name, "crc32c=AAAAAA==".to_string())
+                } else {
+                    (name, value)
+                }
+            })
+            .collect();
+        let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
+        let head = backend(&endpoint).head("composed").await.expect("HEAD should map");
+        assert_eq!(head.etag.as_deref(), Some("CJizy9Wq0McCEAE="));
+        assert!(head.etag_is_opaque, "a composite ETag describes the composition, not the bytes");
+    }
+
+    #[tokio::test]
+    async fn customer_supplied_key_objects_are_refused() {
+        let mut headers = object_headers();
+        headers.push(("x-goog-encryption-key-sha256", "abc".to_string()));
+        let (endpoint, _) = scripted_server(vec![ScriptedResponse::new(200, headers, String::new())]).await;
+        let err = backend(&endpoint)
+            .head("a.txt")
+            .await
+            .expect_err("CSEK objects are unsupported");
+        assert!(matches!(err, SourceError::Unsupported(_)), "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn list_and_probe_address_the_json_api() {
+        let (endpoint, recorded) = scripted_server(vec![
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
+        ])
+        .await;
+        let backend = backend(&endpoint);
+
+        backend
+            .list(&SourceListRequest {
+                prefix: Some("dir/"),
+                delimiter: Some("/"),
+                continuation_token: Some("cursor-0"),
+                max_keys: 2,
+                ..Default::default()
+            })
+            .await
+            .expect("listing should succeed");
+        backend.probe().await.expect("probe should succeed");
+
+        let recorded = recorded.lock().expect("recorder lock").clone();
+        assert!(recorded[0].target.starts_with("/storage/v1/b/legacy/o?"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("prefix=dir%2F"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("delimiter=%2F"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("pageToken=cursor-0"), "{}", recorded[0].target);
+        assert!(recorded[0].target.contains("maxResults=2"), "{}", recorded[0].target);
+        assert_eq!(
+            recorded[1].target, "/storage/v1/b/legacy/o?maxResults=1",
+            "the probe uses the listing permission the pipeline already needs"
+        );
+    }
+
+    #[tokio::test]
+    async fn gcs_native_backend_satisfies_the_shared_backend_contract() {
+        let mut ranged = object_headers();
+        ranged.push(("Content-Range", "bytes 1-3/5".to_string()));
+        // A HEAD reports the object size with no body, exactly as GCS does.
+        let mut head_only = object_headers();
+        head_only.push(("Content-Length", "5".to_string()));
+        let (endpoint, _) = scripted_server(vec![
+            ScriptedResponse::new(200, head_only, String::new()),
+            ScriptedResponse::new(200, object_headers(), "hello".to_string()),
+            ScriptedResponse::new(206, ranged, "ell".to_string()),
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_ONE.to_string()),
+            ScriptedResponse::new(200, Vec::new(), LIST_PAGE_TWO.to_string()),
+            // GCS has no tagging call, so the contract's tag step issues no
+            // request; the probe is the next one on the wire.
+            ScriptedResponse::new(200, Vec::new(), "{}".to_string()),
+            ScriptedResponse::new(404, Vec::new(), String::new()),
+            ScriptedResponse::new(403, Vec::new(), String::new()),
+        ])
+        .await;
+
+        assert_backend_contract(
+            &backend(&endpoint),
+            BackendCapabilities {
+                etag_is_opaque: false,
+                supports_start_after: false,
+                // GCS objects have no tags; the contract's tag step is skipped.
+                supports_tagging: false,
+            },
+        )
+        .await;
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/mod.rs

@@ -19,25 +19,36 @@
 //! client, and the per-node runtime (`sys`) that turns configs into live
 //! clients guarded by a breaker, a negative cache, singleflight and a pull
 //! concurrency limit (rustfs/backlog#2147).
+//!
+//! A source is reached through one `SourceBackend`: the S3 dialect for every
+//! S3-compatible provider, and a native backend for the providers that have no
+//! S3 API (`azure`, `gcs_native`).
 
+pub mod azure;
+#[cfg(test)]
+mod backend_contract;
 pub mod backfill;
 pub mod breaker;
 pub mod config;
+pub mod gcs;
 pub mod list_through;
+mod native_http;
 pub mod negative_cache;
 pub mod pull;
 pub mod source_client;
 pub mod stats;
 pub mod sys;
+#[cfg(test)]
+mod test_http_fixture;
 
 pub use breaker::{
     BREAKER_FAILURE_THRESHOLD, BREAKER_FAILURE_WINDOW, BREAKER_HALF_OPEN_MAX_PROBES, BREAKER_OPEN_DURATION, Breaker,
     BreakerState, BreakerTransition, BreakerVerdict,
 };
 pub use config::{
-    ConfigPublishHook, FilterConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK, ON_DEMAND_MIGRATION_CONFIG_VERSION,
-    OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider, RangeGetPolicy, SourceConfig,
-    SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
+    AzureSourceConfig, ConfigPublishHook, FilterConfig, GcsSourceConfig, HeadPolicy, ON_DEMAND_MIGRATION_CONFIG_HOOK,
+    ON_DEMAND_MIGRATION_CONFIG_VERSION, OnDemandMigrationConfig, OnDemandMigrationConfigError, PathStyle, PolicyConfig, Provider,
+    RangeGetPolicy, SourceConfig, SourceCredentials, SourceErrorPolicy, SourceTimeout, TlsConfig, ValidationContext,
 };
 pub use list_through::{
     FetchRequest, LIST_THROUGH_TOKEN_VERSION, ListEntryKey, ListThroughCursor, ListThroughMerger, ListThroughToken,
@@ -56,5 +67,6 @@ pub use stats::{
 };
 pub use sys::{
     ApplyOutcome, BucketOdmState, GLOBAL_ON_DEMAND_MIGRATION_SYS, OdmBucketSnapshot, OdmLookup, OdmStateError,
-    OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_client_spec,
+    OnDemandMigrationSys, PullError, PullFollower, PullLeader, PullOutcome, PullResult, PullSlot, source_backend_spec,
+    source_client_spec,
 };

crates/ecstore/src/bucket/on_demand_migration/native_http.rs

@@ -0,0 +1,415 @@
+// 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.
+
+//! Shared HTTP transport for the on-demand migration source backends that do
+//! not speak S3 (Azure Blob, native GCS).
+//!
+//! The S3 backend rides the AWS SDK; these providers have no SigV4 dialect, so
+//! they talk plain HTTP through one `reqwest` client that carries the same
+//! connect/read timeouts and TLS policy the operator configured for the source.
+//! Redirects are refused: the endpoint passed the outbound policy gate once, and
+//! following a source-chosen `Location` would leave that gate behind.
+//!
+//! Errors never render the request URL. A SAS token lives in the query string,
+//! so a `reqwest` error rendered with its URL would print the credential into
+//! the log line and the admin response.
+
+use super::source_client::{SourceError, SourceHead, SourceTimeouts, USER_AGENT_SUFFIX, classify_status, is_multipart_etag};
+use crate::bucket::remote_s3_client::{RemoteS3ClientError, validate_remote_endpoint, validate_target_ca_pem};
+use aws_sdk_s3::primitives::ByteStream;
+use aws_smithy_types::body::SdkBody;
+use futures::StreamExt;
+use http::HeaderMap;
+use std::collections::HashMap;
+use std::time::SystemTime;
+use time::OffsetDateTime;
+use time::format_description::well_known::{Rfc2822, Rfc3339};
+use url::Url;
+
+/// Origin the native backends are allowed to address, plus the HTTP client
+/// that reaches it.
+pub(super) struct NativeHttp {
+    client: reqwest::Client,
+    endpoint: Url,
+}
+
+impl NativeHttp {
+    /// `endpoint` must be a bare `scheme://host[:port]` origin; it is checked
+    /// against the outbound policy exactly like an S3 source endpoint.
+    pub(super) fn new(
+        endpoint: &str,
+        timeouts: SourceTimeouts,
+        skip_tls_verify: bool,
+        ca_cert_pem: Option<&str>,
+    ) -> Result<Self, RemoteS3ClientError> {
+        let endpoint = Url::parse(endpoint.trim()).map_err(|err| RemoteS3ClientError::InvalidEndpoint(err.to_string()))?;
+        if !matches!(endpoint.scheme(), "http" | "https") {
+            return Err(RemoteS3ClientError::InvalidEndpoint(format!(
+                "unsupported scheme {}; expected http or https",
+                endpoint.scheme()
+            )));
+        }
+        if endpoint.host_str().is_none_or(str::is_empty) {
+            return Err(RemoteS3ClientError::InvalidEndpoint("endpoint has no host".to_string()));
+        }
+        if !endpoint.username().is_empty() || endpoint.password().is_some() {
+            return Err(RemoteS3ClientError::InvalidEndpoint("endpoint must not carry userinfo".to_string()));
+        }
+        if !matches!(endpoint.path(), "" | "/") || endpoint.query().is_some() || endpoint.fragment().is_some() {
+            return Err(RemoteS3ClientError::InvalidEndpoint(
+                "endpoint must be an origin without path, query or fragment".to_string(),
+            ));
+        }
+        validate_remote_endpoint(&endpoint).map_err(RemoteS3ClientError::EndpointNotAllowed)?;
+
+        let mut builder = reqwest::Client::builder()
+            .connect_timeout(timeouts.connect)
+            .read_timeout(timeouts.read)
+            .redirect(reqwest::redirect::Policy::none())
+            .user_agent(USER_AGENT_SUFFIX);
+        if skip_tls_verify {
+            builder = builder.danger_accept_invalid_certs(true);
+        } else if let Some(pem) = ca_cert_pem.map(str::trim).filter(|pem| !pem.is_empty()) {
+            // Reject a malformed bundle the same way the S3 path does, so the
+            // operator sees "invalid CA PEM" instead of a TLS handshake failure.
+            validate_target_ca_pem(pem)?;
+            let certificate = reqwest::Certificate::from_pem(pem.as_bytes())
+                .map_err(|err| RemoteS3ClientError::InvalidCaPem(err.to_string()))?;
+            builder = builder.add_root_certificate(certificate);
+        }
+
+        let client = builder
+            .build()
+            .map_err(|err| RemoteS3ClientError::InvalidEndpoint(format!("http client cannot be built: {err}")))?;
+        Ok(Self { client, endpoint })
+    }
+
+    #[cfg(test)]
+    pub(super) fn for_test(endpoint: Url) -> Self {
+        Self {
+            client: reqwest::Client::builder()
+                .redirect(reqwest::redirect::Policy::none())
+                .build()
+                .expect("test http client should build"),
+            endpoint,
+        }
+    }
+
+    /// A URL under the endpoint origin. `segments` are percent-encoded as
+    /// path segments, so a key containing `?`, `#` or a space cannot rewrite
+    /// the request target.
+    pub(super) fn url<'a>(&self, segments: impl IntoIterator<Item = &'a str>) -> Result<Url, SourceError> {
+        let mut url = self.endpoint.clone();
+        {
+            let mut path = url
+                .path_segments_mut()
+                .map_err(|_| SourceError::Other("source endpoint cannot carry a path".to_string()))?;
+            path.clear();
+            path.extend(segments);
+        }
+        Ok(url)
+    }
+
+    /// Sends the request and returns the response only for a 2xx status.
+    /// Non-2xx statuses are classified from the status and the provider's own
+    /// error-code header; response bodies are not read, so no provider message
+    /// can smuggle credentials or markup into a log line.
+    pub(super) async fn send(
+        &self,
+        request: reqwest::Request,
+        error_code_header: &str,
+    ) -> Result<reqwest::Response, SourceError> {
+        let response = self.client.execute(request).await.map_err(classify_transport_error)?;
+        let status = response.status();
+        if status.is_success() {
+            return Ok(response);
+        }
+        let code = response
+            .headers()
+            .get(error_code_header)
+            .and_then(|value| value.to_str().ok())
+            .map(str::to_string);
+        Err(classify_status(
+            status.as_u16(),
+            None,
+            match &code {
+                Some(code) => format!("source returned HTTP {status} ({code})"),
+                None => format!("source returned HTTP {status}"),
+            },
+        ))
+    }
+}
+
+/// Renders a transport failure without the request URL: a SAS token or a
+/// signed query would otherwise reach logs and admin responses.
+pub(super) fn classify_transport_error(err: reqwest::Error) -> SourceError {
+    let is_timeout = err.is_timeout();
+    let is_connect = err.is_connect();
+    let message = err.without_url().to_string();
+    if is_timeout {
+        SourceError::Timeout
+    } else if is_connect {
+        SourceError::Connect(message)
+    } else {
+        SourceError::Other(message)
+    }
+}
+
+/// Streams the response body without buffering it.
+pub(super) fn response_body(response: reqwest::Response) -> ByteStream {
+    let stream = response.bytes_stream().map(|chunk| {
+        chunk
+            .map(http_body::Frame::data)
+            .map_err(|err| std::io::Error::other(err.without_url().to_string()))
+    });
+    ByteStream::new(SdkBody::from_body_1_x(http_body_util::StreamBody::new(stream)))
+}
+
+/// Reads a bounded response body as UTF-8, for the XML and JSON listings.
+pub(super) async fn read_text(response: reqwest::Response, max_bytes: usize) -> Result<String, SourceError> {
+    let mut body = Vec::new();
+    let mut stream = response.bytes_stream();
+    while let Some(chunk) = stream.next().await {
+        let chunk = chunk.map_err(classify_transport_error)?;
+        if body.len().saturating_add(chunk.len()) > max_bytes {
+            return Err(SourceError::Other("source listing response exceeded the size limit".to_string()));
+        }
+        body.extend_from_slice(&chunk);
+    }
+    String::from_utf8(body).map_err(|_| SourceError::Other("source listing response is not valid UTF-8".to_string()))
+}
+
+/// Base64 digest (`Content-MD5`, `md5Hash`, `x-goog-hash`) as lowercase hex.
+/// `None` when the value is not a 16-byte digest, so a CRC32C never passes as
+/// an MD5.
+pub(super) fn base64_md5_to_hex(value: &str) -> Option<String> {
+    let raw = base64_simd::STANDARD.decode_to_vec(value.trim().as_bytes()).ok()?;
+    (raw.len() == 16).then(|| faster_hex::hex_string(&raw))
+}
+
+pub(super) fn header<'a>(headers: &'a HeaderMap, name: &str) -> Option<&'a str> {
+    headers.get(name).and_then(|value| value.to_str().ok()).map(str::trim)
+}
+
+fn header_string(headers: &HeaderMap, name: &str) -> Option<String> {
+    header(headers, name).filter(|value| !value.is_empty()).map(str::to_string)
+}
+
+/// `Last-Modified` and friends arrive as an HTTP date; the JSON dialects use
+/// RFC 3339 for the same field, so both are accepted.
+pub(super) fn parse_http_timestamp(value: &str) -> Option<SystemTime> {
+    OffsetDateTime::parse(value, &Rfc2822)
+        .or_else(|_| OffsetDateTime::parse(value, &Rfc3339))
+        .ok()
+        .map(SystemTime::from)
+}
+
+/// Provider-specific fields the shared header mapping cannot infer.
+pub(super) struct NativeHeadFields {
+    pub(super) etag: Option<String>,
+    /// The ETag is an opaque token rather than a digest of the bytes.
+    pub(super) etag_is_opaque: bool,
+    pub(super) version_id: Option<String>,
+    pub(super) storage_class: Option<String>,
+}
+
+/// Maps a HEAD or GET response onto [`SourceHead`]. `metadata_prefix` is the
+/// provider's user-metadata header prefix (`x-ms-meta-`, `x-goog-meta-`); the
+/// stored shape drops it, matching the `x-amz-meta-` handling of the S3 path.
+pub(super) fn native_source_head(
+    headers: &HeaderMap,
+    metadata_prefix: &str,
+    fields: NativeHeadFields,
+) -> Result<SourceHead, SourceError> {
+    let size = header(headers, "content-length")
+        .and_then(|value| value.parse::<u64>().ok())
+        .ok_or_else(|| SourceError::Other("source response has no valid content-length".to_string()))?;
+
+    let mut user_metadata = HashMap::new();
+    for (name, value) in headers {
+        let name = name.as_str();
+        if let Some(key) = name.strip_prefix(metadata_prefix)
+            && !key.is_empty()
+            && let Ok(value) = value.to_str()
+        {
+            user_metadata.insert(key.to_string(), value.to_string());
+        }
+    }
+
+    let etag = fields
+        .etag
+        .map(|etag| etag.trim().trim_matches('"').to_string())
+        .filter(|etag| !etag.is_empty());
+    // An opaque ETag never encodes a part count, so the multipart flag stays
+    // false for it however the provider happens to spell the token.
+    let is_multipart_etag = !fields.etag_is_opaque && etag.as_deref().is_some_and(is_multipart_etag);
+
+    Ok(SourceHead {
+        etag,
+        size,
+        last_modified: header(headers, "last-modified").and_then(parse_http_timestamp),
+        content_type: header_string(headers, "content-type"),
+        content_encoding: header_string(headers, "content-encoding"),
+        content_disposition: header_string(headers, "content-disposition"),
+        content_language: header_string(headers, "content-language"),
+        cache_control: header_string(headers, "cache-control"),
+        expires: header_string(headers, "expires"),
+        user_metadata,
+        version_id: fields.version_id,
+        storage_class: fields.storage_class,
+        // Neither native provider hands back ciphertext: a customer-key object
+        // is refused by the backend before it reaches this mapping, and the
+        // service-managed encryption is transparent to the reader.
+        sse: None,
+        is_multipart_etag,
+        etag_is_opaque: fields.etag_is_opaque,
+    })
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use http::HeaderValue;
+
+    fn headers(pairs: &[(&str, &str)]) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        for (name, value) in pairs {
+            headers.insert(
+                http::HeaderName::from_bytes(name.as_bytes()).expect("test header name"),
+                HeaderValue::from_str(value).expect("test header value"),
+            );
+        }
+        headers
+    }
+
+    fn fields() -> NativeHeadFields {
+        NativeHeadFields {
+            etag: None,
+            etag_is_opaque: false,
+            version_id: None,
+            storage_class: None,
+        }
+    }
+
+    #[test]
+    fn native_source_head_maps_content_headers_and_prefixed_metadata() {
+        let headers = headers(&[
+            ("content-length", "1234"),
+            ("content-type", "text/plain"),
+            ("content-encoding", "gzip"),
+            ("content-language", "en"),
+            ("content-disposition", "attachment"),
+            ("cache-control", "max-age=60"),
+            ("expires", "Thu, 01 Jan 2026 00:00:00 GMT"),
+            ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT"),
+            ("x-ms-meta-owner", "alice"),
+            ("x-goog-meta-owner", "not-mine"),
+        ]);
+        let head = native_source_head(
+            &headers,
+            "x-ms-meta-",
+            NativeHeadFields {
+                etag: Some("\"0x8DCE1D2\"".to_string()),
+                etag_is_opaque: true,
+                version_id: Some("2026-01-01T00:00:00.0000000Z".to_string()),
+                storage_class: Some("Hot".to_string()),
+            },
+        )
+        .expect("head should map");
+
+        assert_eq!(head.size, 1234);
+        assert_eq!(head.content_type.as_deref(), Some("text/plain"));
+        assert_eq!(head.content_encoding.as_deref(), Some("gzip"));
+        assert_eq!(head.content_language.as_deref(), Some("en"));
+        assert_eq!(head.content_disposition.as_deref(), Some("attachment"));
+        assert_eq!(head.cache_control.as_deref(), Some("max-age=60"));
+        assert_eq!(head.expires.as_deref(), Some("Thu, 01 Jan 2026 00:00:00 GMT"));
+        assert_eq!(
+            head.last_modified,
+            Some(SystemTime::UNIX_EPOCH + std::time::Duration::from_secs(1_445_412_480)),
+            "HTTP-date Last-Modified must parse"
+        );
+        assert_eq!(
+            head.user_metadata,
+            HashMap::from([("owner".to_string(), "alice".to_string())]),
+            "only the provider's own metadata prefix is read"
+        );
+        assert_eq!(head.etag.as_deref(), Some("0x8DCE1D2"), "quotes are stripped, the token is kept");
+        assert!(head.etag_is_opaque);
+        assert!(!head.is_multipart_etag);
+        assert_eq!(head.storage_class.as_deref(), Some("Hot"));
+        assert!(head.sse.is_none());
+    }
+
+    #[test]
+    fn native_source_head_requires_a_content_length() {
+        let err = native_source_head(&headers(&[("content-type", "text/plain")]), "x-ms-meta-", fields())
+            .expect_err("a response without content-length is unusable");
+        assert!(matches!(err, SourceError::Other(_)), "{err:?}");
+    }
+
+    #[test]
+    fn opaque_etag_never_reads_as_a_multipart_etag() {
+        // A digest-shaped ETag keeps the S3 reading; the same string marked
+        // opaque must not be split into "digest-partcount".
+        for (opaque, expected) in [(false, true), (true, false)] {
+            let head = native_source_head(
+                &headers(&[("content-length", "1")]),
+                "x-ms-meta-",
+                NativeHeadFields {
+                    etag: Some("d41d8cd98f00b204e9800998ecf8427e-3".to_string()),
+                    etag_is_opaque: opaque,
+                    ..fields()
+                },
+            )
+            .expect("head should map");
+            assert_eq!(head.is_multipart_etag, expected, "opaque = {opaque}");
+        }
+    }
+
+    #[test]
+    fn base64_md5_converts_only_sixteen_byte_digests() {
+        assert_eq!(
+            base64_md5_to_hex("1B2M2Y8AsgTpgAmY7PhCfg==").as_deref(),
+            Some("d41d8cd98f00b204e9800998ecf8427e")
+        );
+        assert_eq!(base64_md5_to_hex("not base64!").as_deref(), None);
+        // A CRC32C digest is four bytes: it must not pass as an MD5.
+        assert_eq!(base64_md5_to_hex("AAAAAA==").as_deref(), None);
+    }
+
+    #[test]
+    fn native_http_rejects_endpoints_that_are_not_bare_origins() {
+        for bad in [
+            "ftp://source.example.com",
+            "https://user:pw@source.example.com",
+            "https://source.example.com/container",
+            "https://source.example.com/?x=1",
+            "not a url",
+        ] {
+            assert!(
+                NativeHttp::new(bad, SourceTimeouts::default(), false, None).is_err(),
+                "{bad} must be rejected"
+            );
+        }
+    }
+
+    #[test]
+    fn native_http_percent_encodes_every_path_segment() {
+        let http = NativeHttp::for_test(Url::parse("https://acct.blob.core.windows.net").expect("origin"));
+        let url = http.url(["container", "dir", "a b?c#d.txt"]).expect("url should build");
+        assert_eq!(url.as_str(), "https://acct.blob.core.windows.net/container/dir/a%20b%3Fc%23d.txt");
+        assert_eq!(url.query(), None, "a key with '?' must not become a query");
+    }
+}

crates/ecstore/src/bucket/on_demand_migration/pull.rs

@@ -1119,6 +1119,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig::default(),
             policy: PolicyConfig::default(),

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -25,6 +25,8 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.
 
+use super::azure::AzureSourceBackend;
+use super::gcs::GcsNativeSourceBackend;
 use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
     PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
@@ -65,6 +67,10 @@ pub enum SourceProvider {
     /// Generic S3-compatible service.
     #[default]
     S3,
+    /// Native Azure Blob service; not an S3 dialect.
+    Azure,
+    /// Native GCS JSON API with a service-account key; not an S3 dialect.
+    GcsNative,
 }
 
 impl SourceProvider {
@@ -76,6 +82,8 @@ impl SourceProvider {
             "minio" => Some(Self::Minio),
             "rustfs" => Some(Self::Rustfs),
             "s3" => Some(Self::S3),
+            "azure" => Some(Self::Azure),
+            "gcs_native" => Some(Self::GcsNative),
             _ => None,
         }
     }
@@ -88,6 +96,8 @@ impl SourceProvider {
             Self::Minio => "minio",
             Self::Rustfs => "rustfs",
             Self::S3 => "s3",
+            Self::Azure => "azure",
+            Self::GcsNative => "gcs_native",
         }
     }
 
@@ -159,6 +169,69 @@ pub struct SourceClientSpec {
     /// Bytes per second the pull pipeline may consume from this source;
     /// `None` means unlimited. Enforced by the consumer, not by this client.
     pub bandwidth_limit: Option<NonZeroU64>,
+    /// Which [`SourceBackend`] to build. The S3 variant reads `region`,
+    /// `path_style` and `credentials`; the native variants ignore all three
+    /// and carry their own credentials.
+    pub backend: SourceBackendSpec,
+}
+
+/// Provider-specific half of [`SourceClientSpec`].
+#[derive(Clone, Debug, Default, PartialEq, Eq)]
+pub enum SourceBackendSpec {
+    #[default]
+    S3,
+    Azure(AzureSourceSpec),
+    Gcs(GcsSourceSpec),
+}
+
+/// Native Azure Blob parameters. The container is [`SourceClientSpec::bucket`].
+#[derive(Clone, PartialEq, Eq)]
+pub struct AzureSourceSpec {
+    pub account: String,
+    pub auth: AzureAuth,
+}
+
+impl fmt::Debug for AzureSourceSpec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("AzureSourceSpec")
+            .field("account", &self.account)
+            .field("auth", &self.auth)
+            .finish()
+    }
+}
+
+/// How Azure requests are authorized.
+#[derive(Clone, PartialEq, Eq)]
+pub enum AzureAuth {
+    /// Base64 storage-account key, signed per request with Shared Key.
+    SharedKey(String),
+    /// SAS query string without the leading `?`, appended to every URL.
+    Sas(String),
+}
+
+impl fmt::Debug for AzureAuth {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // Both variants are secrets; only the scheme may be rendered.
+        f.write_str(match self {
+            Self::SharedKey(_) => "SharedKey(REDACTED)",
+            Self::Sas(_) => "Sas(REDACTED)",
+        })
+    }
+}
+
+/// Native GCS parameters. The bucket is [`SourceClientSpec::bucket`].
+#[derive(Clone, PartialEq, Eq)]
+pub struct GcsSourceSpec {
+    /// Service-account key JSON.
+    pub service_account_json: String,
+}
+
+impl fmt::Debug for GcsSourceSpec {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("GcsSourceSpec")
+            .field("service_account_json", &"REDACTED")
+            .finish()
+    }
 }
 
 impl SourceClientSpec {
@@ -272,7 +345,7 @@ const ACCESS_DENIED_CODES: &[&str] = &[
     "InvalidToken",
 ];
 
-fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError {
+pub(super) fn classify_status(status: u16, code: Option<&str>, message: String) -> SourceError {
     if let Some(code) = code {
         if THROTTLE_CODES.contains(&code) {
             return SourceError::Throttled;
@@ -345,6 +418,11 @@ pub struct SourceHead {
     pub storage_class: Option<String>,
     pub sse: Option<SourceSse>,
     pub is_multipart_etag: bool,
+    /// The provider's ETag is not derived from the object bytes (Azure
+    /// stamps an opaque concurrency token). Such an ETag is recorded for
+    /// provenance but must never be read as a content digest, so the
+    /// write-back path refuses to use it as the expected MD5.
+    pub etag_is_opaque: bool,
 }
 
 /// Per-operation fields shared by HEAD and GET outputs.
@@ -366,7 +444,7 @@ struct HeadParts {
     sse_customer_algorithm: Option<String>,
 }
 
-fn normalize_etag(etag: Option<String>) -> Option<String> {
+pub(super) fn normalize_etag(etag: Option<String>) -> Option<String> {
     etag.map(|etag| etag.trim().trim_matches('"').to_string())
         .filter(|etag| !etag.is_empty())
 }
@@ -415,6 +493,7 @@ fn source_head(parts: HeadParts) -> Result<SourceHead, SourceError> {
         storage_class: parts.storage_class,
         sse,
         is_multipart_etag,
+        etag_is_opaque: false,
     })
 }
 
@@ -625,9 +704,48 @@ impl fmt::Debug for SourceClient {
 
 impl SourceClient {
     pub async fn new(spec: &SourceClientSpec) -> Result<Self, RemoteS3ClientError> {
-        let endpoint = spec.endpoint_spec()?;
-        let config = build_remote_s3_config(&endpoint).await?;
-        Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
+        match &spec.backend {
+            SourceBackendSpec::S3 => {
+                let endpoint = spec.endpoint_spec()?;
+                let config = build_remote_s3_config(&endpoint).await?;
+                Ok(Self::from_config_builder(config, endpoint.endpoint_url(), spec))
+            }
+            SourceBackendSpec::Azure(azure) => {
+                let backend = AzureSourceBackend::new(
+                    &spec.endpoint,
+                    &spec.bucket,
+                    azure,
+                    spec.timeouts,
+                    spec.skip_tls_verify,
+                    spec.ca_cert_pem.as_deref(),
+                )?;
+                Ok(Self::from_backend(Box::new(backend), spec))
+            }
+            SourceBackendSpec::Gcs(gcs) => {
+                let backend = GcsNativeSourceBackend::new(
+                    &spec.endpoint,
+                    &spec.bucket,
+                    gcs,
+                    spec.timeouts,
+                    spec.skip_tls_verify,
+                    spec.ca_cert_pem.as_deref(),
+                )?;
+                Ok(Self::from_backend(Box::new(backend), spec))
+            }
+        }
+    }
+
+    /// Wraps a ready backend in the prefix-mapping client. The endpoint is
+    /// kept only for `Debug` and admin status.
+    fn from_backend(backend: Box<dyn SourceBackend>, spec: &SourceClientSpec) -> Self {
+        Self {
+            backend,
+            endpoint: spec.endpoint.clone(),
+            bucket: spec.bucket.clone(),
+            source_prefix: spec.source_prefix.clone().filter(|prefix| !prefix.is_empty()),
+            timeouts: spec.timeouts,
+            bandwidth_limit: spec.bandwidth_limit,
+        }
     }
 
     /// `config` must come from [`SourceClientSpec::endpoint_spec`], which is
@@ -866,6 +984,7 @@ fn s3_source_object(object: SdkObject) -> Option<SourceObject> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::bucket::on_demand_migration::backend_contract::{BackendCapabilities, OBJECT_MD5, assert_backend_contract};
     use aws_smithy_runtime_api::client::http::{HttpConnector, HttpConnectorFuture, SharedHttpConnector, http_client_fn};
     use aws_smithy_runtime_api::client::orchestrator::HttpRequest;
     use aws_smithy_runtime_api::client::result::ConnectorError;
@@ -983,6 +1102,7 @@ mod tests {
             retry: RemoteS3RetryPolicy::Disabled,
             timeouts: SourceTimeouts::default(),
             bandwidth_limit: NonZeroU64::new(1_000_000),
+            backend: SourceBackendSpec::S3,
         }
     }
 
@@ -1586,7 +1706,101 @@ mod tests {
         assert_eq!(resolve_path_style(PathStyle::VirtualHost, Minio, "10.0.0.1"), PathStyle::VirtualHost);
         assert_eq!(resolve_path_style(PathStyle::Path, Aws, "s3.amazonaws.com"), PathStyle::Path);
         assert_eq!(SourceProvider::from_label(" AWS "), Some(Aws));
-        assert_eq!(SourceProvider::from_label("azure"), None);
+        assert_eq!(SourceProvider::from_label(" Azure "), Some(Azure));
+        assert_eq!(SourceProvider::from_label("gcs_native"), Some(GcsNative));
+        assert_eq!(SourceProvider::from_label("swift"), None);
+    }
+
+    const CONTRACT_LIST_PAGE_ONE: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Name>source-bucket</Name>
+  <IsTruncated>true</IsTruncated>
+  <NextContinuationToken>cursor-1</NextContinuationToken>
+  <Contents>
+    <Key>dir/a.txt</Key>
+    <LastModified>2015-10-21T07:28:00.000Z</LastModified>
+    <ETag>&quot;5d41402abc4b2a76b9719d911017c592&quot;</ETag>
+    <Size>5</Size>
+    <StorageClass>STANDARD</StorageClass>
+  </Contents>
+  <CommonPrefixes><Prefix>dir/sub/</Prefix></CommonPrefixes>
+</ListBucketResult>"#;
+
+    const CONTRACT_LIST_PAGE_TWO: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<ListBucketResult xmlns="http://s3.amazonaws.com/doc/2006-03-01/">
+  <Name>source-bucket</Name>
+  <IsTruncated>false</IsTruncated>
+  <Contents>
+    <Key>dir/b.txt</Key>
+    <LastModified>2015-10-21T07:28:00.000Z</LastModified>
+    <ETag>&quot;7d41402abc4b2a76b9719d911017c592&quot;</ETag>
+    <Size>7</Size>
+  </Contents>
+</ListBucketResult>"#;
+
+    const CONTRACT_TAGGING: &str = r#"<?xml version="1.0" encoding="UTF-8"?>
+<Tagging xmlns="http://s3.amazonaws.com/doc/2006-03-01/"><TagSet>
+  <Tag><Key>env</Key><Value>prod</Value></Tag>
+</TagSet></Tagging>"#;
+
+    fn contract_object_headers(content_length: u64) -> Vec<(&'static str, String)> {
+        vec![
+            ("etag", format!("\"{OBJECT_MD5}\"")),
+            ("content-length", content_length.to_string()),
+            ("content-type", "text/plain".to_string()),
+            ("last-modified", "Wed, 21 Oct 2015 07:28:00 GMT".to_string()),
+            ("x-amz-meta-owner", "alice".to_string()),
+            ("x-amz-storage-class", "STANDARD".to_string()),
+        ]
+    }
+
+    /// The S3 backend behind the scripted connector, without the prefix-mapping
+    /// client on top: the contract is a property of the backend itself.
+    async fn scripted_s3_backend(responses: Vec<Scripted>) -> S3SourceBackend {
+        let spec = spec(None);
+        let connector = SharedHttpConnector::new(ScriptedConnector {
+            requests: Arc::new(Mutex::new(Vec::new())),
+            responses: Arc::new(Mutex::new(responses.into_iter().collect())),
+        });
+        let http_client = http_client_fn(move |_settings, _components| connector.clone());
+        let endpoint = spec.endpoint_spec().expect("test spec endpoint should parse");
+        let config = build_remote_s3_config(&endpoint)
+            .await
+            .expect("test spec should build")
+            .http_client(http_client)
+            .interceptor(SourceProxyMarkerInterceptor::new());
+        S3SourceBackend {
+            client: S3Client::from_conf(config.build()),
+            bucket: spec.bucket.clone(),
+        }
+    }
+
+    #[tokio::test]
+    async fn s3_backend_satisfies_the_shared_backend_contract() {
+        let mut ranged = contract_object_headers(3);
+        ranged.push(("content-range", "bytes 1-3/5".to_string()));
+        let backend = scripted_s3_backend(vec![
+            ok(contract_object_headers(5), ""),
+            ok(contract_object_headers(5), "hello"),
+            ok(ranged, "ell"),
+            ok(Vec::new(), CONTRACT_LIST_PAGE_ONE),
+            ok(Vec::new(), CONTRACT_LIST_PAGE_TWO),
+            ok(Vec::new(), CONTRACT_TAGGING),
+            ok(Vec::new(), ""),
+            status(404, ""),
+            status(403, ACCESS_DENIED_BODY),
+        ])
+        .await;
+
+        assert_backend_contract(
+            &backend,
+            BackendCapabilities {
+                etag_is_opaque: false,
+                supports_start_after: true,
+                supports_tagging: true,
+            },
+        )
+        .await;
     }
 
     fn prefix_client(prefix: Option<String>) -> SourceClient {

crates/ecstore/src/bucket/on_demand_migration/sys.rs

@@ -47,7 +47,10 @@ use super::config::{
 use super::list_through::{SOURCE_LIST_RATE_PER_SEC, SourceListRateLimiter};
 use super::negative_cache::NegativeCache;
 use super::pull::{OdmWriteBack, PullQueue};
-use super::source_client::{SourceClient, SourceClientSpec, SourceError, SourceProvider, SourceTimeouts};
+use super::source_client::{
+    AzureAuth, AzureSourceSpec, GcsSourceSpec, SourceBackendSpec, SourceClient, SourceClientSpec, SourceError, SourceProvider,
+    SourceTimeouts,
+};
 use super::stats::{GaugeGuard, OdmStats, OdmStatsSnapshot, PullFailureReason};
 use crate::bucket::remote_s3_client::{
     PathStyle as ClientPathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
@@ -619,6 +622,7 @@ pub fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClientSpec
         // load on a source that is already failing.
         retry: RemoteS3RetryPolicy::Disabled,
         bandwidth_limit: policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
+        backend: source_backend_spec(source),
     }
 }
 
@@ -630,6 +634,31 @@ fn source_provider(provider: Provider) -> SourceProvider {
         Provider::Rustfs => SourceProvider::Rustfs,
         Provider::R2 => SourceProvider::R2,
         Provider::Gcs => SourceProvider::Gcs,
+        Provider::Azure => SourceProvider::Azure,
+        Provider::GcsNative => SourceProvider::GcsNative,
+    }
+}
+
+/// Which backend the client builds. A native provider whose block is missing
+/// falls back to the S3 spec, where the builder reports the missing
+/// credentials: the config layer already refuses to store that shape, so this
+/// only covers a config written by an older or hand-edited build.
+pub fn source_backend_spec(source: &SourceConfig) -> SourceBackendSpec {
+    match (source.provider, source.azure.as_ref(), source.gcs.as_ref()) {
+        (Provider::Azure, Some(azure), _) => SourceBackendSpec::Azure(AzureSourceSpec {
+            account: azure.account.clone(),
+            auth: match (&azure.account_key, &azure.sas_token) {
+                (Some(key), _) => AzureAuth::SharedKey(key.clone()),
+                (None, Some(sas)) => AzureAuth::Sas(sas.clone()),
+                // Refused by `SourceConfig::validate`; an empty shared key
+                // fails closed at the builder rather than signing with none.
+                (None, None) => AzureAuth::SharedKey(String::new()),
+            },
+        }),
+        (Provider::GcsNative, _, Some(gcs)) => SourceBackendSpec::Gcs(GcsSourceSpec {
+            service_account_json: gcs.service_account_json.clone(),
+        }),
+        _ => SourceBackendSpec::S3,
     }
 }
 
@@ -929,6 +958,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: prefix.map(str::to_string),

crates/ecstore/src/bucket/on_demand_migration/test_http_fixture.rs

@@ -0,0 +1,120 @@
+// 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.
+
+//! Scripted HTTP server for the native source backends' tests.
+//!
+//! The S3 backend can be driven through the SDK's own connector; the native
+//! backends talk to a real socket, so their tests need a server that answers a
+//! fixed script and records what it was asked. Every response closes its
+//! connection, which keeps one request on one socket and makes the script order
+//! exactly the request order.
+
+use std::sync::{Arc, Mutex};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpListener;
+use url::Url;
+
+pub(super) struct ScriptedResponse {
+    status: u16,
+    headers: Vec<(&'static str, String)>,
+    body: String,
+}
+
+impl ScriptedResponse {
+    pub(super) fn new(status: u16, headers: Vec<(&'static str, String)>, body: String) -> Self {
+        Self { status, headers, body }
+    }
+}
+
+#[derive(Clone, Debug)]
+pub(super) struct RecordedRequest {
+    pub(super) method: String,
+    /// Request target as it appeared on the wire: path plus query.
+    pub(super) target: String,
+    pub(super) headers: Vec<(String, String)>,
+}
+
+impl RecordedRequest {
+    pub(super) fn header(&self, name: &str) -> Option<&str> {
+        self.headers
+            .iter()
+            .find(|(key, _)| key.eq_ignore_ascii_case(name))
+            .map(|(_, value)| value.as_str())
+    }
+}
+
+pub(super) type Recorder = Arc<Mutex<Vec<RecordedRequest>>>;
+
+/// Binds a loopback listener that answers `responses` in order and returns its
+/// origin plus the recorder. The task ends once the script is exhausted.
+pub(super) async fn scripted_server(responses: Vec<ScriptedResponse>) -> (Url, Recorder) {
+    let listener = TcpListener::bind("127.0.0.1:0").await.expect("fixture listener should bind");
+    let port = listener.local_addr().expect("fixture address").port();
+    let recorder: Recorder = Arc::new(Mutex::new(Vec::new()));
+    let sink = Arc::clone(&recorder);
+
+    tokio::spawn(async move {
+        for response in responses {
+            let Ok((mut stream, _)) = listener.accept().await else {
+                return;
+            };
+            let mut request = Vec::new();
+            let mut buffer = [0_u8; 2048];
+            while !request.windows(4).any(|window| window == b"\r\n\r\n") {
+                match stream.read(&mut buffer).await {
+                    Ok(0) | Err(_) => break,
+                    Ok(read) => request.extend_from_slice(&buffer[..read]),
+                }
+            }
+            let text = String::from_utf8_lossy(&request).into_owned();
+            let mut lines = text.lines();
+            let start = lines.next().unwrap_or_default().to_string();
+            let mut parts = start.split_whitespace();
+            sink.lock().expect("recorder lock").push(RecordedRequest {
+                method: parts.next().unwrap_or_default().to_string(),
+                target: parts.next().unwrap_or_default().to_string(),
+                headers: lines
+                    .take_while(|line| !line.is_empty())
+                    .filter_map(|line| line.split_once(':'))
+                    .map(|(name, value)| (name.trim().to_string(), value.trim().to_string()))
+                    .collect(),
+            });
+
+            // A scripted HEAD declares the object size in its own headers while
+            // carrying no body, so an explicit `Content-Length` wins over the
+            // body length.
+            let declares_length = response
+                .headers
+                .iter()
+                .any(|(name, _)| name.eq_ignore_ascii_case("content-length"));
+            let mut rendered = match declares_length {
+                true => format!("HTTP/1.1 {} Scripted\r\nConnection: close\r\n", response.status),
+                false => format!(
+                    "HTTP/1.1 {} Scripted\r\nContent-Length: {}\r\nConnection: close\r\n",
+                    response.status,
+                    response.body.len()
+                ),
+            };
+            for (name, value) in response.headers {
+                rendered.push_str(&format!("{name}: {value}\r\n"));
+            }
+            rendered.push_str("\r\n");
+            rendered.push_str(&response.body);
+            let _ = stream.write_all(rendered.as_bytes()).await;
+            let _ = stream.flush().await;
+        }
+    });
+
+    (Url::parse(&format!("http://127.0.0.1:{port}")).expect("fixture endpoint"), recorder)
+}

crates/madmin/fixtures/on_demand_migration/get_response.json

@@ -1 +1 @@
-{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}
+{"bucket":"photos","config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z"}

crates/madmin/fixtures/on_demand_migration/set_request.json

@@ -1 +1 @@
-{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}
+{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"sourceSecretKey123","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}}

crates/madmin/fixtures/on_demand_migration/set_response.json

@@ -1 +1 @@
-{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null}},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}
+{"bucket":"photos","dry_run":false,"config":{"version":1,"enabled":true,"source":{"provider":"minio","endpoint":"https://source.example.com:9000","region":"us-east-1","bucket":"legacy-photos","path_style":"auto","credentials":{"access_key":"AKIASOURCE","secret_key":"REDACTED","session_token":null},"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":null},"filter":{"prefix":null,"source_prefix":"photos/"},"policy":{"head":"proxy","range_get":"serve_and_backfill","source_error":"propagate","list_through":false,"respect_local_delete_marker":true,"preserve_etag":true,"copy_tags":false,"emit_events":true,"negative_cache_ttl_secs":30,"inline_max_bytes":16777216,"multipart_part_size_bytes":67108864,"max_concurrent_pulls":8,"pull_queue_capacity":1024,"source_timeout":{"connect_ms":5000,"first_byte_ms":15000,"idle_ms":30000},"bandwidth_limit_bytes_per_sec":null}},"updated_at":"2026-09-02T10:00:00Z","probe":{"reachable":true,"listable":true,"sample_key":"photos/2024/01.jpg"}}

crates/madmin/src/on_demand_migration.rs

@@ -78,10 +78,18 @@ pub struct OnDemandMigrationSource {
     #[serde(default)]
     pub path_style: OnDemandMigrationPathStyle,
     /// `None` means anonymous access to a public source bucket.
+    /// `None` means anonymous access to a public source bucket. The native
+    /// providers carry their credentials in `azure` / `gcs` instead.
     #[serde(default)]
     pub credentials: Option<OnDemandMigrationCredentials>,
     #[serde(default)]
     pub tls: OnDemandMigrationTls,
+    /// Required for `azure` and rejected for every other provider.
+    #[serde(default)]
+    pub azure: Option<OnDemandMigrationAzure>,
+    /// Required for `gcs_native` and rejected for every other provider.
+    #[serde(default)]
+    pub gcs: Option<OnDemandMigrationGcs>,
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -92,7 +100,49 @@ pub enum OnDemandMigrationProvider {
     Minio,
     Rustfs,
     R2,
+    /// GCS XML interoperability API with HMAC keys.
     Gcs,
+    /// Native Azure Blob service.
+    Azure,
+    /// Native GCS JSON API with a service-account key.
+    #[serde(rename = "gcs_native")]
+    GcsNative,
+}
+
+/// Native Azure Blob parameters. The container is `source.bucket`; exactly one
+/// of `account_key` and `sas_token` is set. Responses carry both as `REDACTED`.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationAzure {
+    pub account: String,
+    #[serde(default)]
+    pub account_key: Option<String>,
+    #[serde(default)]
+    pub sas_token: Option<String>,
+}
+
+impl fmt::Debug for OnDemandMigrationAzure {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("OnDemandMigrationAzure")
+            .field("account", &self.account)
+            .field("account_key", &self.account_key.as_ref().map(|_| "REDACTED"))
+            .field("sas_token", &self.sas_token.as_ref().map(|_| "REDACTED"))
+            .finish()
+    }
+}
+
+/// Native GCS parameters. The bucket is `source.bucket`; the key JSON embeds a
+/// private key, so responses carry it as `REDACTED`.
+#[derive(Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct OnDemandMigrationGcs {
+    pub service_account_json: String,
+}
+
+impl fmt::Debug for OnDemandMigrationGcs {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("OnDemandMigrationGcs")
+            .field("service_account_json", &"REDACTED")
+            .finish()
+    }
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Default, Serialize, Deserialize)]
@@ -806,6 +856,8 @@ mod tests {
                 session_token: None,
             }),
             tls: OnDemandMigrationTls::default(),
+            azure: None,
+            gcs: None,
         });
         let mut expected: OnDemandMigrationConfig = serde_json::from_str(SET_REQUEST_FIXTURE.trim()).expect("fixture");
         expected.filter.source_prefix = None;
@@ -821,6 +873,42 @@ mod tests {
         assert!(minimal.source.credentials.is_none());
     }
 
+    #[test]
+    fn native_provider_documents_round_trip_and_hide_their_secrets() {
+        for (label, json) in [
+            (
+                "azure",
+                r#"{"provider":"azure","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":{"account":"legacyaccount","account_key":null,"sas_token":"sv=2021-08-06&sig=topsecret"},"gcs":null}"#,
+            ),
+            (
+                "gcs_native",
+                r#"{"provider":"gcs_native","endpoint":null,"region":"auto","bucket":"legacy-photos","path_style":"auto","credentials":null,"tls":{"skip_verify":false,"ca_cert_pem":null},"azure":null,"gcs":{"service_account_json":"{\"type\":\"service_account\"}"}}"#,
+            ),
+        ] {
+            let source: OnDemandMigrationSource = serde_json::from_str(json).unwrap_or_else(|err| panic!("{label}: {err}"));
+            assert_eq!(
+                serde_json::to_string(&source).expect("re-encodes"),
+                json,
+                "{label} must reproduce the server wire shape byte for byte"
+            );
+        }
+
+        let azure = OnDemandMigrationAzure {
+            account: "legacyaccount".to_string(),
+            account_key: Some("c2VjcmV0".to_string()),
+            sas_token: Some("sig=topsecret".to_string()),
+        };
+        let rendered = format!("{azure:?}");
+        assert!(rendered.contains("legacyaccount"));
+        assert!(!rendered.contains("c2VjcmV0"), "{rendered}");
+        assert!(!rendered.contains("topsecret"), "{rendered}");
+
+        let gcs = OnDemandMigrationGcs {
+            service_account_json: r#"{"private_key":"-----BEGIN PRIVATE KEY-----"}"#.to_string(),
+        };
+        assert!(!format!("{gcs:?}").contains("PRIVATE KEY"), "{gcs:?}");
+    }
+
     #[test]
     fn credentials_debug_never_prints_secrets() {
         let credentials = OnDemandMigrationCredentials {

docs/operations/on-demand-migration.md

@@ -91,14 +91,20 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
 |---|---|---|---|
 | `version` | integer | `1` | Must be `1` |
 | `enabled` | bool | `true` | `false` keeps the config but stops all source traffic |
-| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` | — (required) | Drives endpoint and addressing defaults |
-| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required for every provider except `aws`, where it is derived from `region` |
-| `source.region` | string | — (required) | Non-empty. `auto` is accepted only for `r2`, `minio`, `rustfs` and is signed as `us-east-1` |
-| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace |
+| `source.provider` | `s3` \| `aws` \| `minio` \| `rustfs` \| `r2` \| `gcs` \| `azure` \| `gcs_native` | — (required) | Drives endpoint and addressing defaults, and which backend the client builds: every value but `azure` and `gcs_native` speaks S3 |
+| `source.endpoint` | string \| null | — | `http(s)://host[:port]`, no path, query, fragment or userinfo. Required except for `aws` (derived from `region`), `azure` (derived as `https://<account>.blob.core.windows.net`) and `gcs_native` (`https://storage.googleapis.com`). Set it explicitly to point at Azurite or fake-gcs-server, subject to the same outbound policy as any other source endpoint |
+| `source.region` | string | — (required) | Non-empty. `auto` is accepted for `r2`, `minio`, `rustfs` and for the native providers, and is signed as `us-east-1`. `azure` and `gcs_native` never sign with a region, so `auto` is the honest value there |
+| `source.bucket` | string | — (required) | Non-empty, no `/` and no whitespace. For `azure` this is the container name, for `gcs_native` the bucket name; the provider block never repeats it |
 | `source.path_style` | `auto` \| `path` \| `virtual` | `auto` | `auto` resolves to path-style for IP-literal or `localhost` endpoints and for `s3`/`minio`/`rustfs`; virtual-host for `aws`/`gcs`/`r2` |
-| `source.credentials` | object \| null | `null` | `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present |
+| `source.credentials` | object \| null | `null` | Read only by the S3 providers; `azure` and `gcs_native` must leave it `null` and carry their credentials in their own block. `null` means anonymous, which the client builder does not support yet: the admin `PUT` refuses it with `InvalidArgument`, and a config that reached the metadata another way resolves as unavailable. `access_key` and `secret_key` must be non-empty; `session_token` is optional but must be non-empty when present |
 | `source.tls.skip_verify` | bool | `false` | Disables certificate verification for the source connection |
 | `source.tls.ca_cert_pem` | string \| null | `null` | Must contain `-----BEGIN CERTIFICATE-----` |
+| `source.azure` | object \| null | `null` | Required for `provider = "azure"` and rejected for every other provider |
+| `source.azure.account` | string | — (required) | Storage account name; `[A-Za-z0-9-]` only, because it becomes the first label of the derived host |
+| `source.azure.account_key` | string \| null | `null` | Base64 storage-account key, signed per request with Shared Key. Mutually exclusive with `sas_token`; exactly one of the two is required |
+| `source.azure.sas_token` | string \| null | `null` | SAS query string without the leading `?` and without whitespace, appended to every request URL |
+| `source.gcs` | object \| null | `null` | Required for `provider = "gcs_native"` and rejected for every other provider |
+| `source.gcs.service_account_json` | string | — (required) | Service-account key JSON; must parse and carry `type: service_account`, `client_email` and `private_key`. Tokens are minted read-only (`devstorage.read_only`) |
 | `filter.prefix` | string \| null | `null` | Null or non-empty. Only local keys with this prefix consult the source |
 | `filter.source_prefix` | string \| null | `null` | Null or non-empty. Prepended to the local key to form the source key |
 | `policy.head` | `proxy` \| `local_only` | `proxy` | `local_only` answers a HEAD miss with 404 and no source traffic |
@@ -107,7 +113,7 @@ The persisted blob is `on-demand-migration.json` in the bucket's metadata. Unkno
 | `policy.list_through` | bool | `false` | Merges the source listing into `ListObjectsV2` so clients see the whole namespace during the migration. Off by default: it puts the source in the path of every listing |
 | `policy.respect_local_delete_marker` | bool | `true` | A local delete marker is the final answer; only a versioned bucket can produce one |
 | `policy.preserve_etag` | bool | `true` | Keeps the source ETag on the stored object unless the bucket encrypts by default |
-| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull |
+| `policy.copy_tags` | bool | `false` | Copies source object tags; needs `s3:GetObjectTagging` and costs one extra source call per inline pull. `azure` reads blob tags instead; `gcs_native` has no tags and always finds none |
 | `policy.emit_events` | bool | `true` | Whether a write-back emits `ObjectCreated` notifications |
 | `policy.negative_cache_ttl_secs` | integer | `30` | `0..=3600`; `0` disables the negative cache |
 | `policy.inline_max_bytes` | integer | `16777216` (16 MiB) | `0..=268435456` (256 MiB). At or below this size a GET miss is teed inline; above it the response streams through and a background pull stores the object |
@@ -133,8 +139,14 @@ Validation also rejects two shapes outright: a source whose endpoint and bucket
 | `rustfs` | Required | Path-style | `auto` allowed | A RustFS source answers the migration request locally thanks to the anti-loop marker | `real_source_test.rs` in the `e2e-nightly` lane |
 | `r2` | `https://<account-id>.r2.cloudflarestorage.com` | Virtual-host | `auto` allowed (signed as `us-east-1`) | | `cloud-source (r2)`, only while `ODM_INTEROP_R2_*` are configured; no difference recorded yet |
 | `gcs` | `https://storage.googleapis.com` | Virtual-host | Real region required | Uses the GCS XML interoperability API with an HMAC key pair, not a service-account JSON key | `cloud-source (gcs)`, only while `ODM_INTEROP_GCS_HMAC_*` are configured; no difference recorded yet |
+| `azure` | Optional; derived as `https://<account>.blob.core.windows.net` | Native Blob REST, not S3 | Unused; write `auto` | Needs `source.azure`; the container is `source.bucket`. Reads need `Read` on the blob and `List` on the container, plus `Tags` when `policy.copy_tags` is on | None yet: no interop job covers Azure |
+| `gcs_native` | Optional; derived as `https://storage.googleapis.com` | Native GCS API, not S3 | Unused; write `auto` | Needs `source.gcs`. Reads use the XML API for objects and `objects.list` for listings, both with an OAuth token minted from the service-account key; the key needs `storage.objects.get` and `storage.objects.list` | None yet: no interop job covers native GCS |
 
-Azure Blob has no preset; a native provider is deferred (rustfs/backlog#2166).
+Every backend answers the same trait contract, pinned by `backend_contract.rs` in `crates/ecstore/src/bucket/on_demand_migration/`, and the three differences that contract allows are the ones documented here.
+
+`azure` differs in two of them. Its ETag is a concurrency token rather than a digest of the bytes, so it is stored as `odm-source-etag` provenance and never used as the expected MD5 of a pulled object — the write-back integrity check falls back to the local digest. And its listing paginates only with an opaque marker: there is no "start after this key" form, so a caller that asks for one gets `Unsupported` instead of a listing that silently starts over.
+
+`gcs_native` differs in the other two. Its listing also has no exclusive "start after" form (`startOffset` is inclusive), so it refuses one the same way. And GCS has no object tagging at all: `policy.copy_tags` finds no tags rather than failing the pull, because GCS custom metadata is already carried by the head mapping. Its ETag is normally usable: the `x-goog-hash` MD5 is converted to hex and checked against the pulled bytes, except on a composite object, which has no MD5 and whose ETag is then treated as opaque.
 
 The "Interop evidence" column names the job in `.github/workflows/on-demand-migration-interop.yml` (rustfs/backlog#2167) that last exercised the preset against a real implementation, and is where a provider difference belongs once the lane finds one. That lane is report-only and scheduled: it runs `crates/e2e_test/src/on_demand_migration/interop_test.rs` — the same case bodies as the merge-gate suite, with the source injected through `RUSTFS_ODM_INTEROP_*` — against a pinned MinIO container, and against each cloud provider whose repository secrets are configured. A provider without secrets is skipped with a note in the run summary rather than failing, so "no difference recorded yet" means exactly that and not "verified clean"; see [ci-gates.md](../testing/ci-gates.md) for the row.
 

docs/testing/ci-gates.md

@@ -91,12 +91,6 @@ Scheduled lanes never block a PR. Their workflow-local gate fails the run, sched
 
 Manual `workflow_dispatch` runs are debugging evidence and do not open scheduled-failure issues. A manual performance run may explicitly allow a known regression; that override is not a passing baseline.
 
-## Packaged functional acceptance
-
-`rustfs-functional-chain.yml` dispatches the packaged-build suites in `rustfs-*-test.yml` on the shared lab runners. A failing suite step or job must fail its workflow. Report collection, cleanup, and dispatch of the next suite can still run with `always()`; continuing diagnostics does not make the failed suite successful.
-
-Workflow status preserves errors that the test scripts report. It does not establish complete execution or a common package identity across the chain: inspect the current run's case results, package identity, and test-script revision as well. A script that returns zero after a failed tool invocation needs its own result check.
-
 ## Release validation
 
 Post-merge and tag-driven; not a substitute for a PR gate.

rustfs/src/admin/handlers/on_demand_migration.rs

@@ -46,6 +46,7 @@ use crate::admin::storage_api::bucket::on_demand_migration::source_client::{
 };
 use crate::admin::storage_api::bucket::on_demand_migration::{
     OdmBucketSnapshot, OnDemandMigrationConfig, OnDemandMigrationConfigError, OnDemandMigrationSys, PathStyle, ValidationContext,
+    source_backend_spec,
 };
 use crate::admin::storage_api::bucket::remote_s3_client::{
     PathStyle as RemotePathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3RetryPolicy,
@@ -585,6 +586,8 @@ fn source_provider(config: &OnDemandMigrationConfig) -> SourceProvider {
         Provider::Rustfs => SourceProvider::Rustfs,
         Provider::R2 => SourceProvider::R2,
         Provider::Gcs => SourceProvider::Gcs,
+        Provider::Azure => SourceProvider::Azure,
+        Provider::GcsNative => SourceProvider::GcsNative,
     }
 }
 
@@ -621,6 +624,9 @@ pub(crate) fn source_client_spec(config: &OnDemandMigrationConfig) -> SourceClie
         // a flapping source behind a success and triple the probe's cost.
         retry: RemoteS3RetryPolicy::Disabled,
         bandwidth_limit: config.policy.bandwidth_limit_bytes_per_sec.and_then(NonZeroU64::new),
+        // One mapping serves the probe and the runtime, so an admin probe
+        // always exercises the backend the runtime will build.
+        backend: source_backend_spec(source),
     }
 }
 

rustfs/src/admin/storage_api.rs

@@ -292,6 +292,7 @@ pub(crate) mod on_demand_migration {
     pub(crate) type PathStyle = super::ecstore_bucket::on_demand_migration::PathStyle;
     pub(crate) type Provider = super::ecstore_bucket::on_demand_migration::Provider;
     pub(crate) type ValidationContext<'a> = super::ecstore_bucket::on_demand_migration::ValidationContext<'a>;
+    pub(crate) use super::ecstore_bucket::on_demand_migration::source_backend_spec;
 
     pub(crate) mod backfill {
         pub(crate) type BackfillCheckpoint = super::super::ecstore_bucket::on_demand_migration::backfill::BackfillCheckpoint;

rustfs/src/app/object/get.rs

@@ -4821,6 +4821,8 @@ mod on_demand_migration_tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: None,

rustfs/src/app/object/head.rs

@@ -665,6 +665,8 @@ mod tests {
                     session_token: None,
                 }),
                 tls: TlsConfig::default(),
+                azure: None,
+                gcs: None,
             },
             filter: FilterConfig {
                 prefix: None,
@@ -743,6 +745,7 @@ mod tests {
                 },
             ),
             is_multipart_etag: true,
+            etag_is_opaque: false,
         }
     }
 

rustfs/src/app/object/on_demand_migration_put.rs

@@ -115,6 +115,12 @@ pub(super) fn expected_md5_hex(head: &SourceHead) -> Option<String> {
     if head.sse.is_some() {
         return None;
     }
+    // Azure stamps an opaque concurrency token in the ETag slot. It is
+    // recorded as provenance, but reading it as a digest would compare the
+    // pulled bytes against a value that never described them.
+    if head.etag_is_opaque {
+        return None;
+    }
     let etag = head.etag.as_deref()?;
     if etag.len() != 32 || is_multipart_etag(etag) || !etag.bytes().all(|byte| byte.is_ascii_hexdigit()) {
         return None;
@@ -874,6 +880,12 @@ mod tests {
         head.sse = None;
         head.etag = None;
         assert_eq!(expected_md5_hex(&head), None);
+
+        // An Azure ETag can be any string the service chooses; even one that
+        // happens to look like an MD5 must not be checked against the bytes.
+        let mut head = source_head(b"abc");
+        head.etag_is_opaque = true;
+        assert_eq!(expected_md5_hex(&head), None, "opaque provider ETag");
     }
 
     #[test]

rustfs/src/site_replication/tests.rs

@@ -867,6 +867,7 @@ fn test_retry_drain_bounds_each_peer_round_to_one_small_request_chain() {
             r#type: "tags".to_string(),
             ..Default::default()
         }],
+        ..Default::default()
     };
     let make = RetryDrainAction::BucketOpReplay {
         operation: SITE_REPLICATION_BUCKET_OP_MAKE_WITH_VERSIONING.to_string(),
@@ -972,7 +973,7 @@ fn test_lightweight_bucket_retry_plan_orders_real_metadata_and_counts_it() {
     operator_replication.rules.push(operator_rule("operator-backup"));
     let mut bucket_with_operator_rule = bucket;
     bucket_with_operator_rule.replication_config =
-        Some(BASE64_STANDARD.encode_to_string(serialize(&operator_replication).expect("operator replication config")));
+        Some(BASE64_STANDARD.encode_to_string(&serialize(&operator_replication).expect("operator replication config")));
     let plan = site_replication_bucket_retry_plan_from_info(&bucket_with_operator_rule, false).expect("targeted retry plan");
     assert!(
         plan.bucket_items.iter().any(|item| item.r#type == "replication-config"),
@@ -1049,7 +1050,7 @@ fn test_reachable_probe_promotion_is_fenced_by_the_observed_event() {
         .peers
         .insert("remote".to_string(), peer("remote", "https://remote.example.com"));
 
-    assert_eq!(mark_reachable_deferred_retry_events(&mut state, std::slice::from_ref(&recovered)), 1);
+    assert_eq!(mark_reachable_deferred_retry_events(&mut state, &[recovered.clone()]), 1);
     assert_eq!(state.retry_queue[0].updated_at, None);
     assert!(!state.retry_queue[0].peer_unreachable);
     assert_eq!(

scripts/functional_case_report.py

@@ -1,77 +0,0 @@
-#!/usr/bin/env python3
-"""Preserve every functional case execution and its suite context in reports."""
-
-from __future__ import annotations
-
-import argparse
-from pathlib import Path
-import re
-
-
-def generate_report(log_file: Path, case_file: Path, matrix_file: Path | None = None) -> bool:
-    ansi = re.compile(r"\x1b\[[0-9;]*m")
-    start_re = re.compile(r"^---\s+([A-Z][A-Z0-9]*-[0-9]+)\s+(.+?)\s+---$")
-    done_re = re.compile(r"^\[(PASS|FAIL|UNSUPPORTED)\]\s+([A-Z][A-Z0-9]*-[0-9]+)\b")
-    context_re = re.compile(r"^(?:\[INFO\]\s+)?==\s+((?:topology|suite):.+?)\s+==$")
-    topo_re = re.compile(r"^\[UPG-TOPO\]\s+(\S+)\s+(\S+)\s+(\S+)\s+(\S+)\s+PASS=(\d+)\s+FAIL=(\d+)\s*$")
-    rows = []
-    pending = {}
-    topo_rows = []
-    context = "context not recorded"
-    complete = True
-    try:
-        with log_file.open(encoding="utf-8", errors="replace") as log:
-            for raw in log:
-                line = ansi.sub("", raw).strip()
-                if match := context_re.match(line):
-                    context = match[1]
-                    pending.clear()
-                elif match := topo_re.match(line):
-                    topo_rows.append(match.groups())
-                elif match := start_re.match(line):
-                    case_id, name = match.groups()
-                    pending[case_id] = len(rows)
-                    rows.append([case_id, f"{name} ({context})", "RUNNING"])
-                elif match := done_re.match(line):
-                    status, case_id = match.groups()
-                    index = pending.pop(case_id, None)
-                    if index is None:
-                        complete = False
-                        rows.append([case_id, f"{case_id} ({context}; start not recorded)", status])
-                    else:
-                        rows[index][2] = status
-    except FileNotFoundError:
-        pass
-
-    counts = {status: sum(row[2] == status for row in rows) for status in ("PASS", "FAIL", "UNSUPPORTED", "RUNNING")}
-    with case_file.open("w", encoding="utf-8") as out:
-        out.write(f"## Case Summary\n\n- Total: {len(rows)}\n")
-        for status, count in counts.items():
-            out.write(f"- {status}: {count}\n")
-        out.write("\n| Case | Name | Status |\n| --- | --- | --- |\n")
-        for row in rows:
-            out.write("| " + " | ".join(value.replace("|", "|") for value in row) + " |\n")
-        if not rows:
-            out.write("\nNo case execution was recorded; the log is missing, empty, or stopped before the cases.\n")
-
-    valid = complete and bool(rows) and not counts["FAIL"] and not counts["RUNNING"]
-    if matrix_file is not None:
-        with matrix_file.open("w", encoding="utf-8") as out:
-            out.write("## Upgrade Matrix\n\n| Topology | KMS Backend | From Version | To Version | Result |\n")
-            out.write("| --- | --- | --- | --- | --- |\n")
-            for topo, backend, old_v, new_v, npass, nfail in topo_rows:
-                result = "PASS" if nfail == "0" else "FAIL"
-                out.write(f"| {topo} | {backend} | {old_v} | {new_v} | {result} (PASS={npass} FAIL={nfail}) |\n")
-            if not topo_rows:
-                out.write("| - | - | - | - | NOT RUN (suite failed before upgrade) |\n")
-        valid = valid and bool(topo_rows) and all(row[-1] == "0" for row in topo_rows)
-    return valid
-
-
-if __name__ == "__main__":
-    parser = argparse.ArgumentParser(description=__doc__)
-    parser.add_argument("log_file", type=Path)
-    parser.add_argument("case_file", type=Path)
-    parser.add_argument("matrix_file", type=Path, nargs="?")
-    args = parser.parse_args()
-    raise SystemExit(0 if generate_report(args.log_file, args.case_file, args.matrix_file) else 1)

scripts/test_security_workflow.py

@@ -1,18 +1,16 @@
 #!/usr/bin/env python3
-"""Exercise functional workflow failures and security evidence without remote VMs."""
+"""Run the security workflow's evidence and result steps without remote VMs."""
 
 from __future__ import annotations
 
 import os
 import re
 import subprocess
-import sys
 import tempfile
 import unittest
 from pathlib import Path
 
 from check_test_wiring import yaml_block
-from functional_case_report import generate_report
 
 
 ROOT = Path(__file__).resolve().parents[1]
@@ -20,61 +18,16 @@ WORKFLOW = ROOT / ".github/workflows/rustfs-security-test.yml"
 CASE_ROW = "| IAM-101 | user CRUD lifecycle | PASS |"
 
 
-def named_steps(job: list[str]) -> dict[str, list[str]]:
-    starts = [i for i, line in enumerate(job) if line.startswith("      - name: ")]
-    return {
-        job[start].split(": ", 1)[1].strip('"'): job[start:end]
-        for start, end in zip(starts, starts[1:] + [len(job)])
-    }
-
-
-def shell_body(lines: list[str]) -> str:
-    start = lines.index("        run: |") + 1
-    shell_lines = []
-    for line in lines[start:]:
-        if line.strip() and not line.startswith("          "):
-            break
-        shell_lines.append(line[10:])
-    if not shell_lines:
-        raise ValueError("missing literal shell body")
-    return "\n".join(shell_lines)
-
-
-class WorkflowSteps:
-    def render(self, value: str) -> str:
-        return re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: self.context[match[1]], value)
-
-    def step_env(self, lines: list[str], indent: int = 8) -> dict[str, str]:
-        result = {}
-        for line in yaml_block(lines, "env", indent) or []:
-            if line.strip() and not line.lstrip().startswith("#"):
-                key, value = line.strip().split(": ", 1)
-                result[key] = self.render(value.strip("'\""))
-        return result
-
-    def run_step(self, name: str) -> subprocess.CompletedProcess[str]:
-        lines = self.steps[name]
-        result = subprocess.run(
-            ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.render(shell_body(lines))],
-            cwd=self.directory, env={**self.env, **self.step_env(lines)}, capture_output=True, text=True,
-        )
-        for line in lines:
-            if line.startswith("        id: "):
-                self.context[f"steps.{line.split(': ', 1)[1]}.outcome"] = "failure" if result.returncode else "success"
-        if Path(self.env["GITHUB_ENV"]).exists():
-            for line in Path(self.env["GITHUB_ENV"]).read_text().splitlines():
-                key, value = line.split("=", 1)
-                self.env[key] = value
-                self.context[f"env.{key}"] = value
-        return result
-
-
-class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
+class SecurityWorkflowTests(unittest.TestCase):
     def setUp(self) -> None:
         self.source = WORKFLOW.read_text()
         self.job = yaml_block(self.source.splitlines(), "security-test", 2)
         self.assertIsNotNone(self.job)
-        self.steps = named_steps(self.job)
+        starts = [i for i, line in enumerate(self.job) if line.startswith("      - name: ")]
+        self.steps = {
+            self.job[start].split(": ", 1)[1].strip('"'): self.job[start:end]
+            for start, end in zip(starts, starts[1:] + [len(self.job)])
+        }
         self.temp = tempfile.TemporaryDirectory()
         self.addCleanup(self.temp.cleanup)
         self.directory = Path(self.temp.name)
@@ -117,6 +70,40 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
             'exit "$FAKE_EXIT"\n'
         )
 
+    def render(self, value: str) -> str:
+        return re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: self.context[match[1]], value)
+
+    def step_env(self, lines: list[str], indent: int = 8) -> dict[str, str]:
+        result = {}
+        for line in yaml_block(lines, "env", indent) or []:
+            if line.strip() and not line.lstrip().startswith("#"):
+                key, value = line.strip().split(": ", 1)
+                result[key] = self.render(value.strip("'\""))
+        return result
+
+    def run_step(self, name: str) -> subprocess.CompletedProcess[str]:
+        lines = self.steps[name]
+        start = lines.index("        run: |") + 1
+        shell_lines = []
+        for line in lines[start:]:
+            if line.strip() and not line.startswith("          "):
+                break
+            shell_lines.append(line[10:])
+        self.assertTrue(shell_lines, f"missing literal shell body: {name}")
+        result = subprocess.run(
+            ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", self.render("\n".join(shell_lines))],
+            cwd=self.directory, env={**self.env, **self.step_env(lines)}, capture_output=True, text=True,
+        )
+        for line in lines:
+            if line.startswith("        id: "):
+                self.context[f"steps.{line.split(': ', 1)[1]}.outcome"] = "failure" if result.returncode else "success"
+        if Path(self.env["GITHUB_ENV"]).exists():
+            for line in Path(self.env["GITHUB_ENV"]).read_text().splitlines():
+                key, value = line.split("=", 1)
+                self.env[key] = value
+                self.context[f"env.{key}"] = value
+        return result
+
     def test_workflow_wiring(self) -> None:
         names = list(self.steps)
         self.assertLess(names.index("Checkout repository (for the OIDC live gate script)"), names.index("Checkout auto-testing scripts (with retry)"))
@@ -206,391 +193,5 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
         self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
 
 
-class FunctionalWorkflowTests(unittest.TestCase):
-    JOBS = {
-        "kms": "kms-test", "storage": "storage-test", "s3-compat": "s3-compat-test",
-        "upgrade": "upgrade-test", "replication": "replication-test", "heal": "heal-test",
-        "tier": "tier-test", "pool-expand": "pool-expansion-test", "performance": "performance-test",
-    }
-    DIRECT_TESTS = {
-        "kms": "Run KMS suite", "storage": "Run storage engine suite",
-        "s3-compat": "Run S3 compatibility suite", "upgrade": "Run upgrade compatibility suite",
-        "replication": "Run replication suite",
-    }
-
-    def test_failure_and_always_step_wiring(self) -> None:
-        for suite, job_id in self.JOBS.items():
-            with self.subTest(suite=suite):
-                source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
-                job = yaml_block(source.splitlines(), job_id, 2)
-                self.assertIsNotNone(job)
-                self.assertNotRegex("\n".join(job), r'''(?m)^    ["']?continue-on-error["']?\s*:''')
-                steps = named_steps(job)
-                if suite in self.DIRECT_TESTS:
-                    test = steps[self.DIRECT_TESTS[suite]]
-                    self.assertNotRegex("\n".join(test), r'''(?m)^        ["']?continue-on-error["']?\s*:''')
-                    self.assertIn("        if: ${{ always() && steps.evidence.outcome == 'success' }}", steps["Generate report"])
-                cleanup = steps["Reset test environment (after)" if suite == "performance" else "Cleanup environment (after)"]
-                condition = next(line.strip() for line in cleanup if line.startswith("        if:"))
-                self.assertIn(condition, (
-                    "if: always()",
-                    "if: ${{ always() && inputs.cleanup_after != 'false' }}",
-                    "if: ${{ always() && (inputs.cleanup_after != 'false' || github.event_name != 'workflow_dispatch') }}",
-                ))
-                if suite != "performance":
-                    handoff = steps["Chain complete"] if suite == "replication" else next(
-                        value for name, value in steps.items() if name.startswith("Continue functional chain")
-                    )
-                    self.assertIn("        if: ${{ always() && github.event_name == 'repository_dispatch' }}", handoff)
-
-    def test_failed_suite_preserves_exit_and_cleanup_and_dispatch_execute(self) -> None:
-        for suite, test_name in self.DIRECT_TESTS.items():
-            with self.subTest(suite=suite), tempfile.TemporaryDirectory() as directory:
-                root = Path(directory)
-                (root / "auto-testing").mkdir()
-                script = root / f"auto-testing/rustfs-{suite}-test.sh"
-                script.write_text('#!/bin/sh\nprintf "partial suite diagnostics\\n"\nexit 17\n')
-                script.chmod(0o755)
-                fake_bin = root / "bin"
-                fake_bin.mkdir()
-                for command, marker in (("ssh", "cleanup"), ("gh", "dispatch")):
-                    fake = fake_bin / command
-                    fake.write_text(f'#!/bin/sh\nprintf "{marker}\\n" >> "$EXECUTED"\n')
-                    fake.chmod(0o755)
-                env = {
-                    **os.environ, "PATH": f"{fake_bin}{os.pathsep}{os.environ['PATH']}",
-                    "EXECUTED": str(root / "executed"), "RUSTFS_NODES": "fixture-node",
-                    "RUSTFS_SSH_USER": "fixture-user", "RUSTFS_NIGHTLY_PACKAGE_URL": "https://example.invalid/package.deb",
-                    "GH_TOKEN": "local-fixture", "GITHUB_EVENT_NAME": "repository_dispatch", "GITHUB_RUN_ID": "314159",
-                }
-                source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
-                steps = named_steps(yaml_block(source.splitlines(), self.JOBS[suite], 2))
-                context = {"github.event_name": "repository_dispatch", "steps.test.outcome": "failure"}
-                for expression in re.findall(r"\$\{\{\s*(.*?)\s*\}\}", source):
-                    if expression.startswith("inputs.") and re.fullmatch(r"inputs\.\w+", expression):
-                        context[expression] = ""
-                def execute(name):
-                    lines = steps[name]
-                    rendered = re.sub(r"\$\{\{\s*(.*?)\s*\}\}", lambda match: context[match[1]], shell_body(lines))
-                    return subprocess.run(
-                        ["bash", "--noprofile", "--norc", "-e", "-o", "pipefail", "-c", rendered],
-                        cwd=root, env={**env, "LOG_FILE": str(root / "suite.log")}, capture_output=True, text=True,
-                    )
-                failed = execute(test_name)
-                self.assertEqual(failed.returncode, 17, failed.stderr)
-                self.assertIn("partial suite diagnostics", failed.stdout)
-                cleanup = execute("Cleanup environment (after)")
-                self.assertEqual(cleanup.returncode, 0, cleanup.stderr)
-                handoff_name = "Chain complete" if suite == "replication" else next(
-                    name for name in steps if name.startswith("Continue functional chain")
-                )
-                handoff = execute(handoff_name)
-                self.assertEqual(handoff.returncode, 0, handoff.stderr)
-                markers = (root / "executed").read_text().splitlines()
-                self.assertEqual(markers, ["cleanup"] if suite == "replication" else ["cleanup", "dispatch"])
-
-
-class FunctionalCaseReportTests(unittest.TestCase):
-    def report(self, text: str | None, matrix: bool = False) -> tuple[bool, str, str]:
-        with tempfile.TemporaryDirectory() as directory:
-            root = Path(directory)
-            log = root / "suite.log"
-            if text is not None:
-                log.write_text(text)
-            valid = generate_report(log, root / "cases.md", root / "matrix.md" if matrix else None)
-            return valid, (root / "cases.md").read_text(), (root / "matrix.md").read_text() if matrix else ""
-
-    def test_repeated_case_executions_preserve_failure_and_context(self):
-        # log() from rustfs/auto-testing@6120aa0a76de, rustfs-kms-test.sh:131.
-        log = subprocess.check_output(["bash", "-c", r'''
-log() { printf '\033[1;36m[INFO]\033[0m %s\n' "$*"; }
-log '== topology: single-single  kms-backend: local =='
-printf '\033[32m--- KMS-101 roundtrip ---\033[0m\n[FAIL] KMS-101\n'
-log '== topology: single-multi  kms-backend: vault-kv2 =='
-printf '%s\n' '--- KMS-101 roundtrip ---' '[PASS] KMS-101'
-printf '%s\n' '--- KMS-101 roundtrip ---' '[UNSUPPORTED] KMS-101'
-'''], text=True)
-        valid, cases, _ = self.report(log)
-        self.assertFalse(valid)
-        self.assertEqual(cases.count("| KMS-101 |"), 3)
-        self.assertIn("- Total: 3\n- PASS: 1\n- FAIL: 1\n- UNSUPPORTED: 1\n- RUNNING: 0\n", cases)
-        self.assertIn("roundtrip (topology: single-single  kms-backend: local) | FAIL |", cases)
-        self.assertIn("roundtrip (topology: single-multi  kms-backend: vault-kv2) | PASS |", cases)
-        self.assertNotIn("\\n", cases)
-
-    def test_missing_empty_unfinished_and_orphan_results_are_not_success(self):
-        for text in (None, "", "setup failed\n", "--- KMS-101 roundtrip ---\n", "[PASS] KMS-101\n",
-                     "--- KMS-101 first ---\n--- KMS-101 second ---\n[PASS] KMS-101\n",
-                     "--- KMS-101 first ---\n[FAIL] KMS-101\n[PASS] KMS-101\n"):
-            with self.subTest(log=text):
-                valid, cases, _ = self.report(text)
-                self.assertFalse(valid)
-                self.assertIn("## Case Summary", cases)
-        valid, cases, _ = self.report("[INFO] == suite: bucket replication (REP-*) ==\n--- REP-101 unsupported ---\n[UNSUPPORTED] REP-101\n")
-        self.assertTrue(valid)
-        self.assertIn("suite: bucket replication", cases)
-        self.assertIn("- UNSUPPORTED: 1\n", cases)
-
-    def test_upgrade_matrix_is_preserved_and_required_for_complete_report(self):
-        case = "--- UPG-101 upgrade ---\n[PASS] UPG-101\n"
-        for suffix, expected in (("", False), ("[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=0\n", True),
-                                 ("[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=1\n", False)):
-            with self.subTest(matrix=suffix):
-                valid, _, matrix = self.report(case + suffix, matrix=True)
-                self.assertEqual(valid, expected)
-                self.assertIn("| Topology | KMS Backend | From Version | To Version | Result |", matrix)
-                self.assertIn("| single-single | local | v1 | v2 |" if suffix else "NOT RUN", matrix)
-
-    def test_s3_case_identifiers_include_digits(self):
-        valid, cases, _ = self.report("--- S3C-101 CreateBucket ---\n[PASS] S3C-101\n")
-        self.assertTrue(valid)
-        self.assertIn("| S3C-101 | CreateBucket (context not recorded) | PASS |", cases)
-
-
-class FunctionalEvidenceTests(WorkflowSteps, unittest.TestCase):
-    SUITES = (*FunctionalWorkflowTests.DIRECT_TESTS, "heal", "performance")
-
-    def prepare(self, suite: str) -> None:
-        self.temp = tempfile.TemporaryDirectory()
-        self.addCleanup(self.temp.cleanup)
-        self.directory = Path(self.temp.name)
-        self.source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text()
-        self.steps = named_steps(yaml_block(self.source.splitlines(), FunctionalWorkflowTests.JOBS[suite], 2))
-        self.context = {expression: "" for expression in re.findall(r"\$\{\{\s*(.*?)\s*\}\}", self.source)}
-        self.context.update({
-            "github.server_url": "https://github.com", "github.repository": "rustfs/rustfs",
-            "github.run_id": "314159", "github.run_attempt": "2", "github.sha": "0123456789abcdef0123456789abcdef01234567",
-            "github.event_name": "repository_dispatch", "steps.test.outcome": "success",
-            "secrets.PF_TESTING_GH_TOKEN": "local-fixture", "env.PF_TESTING_GH_TOKEN": "local-fixture",
-        })
-        self.artifacts = self.directory / f"rustfs-{suite}-314159-2"
-        self.env = {
-            **os.environ, "GITHUB_ENV": str(self.directory / "github-env"), "RUNNER_TEMP": self.temp.name,
-            "GITHUB_STEP_SUMMARY": str(self.directory / "summary.md"), "RUSTFS_NODES": "fixture-node",
-            "RUSTFS_NIGHTLY_PACKAGE_URL": "https://example.invalid/package.deb", "CAPTURE_BODY": str(self.directory / "issue.md"),
-        }
-        for key in ("server_url", "repository", "run_id", "run_attempt", "sha", "event_name"):
-            self.env[f"GITHUB_{key.upper()}"] = self.context[f"github.{key}"]
-        (self.directory / "scripts").mkdir()
-        (self.directory / "scripts/functional_case_report.py").symlink_to(ROOT / "scripts/functional_case_report.py")
-        fake_bin = self.directory / "bin"
-        fake_bin.mkdir()
-        (fake_bin / "python3").symlink_to(sys.executable)
-        for command, body in (
-            ("ssh", 'printf "fixture-version\\n"\n'),
-            ("gh", 'if [ "$1 $2" = "issue create" ]; then\n'
-                   '  while [ "$#" -gt 0 ]; do\n'
-                   '    if [ "$1" = "--body-file" ]; then cat "$2" > "$CAPTURE_BODY"; fi\n'
-                   '    shift\n'
-                   '  done\n'
-                   'elif [ "$1 $2" = "api --method" ]; then cat >/dev/null; fi\n'),
-        ):
-            script = fake_bin / command
-            script.write_text("#!/bin/sh\n" + body)
-            script.chmod(0o755)
-        self.env["PATH"] = f"{fake_bin}{os.pathsep}{os.environ['PATH']}"
-
-    def test_evidence_wiring_and_failed_initialization_cannot_publish_stale_files(self):
-        for suite in self.SUITES:
-            with self.subTest(suite=suite):
-                self.prepare(suite)
-                self.assertNotIn("/tmp/rustfs-", self.source)
-                names = list(self.steps)
-                self.assertLess(names.index("Initialize functional evidence"), names.index("Checkout auto-testing scripts (with retry)"))
-                if suite in FunctionalWorkflowTests.DIRECT_TESTS:
-                    self.assertLess(names.index("Checkout repository (for report parser)"), names.index("Checkout auto-testing scripts (with retry)"))
-                for name, lines in self.steps.items():
-                    if name in ("Generate report", "Upload functional report to dashboard") or any("uses: actions/upload-artifact@" in line for line in lines):
-                        self.assertIn("        if: ${{ always() && steps.evidence.outcome == 'success' }}", lines)
-                    if any("uses: actions/upload-artifact@" in line for line in lines):
-                        self.assertIn("          path: ${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/", lines)
-                        self.assertIn("          if-no-files-found: error", lines)
-                self.artifacts.mkdir()
-                for filename in ("report.md", "suite.log"):
-                    (self.artifacts / filename).write_text("OLD RUN EVIDENCE")
-                self.env.update(REPORT_FILE=str(self.artifacts / "report.md"), LOG_FILE=str(self.artifacts / "suite.log"))
-                initialized = self.run_step("Initialize functional evidence")
-                self.assertNotEqual(initialized.returncode, 0)
-                self.assertFalse(Path(self.env["GITHUB_ENV"]).exists())
-                issue = self.run_step("File failure issue in rustfs/backlog")
-                self.assertEqual(issue.returncode, 0, issue.stderr)
-                body = Path(self.env["CAPTURE_BODY"]).read_text()
-                self.assertNotIn("OLD RUN EVIDENCE", body)
-                self.assertIn("no report or log file was produced", body)
-                self.assertEqual((self.artifacts / "report.md").read_text(), "OLD RUN EVIDENCE")
-
-    def test_reports_use_only_current_complete_suite_evidence(self):
-        for suite in self.SUITES[:-1]:
-            good = "--- KMS-101 roundtrip ---\n[PASS] KMS-101\n"
-            partial = "--- KMS-101 roundtrip ---\n[PASS] KMS-101\n--- KMS-102 unfinished ---\n"
-            if suite == "s3-compat":
-                good, partial = good.replace("KMS-", "S3C-"), partial.replace("KMS-", "S3C-")
-            if suite == "upgrade":
-                good += "[UPG-TOPO] single-single local v1 v2 PASS=1 FAIL=0\n"
-            if suite == "heal":
-                good = "".join(f"[HEAL-STEP] {step} fixture PASS\n" for step in range(1, 8))
-                partial = "[HEAL-STEP] 1 fixture PASS\n"
-            for outcome, log in (("success", good), ("failure", good), ("success", partial), ("success", ""),
-                                 ("success", None), ("skipped", None), ("cancelled", good)):
-                with self.subTest(suite=suite, outcome=outcome, log=log):
-                    self.prepare(suite)
-                    stale = self.directory / "old-suite.log"
-                    stale.write_text("OLD RUN EVIDENCE\n" + good)
-                    self.env.update(LOG_FILE=str(stale), REPORT_FILE=str(stale))
-                    self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
-                    self.assertEqual(self.env["LOG_FILE"], str(self.artifacts / "suite.log"))
-                    self.assertEqual(self.env["TMPDIR"], str(self.artifacts))
-                    if log is not None:
-                        Path(self.env["LOG_FILE"]).write_text(log)
-                    self.context["steps.test.outcome"] = outcome
-                    report = self.run_step("Generate report")
-                    success = outcome == "success" and log == good
-                    self.assertEqual(report.returncode == 0, success, report.stderr)
-                    contents = Path(self.env["REPORT_FILE"]).read_text()
-                    self.assertNotIn("OLD RUN EVIDENCE", contents)
-                    self.assertEqual("| PASS |" in contents, success)
-                    for value in ("actions/runs/314159", "Attempt: 2", "Workflow Commit: " + self.context["github.sha"],
-                                  f"Test Step Outcome: {'success' if success else 'failure'}", f"Suite Step Outcome: {outcome}"):
-                        self.assertIn(value, contents)
-                    self.assertEqual(Path(self.env["GITHUB_STEP_SUMMARY"]).read_text(), contents)
-                    evidence = (self.artifacts / ("steps.md" if suite == "heal" else "cases.md")).read_text()
-                    if log in (good, partial):
-                        self.assertIn("| PASS |", evidence)
-                    self.assertNotIn("OLD RUN EVIDENCE", evidence)
-
-    def test_actual_suite_commands_pass_the_current_log_and_scratch_paths(self):
-        for suite in self.SUITES:
-            with self.subTest(suite=suite):
-                self.prepare(suite)
-                self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
-                scripts = self.directory / "auto-testing"
-                scripts.mkdir()
-                filename = f"rustfs_{suite}_test.sh" if suite in ("heal", "performance") else f"rustfs-{suite}-test.sh"
-                script = scripts / filename
-                script.write_text(
-                    '#!/bin/bash\nset -euo pipefail\nlog=""\n'
-                    'while [ "$#" -gt 0 ]; do\n'
-                    '  if [ "$1" = "--log-file" ]; then log="$2"; shift; fi\n'
-                    '  shift\n'
-                    'done\n'
-                    '[ "$log" = "$LOG_FILE" ] || exit 31\n'
-                    'printf "CURRENT SUITE LOG\\n" > "$log"\n'
-                    'scratch=$(mktemp -d "$TMPDIR/fixture.XXXXXX")\n'
-                    'printf "CURRENT SCRATCH\\n" > "$scratch/trace.log"\n'
-                    'if [ -n "${RUSTFS_RESULT_DIR:-}" ]; then\n'
-                    '  mkdir -p "$RUSTFS_RESULT_DIR"\n'
-                    '  printf "CURRENT RESULTS\\n" > "$RUSTFS_RESULT_DIR/summary.md"\n'
-                    'fi\n'
-                )
-                script.chmod(0o755)
-                name = FunctionalWorkflowTests.DIRECT_TESTS.get(suite) or (
-                    "Run benchmark (GET/PUT/MIXED)" if suite == "performance" else "Run heal test (write -> outage -> heal -> verify)"
-                )
-                result = self.run_step(name)
-                self.assertEqual(result.returncode, 0, result.stderr)
-                self.assertEqual((self.artifacts / "suite.log").read_text(), "CURRENT SUITE LOG\n")
-                self.assertEqual(len(list(self.artifacts.glob("fixture.*/trace.log"))), 1)
-                if suite == "performance":
-                    self.assertEqual((self.artifacts / "results/summary.md").read_text(), "CURRENT RESULTS\n")
-
-    def test_heal_accumulates_actual_staged_steps_without_overwriting_failures(self):
-        self.prepare("heal")
-        self.assertEqual(self.run_step("Initialize functional evidence").returncode, 0)
-        script = self.directory / "auto-testing/rustfs_heal_test.sh"
-        script.parent.mkdir()
-        # Result printf and full-run condition from auto-testing@6120aa0a76de:143,1163-1168.
-        script.write_text(r'''#!/bin/bash
-set -euo pipefail
-SELECTED_STEPS=()
-PREFLIGHT=0
-while [ "$#" -gt 0 ]; do
-  case "$1" in
-    --steps) IFS=',' read -ra SELECTED_STEPS <<< "$2"; shift ;;
-    --log-file) LOG_FILE="$2"; shift ;;
-    --preflight) PREFLIGHT=1 ;;
-  esac
-  shift
-done
-if [ "$PREFLIGHT" -eq 1 ]; then
-  printf '\n' >> "$INVOKED_STEPS"
-  exit 0
-fi
-printf '%s\n' "${SELECTED_STEPS[*]}" >> "$INVOKED_STEPS"
-emit_step_result() {
-  local n="$1" desc="$2" status="$3"
-  printf '[HEAL-STEP] %s %s %s\n' "${n}" "${desc}" "${status}"
-}
-{
-  for step in "${SELECTED_STEPS[@]}"; do
-    emit_step_result "$step" "fixture step $step" PASS
-  done
-  want_all=1
-  for s in 1 2 3 4 5 6 7; do
-    [[ " ${SELECTED_STEPS[*]} " == *" ${s} "* ]] || want_all=0
-  done
-  if [ "${want_all}" -eq 1 ]; then
-    printf '[HEAL-RESULT] PASS all steps passed\n'
-  fi
-} >> "$LOG_FILE"
-''')
-        script.chmod(0o755)
-        self.env["INVOKED_STEPS"] = str(self.directory / "invoked-steps")
-        for name in ("Install RustFS package & start cluster", "Preflight checks", "Run heal test (write -> outage -> heal -> verify)"):
-            result = self.run_step(name)
-            self.assertEqual(result.returncode, 0, result.stderr)
-        self.assertEqual(Path(self.env["INVOKED_STEPS"]).read_text().splitlines(), ["1 2", "", "3 4 5 6 7"])
-        log = Path(self.env["LOG_FILE"]).read_text()
-        self.assertNotIn("[HEAL-RESULT]", log)
-        self.assertEqual(log.count("[HEAL-STEP]"), 7)
-        report = self.run_step("Generate report")
-        self.assertEqual(report.returncode, 0, report.stderr)
-        failed_logs = ["\n".join(line for line in log.splitlines() if not line.startswith(f"[HEAL-STEP] {step} ")) + "\n"
-                       for step in range(1, 8)]
-        failed_logs += [
-            log.replace("[HEAL-STEP] 3", "[HEAL-STEP] 3 original failure FAIL\n[HEAL-STEP] 3"),
-            log + "[HEAL-STEP] 3 later step failure FAIL\n",
-            log + "[HEAL-RESULT] FAIL earlier failure\n[HEAL-RESULT] PASS later success\n",
-            log.replace("[HEAL-STEP] 4 fixture step 4 PASS", "[HEAL-STEP] 4 fixture step 4 SKIP"),
-        ]
-        for failed_log in failed_logs:
-            with self.subTest(log=failed_log):
-                Path(self.env["LOG_FILE"]).write_text(failed_log)
-                report = self.run_step("Generate report")
-                self.assertNotEqual(report.returncode, 0, report.stderr)
-                contents = Path(self.env["REPORT_FILE"]).read_text()
-                self.assertIn("Test Step Outcome: failure", contents)
-                self.assertNotIn("| PASS |", contents)
-                if "original failure" in failed_log:
-                    self.assertIn("| 3 | original failure | FAIL |", (self.artifacts / "steps.md").read_text())
-                if "later step failure" in failed_log:
-                    self.assertIn("| 3 | later step failure | FAIL |", (self.artifacts / "steps.md").read_text())
-
-    def test_performance_results_version_and_report_are_bound_to_the_run(self):
-        self.prepare("performance")
-        initialized = self.run_step("Initialize functional evidence")
-        self.assertEqual(initialized.returncode, 0, initialized.stderr)
-        self.assertEqual(self.env["RUSTFS_RESULT_DIR"], str(self.artifacts / "results"))
-        self.assertEqual(self.env["VERSION_FILE"], str(self.artifacts / "version.txt"))
-        version = self.run_step("Collect RustFS version info")
-        self.assertEqual(version.returncode, 0, version.stderr)
-        self.assertIn("fixture-version", Path(self.env["VERSION_FILE"]).read_text())
-        old_summary = self.directory / "old-results/summary.md"
-        old_summary.parent.mkdir()
-        old_summary.write_text("OLD RUN EVIDENCE")
-        upload = "Upload report to dashboard (reports/YYYY-MM-DD.md)"
-        self.assertNotEqual(self.run_step(upload).returncode, 0)
-        self.assertFalse(Path(self.env["REPORT_FILE"]).exists())
-        results = Path(self.env["RUSTFS_RESULT_DIR"])
-        results.mkdir()
-        (results / "summary.md").write_text("CURRENT PERFORMANCE RESULTS\n")
-        report = self.run_step(upload)
-        self.assertEqual(report.returncode, 0, report.stderr)
-        contents = Path(self.env["REPORT_FILE"]).read_text()
-        for value in ("actions/runs/314159", "**Attempt**: 2", "**Workflow Commit**: " + self.context["github.sha"],
-                      "CURRENT PERFORMANCE RESULTS", "fixture-version"):
-            self.assertIn(value, contents)
-        self.assertNotIn("OLD RUN EVIDENCE", contents)
-
-
 if __name__ == "__main__":
     unittest.main()