mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-12 05:49:01 +00:00
Compare commits
10 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 houseme
|
824231ff1a | ||
|
cxymds
|
66026d7d7b | ||
|
houseme
|
82a4083981 | ||
|
hector
|
9f5ff23fd8 | ||
|
hector
|
5fc92cdd27 | ||
|
cxymds
|
7bd09a00b0 | ||
|
hector
|
729cff5f32 | ||
|
cxymds
|
c478a392e7 | ||
|
cxymds
|
c8ccc1e198 | ||
|
cxymds
|
74ba5c205c |
@@ -0,0 +1,267 @@
|
||||
# RustFS Fault-Tolerance (degradation) Test
|
||||
#
|
||||
# Scenario suite for the 2026-09 degradation report: verifies read/write
|
||||
# behavior under drive and node loss against the erasure-coding contract and
|
||||
# snapshots health-endpoint responses at every tier.
|
||||
#
|
||||
# A single-node 4 drives (SNMD): hide 1/2/3 drives, restore
|
||||
# B multi-node 4x1 (one drive per node): stop 1/2/3 nodes, restore
|
||||
# C multi-node 4x4 (16 drives, EC:4): stop 1 node (read-quorum boundary),
|
||||
# stop 2 nodes, restore
|
||||
# C2 multi-node 4x4 with EC:8: 2 nodes down puts 8 drives online -- reads
|
||||
# satisfy the EC read quorum while the lock majority is broken (the
|
||||
# reported divergence window: reads 503 with lock_quorum_unavailable)
|
||||
#
|
||||
# Expectations come from product source (default_parity_count, erasure set
|
||||
# sizing). By default a "reads refused although the read quorum is met"
|
||||
# observation is reported as known-divergence without failing the suite; the
|
||||
# strict input turns those into failures once the product behavior changes.
|
||||
|
||||
name: RustFS Fault-Tolerance Test
|
||||
|
||||
on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
package_url:
|
||||
description: 'Direct .deb URL. Required unless the nightly default is wanted.'
|
||||
required: false
|
||||
type: string
|
||||
strict:
|
||||
description: 'Fail the suite when reads are refused despite a met read quorum'
|
||||
type: boolean
|
||||
default: false
|
||||
cleanup_before:
|
||||
description: 'Reset the nodes before the test (DESTROYS existing data/config)'
|
||||
type: boolean
|
||||
default: true
|
||||
cleanup_after:
|
||||
description: 'Reset the nodes after the test (DESTROYS test data/config)'
|
||||
type: boolean
|
||||
default: true
|
||||
repository_dispatch:
|
||||
# Chain handoff: dispatched when the replication suite finishes, ahead of
|
||||
# the performance suite.
|
||||
types: [rustfs-chain-fault-tolerance]
|
||||
|
||||
permissions:
|
||||
contents: read
|
||||
|
||||
# The suite stops services and hides drive dirs on the shared fleet; only one
|
||||
# functional suite may touch the environment at a time.
|
||||
concurrency:
|
||||
group: rustfs-shared-functional-tests
|
||||
cancel-in-progress: false
|
||||
|
||||
defaults:
|
||||
run:
|
||||
shell: bash
|
||||
|
||||
env:
|
||||
RUSTFS_ACCESS_KEY: ${{ secrets.RUSTFS_ACCESS_KEY }}
|
||||
RUSTFS_SECRET_KEY: ${{ secrets.RUSTFS_SECRET_KEY }}
|
||||
RUSTFS_NODES: ${{ secrets.RUSTFS_NODES || vars.RUSTFS_NODES }}
|
||||
RUSTFS_SSH_USER: ${{ secrets.RUSTFS_SSH_USER || vars.RUSTFS_SSH_USER }}
|
||||
PF_TESTING_GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
RUSTFS_NIGHTLY_PACKAGE_URL: ${{ vars.RUSTFS_NIGHTLY_PACKAGE_URL || 'https://dl.rustfs.com/artifacts/rustfs/packages/nightly/rustfs-nightly-latest.deb' }}
|
||||
|
||||
jobs:
|
||||
fault-tolerance-test:
|
||||
runs-on: smoke-testing
|
||||
timeout-minutes: 480
|
||||
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-ft-${GITHUB_RUN_ID}-${GITHUB_RUN_ATTEMPT}"
|
||||
mkdir -- "${FUNCTIONAL_ARTIFACTS_DIR}" "${FUNCTIONAL_ARTIFACTS_DIR}/evidence"
|
||||
{
|
||||
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 'EVIDENCE_DIR=%s/evidence\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)
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
rm -rf auto-testing
|
||||
for attempt in 1 2 3 4 5; do
|
||||
if gh repo clone rustfs/auto-testing auto-testing -- --depth 1 --quiet; then
|
||||
echo "auto-testing cloned (attempt ${attempt})"
|
||||
exit 0
|
||||
fi
|
||||
rm -rf auto-testing
|
||||
echo "clone attempt ${attempt} failed; retrying in $((attempt * 15))s" >&2
|
||||
sleep $((attempt * 15))
|
||||
done
|
||||
echo "ERROR: unable to clone rustfs/auto-testing after 5 attempts" >&2
|
||||
exit 1
|
||||
|
||||
- name: Show environment
|
||||
run: |
|
||||
uname -a
|
||||
jq --version
|
||||
aws --version
|
||||
df -h /data | tail -1
|
||||
|
||||
- name: Cleanup environment (before)
|
||||
if: ${{ inputs.cleanup_before != 'false' }}
|
||||
run: |
|
||||
./auto-testing/rustfs-fault-tolerance-test.sh --cleanup -y --log-file "${LOG_FILE}"
|
||||
|
||||
- name: Run fault-tolerance scenarios (A, B, C, C2)
|
||||
id: test
|
||||
run: |
|
||||
ARGS=(--all -y --package-url "${{ inputs.package_url || env.RUSTFS_NIGHTLY_PACKAGE_URL }}" --log-file "${LOG_FILE}")
|
||||
if [ "${{ inputs.strict }}" = "true" ]; then
|
||||
ARGS+=(--strict)
|
||||
fi
|
||||
./auto-testing/rustfs-fault-tolerance-test.sh "${ARGS[@]}"
|
||||
|
||||
- name: Generate report
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
{
|
||||
echo "# RustFS fault-tolerance test report"
|
||||
echo ""
|
||||
echo "- Run: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
|
||||
echo "- Package: ${{ inputs.package_url || 'nightly (R2 latest)' }}"
|
||||
echo "- Strict mode: ${{ inputs.strict || 'false' }}"
|
||||
echo ""
|
||||
echo "## Per-probe results"
|
||||
echo ""
|
||||
echo '```'
|
||||
grep -E '^FT-(CASE|SUMMARY|REPORT)' "${LOG_FILE}" || echo "(no FT-CASE lines found)"
|
||||
echo '```'
|
||||
echo ""
|
||||
echo "## Health snapshots"
|
||||
echo ""
|
||||
for f in "${FUNCTIONAL_ARTIFACTS_DIR}"/evidence/*.code; do
|
||||
[ -e "${f}" ] || continue
|
||||
printf '%s -> %s\n' "$(basename "${f}" .code)" "$(cat "${f}")"
|
||||
done
|
||||
} > "${REPORT_FILE}"
|
||||
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Signal-based lifecycle: dedups against open issues by label
|
||||
# (fault-tolerance + FT case), labels new issues (functional-test,
|
||||
# category, case IDs, env), and closes fixed issues after a fully
|
||||
# green run. Never acts on cancelled runs. Logic lives in
|
||||
# auto-testing/scripts/issue_manager.py, which parses the
|
||||
# FT-CASE verdict lines from the suite log.
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
SUITE: 'fault-tolerance'
|
||||
SUITE_LABEL: 'Fault-Tolerance'
|
||||
RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload test logs & evidence
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6
|
||||
with:
|
||||
name: rustfs-fault-tolerance-${{ github.run_id }}-${{ github.run_attempt }}
|
||||
path: |
|
||||
${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/report.md
|
||||
${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/suite.log
|
||||
${{ env.FUNCTIONAL_ARTIFACTS_DIR }}/evidence/
|
||||
if-no-files-found: warn
|
||||
|
||||
- name: Cleanup environment (after)
|
||||
if: ${{ always() && inputs.cleanup_after != 'false' }}
|
||||
run: |
|
||||
./auto-testing/rustfs-fault-tolerance-test.sh --cleanup -y --log-file "${LOG_FILE}" || true
|
||||
|
||||
- name: "Continue functional chain (next: Performance)"
|
||||
# Only chain-triggered runs forward to the next suite; standalone
|
||||
# workflow_dispatch runs stop after their own cleanup. A failed
|
||||
# handoff retries, then files an alert issue in rustfs/backlog.
|
||||
if: ${{ always() && github.event_name == 'repository_dispatch' }}
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
run: |
|
||||
set -uo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; cannot dispatch the next suite" >&2
|
||||
exit 1
|
||||
fi
|
||||
DISPATCHED=0
|
||||
for attempt in 1 2 3; do
|
||||
if gh api --method POST repos/rustfs/rustfs/dispatches \
|
||||
-f event_type='rustfs-chain-performance' \
|
||||
-F 'client_payload[from_suite]=fault-tolerance'; then
|
||||
echo "dispatched next suite Performance (attempt ${attempt})"
|
||||
DISPATCHED=1
|
||||
break
|
||||
fi
|
||||
echo "dispatch attempt ${attempt} failed; retrying in ${attempt}0s" >&2
|
||||
sleep "${attempt}0"
|
||||
done
|
||||
if [ "${DISPATCHED:-0}" -ne 1 ]; then
|
||||
echo "ERROR: functional chain stalled: could not dispatch Performance after 3 attempts" >&2
|
||||
TITLE="[functional][chain] stalled after fault-tolerance (run ${GITHUB_RUN_ID})"
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The functional chain could not hand off from **fault-tolerance** to **Performance** after 3 attempts."
|
||||
echo ""
|
||||
echo "- Failed suite job: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
|
||||
echo "- Expected next event: 'rustfs-chain-performance'"
|
||||
echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
|
||||
echo "- Recovery: re-dispatch manually with"
|
||||
FENCE="$(printf "\x60\x60\x60")"; echo " ${FENCE}"
|
||||
echo " gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-performance'"
|
||||
FENCE="$(printf "\x60\x60\x60")"; echo " ${FENCE}"
|
||||
} > "${BODY_FILE}"
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test \
|
||||
|| gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}" \
|
||||
|| echo "could not file the stall alert issue either; check the token" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
- name: Notify on failure
|
||||
if: failure()
|
||||
run: |
|
||||
echo "RustFS fault-tolerance test failed"
|
||||
echo "Package source: ${{ inputs.package_url || 'nightly (R2 latest)' }}"
|
||||
echo "See the uploaded log artifact and FT-CASE lines for details."
|
||||
@@ -14,7 +14,7 @@
|
||||
|
||||
# Functional chain driver: runs the ten functional suites in a fixed order
|
||||
# (upgrade -> s3 -> kms -> tier -> storage -> heal -> pool -> security ->
|
||||
# replication -> performance). Each suite attempts the next handoff even
|
||||
# replication -> fault-tolerance -> performance). Each suite attempts the next handoff even
|
||||
# when its tests fail.
|
||||
#
|
||||
# Each suite workflow can still be dispatched standalone (workflow_dispatch);
|
||||
|
||||
@@ -280,65 +280,51 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload test logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -4,9 +4,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -231,65 +230,55 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -237,65 +237,48 @@ jobs:
|
||||
echo "created ${REPORT_PATH} in rustfs/dashboard"
|
||||
fi
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.benchmark.outcome == 'failure' || steps.benchmark.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Signal-based lifecycle: dedups against open issues by label,
|
||||
# labels new issues (functional-test, category, case IDs, env),
|
||||
# closes fixed issues after a fully green run. Never acts on
|
||||
# cancelled runs. Logic lives in auto-testing/scripts/.
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.benchmark.outcome }}" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload test logs & results
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -556,64 +556,54 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.pool_test.outcome == 'failure' || steps.pool_test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() }}
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
SUITE: 'pool'
|
||||
SUITE_LABEL: 'Pool expansion'
|
||||
RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
|
||||
REPORT_FILE: '${{ env.POOL_ARTIFACT_DIR }}/pool-report.md'
|
||||
LOG_FILE: '${{ env.POOL_ARTIFACT_DIR }}/pool-test.log'
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
echo ""
|
||||
echo "- Suite: \`${SUITE}\`"
|
||||
echo "- Run: ${RUN_URL}"
|
||||
echo "- Trigger: ${GITHUB_EVENT_NAME}"
|
||||
echo "- Date: $(date -u +%Y-%m-%d)"
|
||||
echo ""
|
||||
echo "## Report (errors and symptoms)"
|
||||
echo ""
|
||||
if [ -s "${REPORT_FILE}" ]; then
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.pool_test.outcome }}" \
|
||||
--report-file "${POOL_ARTIFACT_DIR}/pool-report.md" \
|
||||
--log "${POOL_ARTIFACT_DIR}/pool-test.log" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload test logs
|
||||
if: always()
|
||||
|
||||
@@ -18,9 +18,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -239,65 +238,55 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
@@ -329,7 +318,7 @@ jobs:
|
||||
'
|
||||
done
|
||||
|
||||
- name: "Continue functional chain (next: Performance)"
|
||||
- name: "Continue functional chain (next: Fault tolerance)"
|
||||
if: ${{ always() && github.event_name == 'repository_dispatch' }}
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
@@ -342,9 +331,9 @@ jobs:
|
||||
DISPATCHED=0
|
||||
for attempt in 1 2 3; do
|
||||
if gh api --method POST repos/rustfs/rustfs/dispatches \
|
||||
-f event_type='rustfs-chain-performance' \
|
||||
-f event_type='rustfs-chain-fault-tolerance' \
|
||||
-F 'client_payload[from_suite]=replication'; then
|
||||
echo "dispatched next suite Performance (attempt ${attempt})"
|
||||
echo "dispatched next suite Fault tolerance (attempt ${attempt})"
|
||||
DISPATCHED=1
|
||||
break
|
||||
fi
|
||||
@@ -352,19 +341,19 @@ jobs:
|
||||
sleep "${attempt}0"
|
||||
done
|
||||
if [ "${DISPATCHED:-0}" -ne 1 ]; then
|
||||
echo "ERROR: functional chain stalled: could not dispatch Performance after 3 attempts" >&2
|
||||
echo "ERROR: functional chain stalled: could not dispatch Fault tolerance after 3 attempts" >&2
|
||||
TITLE="[functional][chain] stalled after replication (run ${GITHUB_RUN_ID})"
|
||||
BODY_FILE="$(mktemp)"
|
||||
trap 'rm -f "${BODY_FILE}"' EXIT
|
||||
{
|
||||
echo "The functional chain could not hand off from **replication** to **Performance** after 3 attempts."
|
||||
echo "The functional chain could not hand off from **replication** to **Fault tolerance** after 3 attempts."
|
||||
echo ""
|
||||
echo "- Failed suite job: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
|
||||
echo "- Expected next event: 'rustfs-chain-performance'"
|
||||
echo "- Expected next event: 'rustfs-chain-fault-tolerance'"
|
||||
echo "- Likely cause: PF_TESTING_GH_TOKEN lacks contents:write on rustfs/rustfs, or the GitHub API was unavailable."
|
||||
echo "- Recovery: re-dispatch manually with"
|
||||
FENCE="$(printf "\x60\x60\x60")"; echo " ${FENCE}"
|
||||
echo " gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-performance'"
|
||||
echo " gh api --method POST repos/rustfs/rustfs/dispatches -f event_type='rustfs-chain-fault-tolerance'"
|
||||
FENCE="$(printf "\x60\x60\x60")"; echo " ${FENCE}"
|
||||
} > "${BODY_FILE}"
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
|
||||
@@ -4,9 +4,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -208,65 +207,55 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -18,9 +18,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -240,65 +239,54 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
SUITE: 'security'
|
||||
SUITE_LABEL: 'Security'
|
||||
EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
|
||||
RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
|
||||
REPORT_FILE: ${{ env.SECURITY_ARTIFACTS_DIR }}/report.md
|
||||
LOG_FILE: ''
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
echo ""
|
||||
echo "- Suite: \`${SUITE}\`"
|
||||
echo "- Run: ${RUN_URL}"
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report-file "${SECURITY_ARTIFACTS_DIR}/report.md" \
|
||||
--log "${SECURITY_ARTIFACTS_DIR}/suite.log" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -4,9 +4,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -223,65 +222,55 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -4,9 +4,8 @@ on:
|
||||
workflow_dispatch:
|
||||
inputs:
|
||||
rustfs_version:
|
||||
description: 'RustFS release tag to test (e.g. 1.0.0-rc.4-preview.1)'
|
||||
description: 'RustFS release tag to test (leave empty to use the latest nightly deb)'
|
||||
required: false
|
||||
default: '1.0.0-rc.4-preview.1'
|
||||
package_url:
|
||||
description: 'Direct .deb URL (nightly/R2/dev). Overrides rustfs_version.'
|
||||
required: false
|
||||
@@ -495,75 +494,55 @@ jobs:
|
||||
fi
|
||||
[ "${failed}" -eq 0 ]
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled' || steps.evidence_verify.outcome == 'failure' || steps.evidence_verify.outcome == 'cancelled' || steps.gate.outcome == 'failure' || steps.gate.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
continue-on-error: true
|
||||
env:
|
||||
GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
|
||||
SUITE: 'tier'
|
||||
SUITE_LABEL: 'Tier'
|
||||
RUN_URL: ${{ github.server_url }}/${{ github.repository }}/actions/runs/${{ github.run_id }}
|
||||
EVIDENCE_DIR: ${{ env.TIER_ARTIFACTS_DIR }}
|
||||
EVIDENCE_OUTCOME: ${{ steps.evidence.outcome }}
|
||||
VERIFY_OUTCOME: ${{ steps.evidence_verify.outcome }}
|
||||
GATE_OUTCOME: ${{ steps.gate.outcome }}
|
||||
REPORT_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier-report.md
|
||||
LOG_FILE: ${{ env.TIER_ARTIFACTS_DIR }}/rustfs-tier.log
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
echo ""
|
||||
echo "- Suite: \`${SUITE}\`"
|
||||
echo "- Run: ${RUN_URL}"
|
||||
echo "- Trigger: ${GITHUB_EVENT_NAME}"
|
||||
echo "- Date: $(date -u +%Y-%m-%d)"
|
||||
echo "- Evidence initialization: ${EVIDENCE_OUTCOME}"
|
||||
echo "- Evidence verification: ${VERIFY_OUTCOME}"
|
||||
echo "- Final gate: ${GATE_OUTCOME}"
|
||||
echo ""
|
||||
echo "## Report (errors and symptoms)"
|
||||
echo ""
|
||||
if [ "${EVIDENCE_OUTCOME}" != "success" ]; then
|
||||
echo "(the run evidence directory was rejected; its contents were not read)"
|
||||
elif [ ! -d "${EVIDENCE_DIR}" ] || [ -L "${EVIDENCE_DIR}" ]; then
|
||||
echo "(the run evidence directory is missing or unsafe; its contents were not read)"
|
||||
elif [ -s "${REPORT_FILE}" ]; then
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
PACKAGE_URL='${{ inputs.package_url }}'
|
||||
RUSTFS_VERSION='${{ inputs.rustfs_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${PACKAGE_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_URL}"
|
||||
elif [ -n "${RUSTFS_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="version ${RUSTFS_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${TIER_ARTIFACTS_DIR}/rustfs-tier-cases.md" \
|
||||
--report-file "${TIER_ARTIFACTS_DIR}/rustfs-tier-report.md" \
|
||||
--log "${TIER_ARTIFACTS_DIR}/rustfs-tier.log" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: "Continue functional chain (next: Storage engine)"
|
||||
# Only chain-triggered runs forward to the next suite; standalone
|
||||
|
||||
@@ -306,65 +306,62 @@ jobs:
|
||||
fi
|
||||
rm -f "${B64_FILE}"
|
||||
|
||||
- name: File failure issue in rustfs/backlog
|
||||
if: ${{ always() && (failure() || steps.test.outcome == 'failure' || steps.test.outcome == 'cancelled') }}
|
||||
- name: Manage backlog issues (dedup / label / auto-close)
|
||||
# Replaces the old per-run failure filing. One entry point that:
|
||||
# - dedups by signal: failing cases are matched against open backlog
|
||||
# issues by label (category + case ID); covered cases become a
|
||||
# comment on the existing issue, only uncovered cases file a new one
|
||||
# - labels new issues (functional-test, category, case IDs, env)
|
||||
# - closes fixed issues after a fully green run
|
||||
# - never files or closes on cancelled runs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
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 }}
|
||||
run: |
|
||||
set -euo pipefail
|
||||
if [ -z "${GH_TOKEN:-}" ]; then
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue"
|
||||
echo "PF_TESTING_GH_TOKEN is not configured; skipping backlog issue management"
|
||||
exit 0
|
||||
fi
|
||||
TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
|
||||
EXISTING="$(gh issue list -R rustfs/backlog --state all \
|
||||
--search "in:title \"run ${GITHUB_RUN_ID}\"" \
|
||||
--json number --jq '.[].number' || true)"
|
||||
if [ -n "${EXISTING}" ]; then
|
||||
echo "backlog issue already exists for run ${GITHUB_RUN_ID}; skipping"
|
||||
if [ ! -f auto-testing/scripts/issue_manager.py ]; then
|
||||
echo "issue_manager.py not found in auto-testing checkout; skipping"
|
||||
exit 0
|
||||
fi
|
||||
redact() {
|
||||
sed -E \
|
||||
-e 's/(RUSTFS_(ACCESS_KEY|SECRET_KEY)[=: ]+)[^[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/(Authorization:).*/\1 [REDACTED]/Ig' \
|
||||
-e 's/(X-Amz-Signature=)[^&[:space:]]+/\1[REDACTED]/Ig' \
|
||||
-e 's/^.*(password|secret|token)[=: ].*/[REDACTED SENSITIVE LINE]/Ig'
|
||||
}
|
||||
BODY_FILE="$(mktemp)"
|
||||
{
|
||||
echo "The **${SUITE_LABEL}** functional suite failed."
|
||||
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
|
||||
redact < "${REPORT_FILE}"
|
||||
elif [ "${EVIDENCE_OUTCOME}" = "success" ] && [ -s "${LOG_FILE:-}" ]; then
|
||||
echo "(report file missing; log tail below)"
|
||||
echo ""
|
||||
tail -n 200 "${LOG_FILE}" | redact
|
||||
else
|
||||
echo "(no report or log file was produced)"
|
||||
fi
|
||||
} | head -c 55000 > "${BODY_FILE}"
|
||||
gh label create functional-test -R rustfs/backlog --color d73a4a 2>/dev/null || true
|
||||
if ! gh issue create -R rustfs/backlog --title "${TITLE}" \
|
||||
--body-file "${BODY_FILE}" --label functional-test; then
|
||||
gh issue create -R rustfs/backlog --title "${TITLE}" --body-file "${BODY_FILE}"
|
||||
FROM_URL='${{ inputs.from_url }}'
|
||||
FROM_VERSION='${{ inputs.from_version }}'
|
||||
TO_URL='${{ inputs.to_url }}'
|
||||
TO_VERSION='${{ inputs.to_version }}'
|
||||
PACKAGE_SOURCE=""
|
||||
if [ -n "${TO_URL}" ]; then
|
||||
PACKAGE_SOURCE="to ${TO_URL}"
|
||||
elif [ -n "${TO_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="to version ${TO_VERSION}"
|
||||
else
|
||||
PACKAGE_SOURCE="to ${RUSTFS_NIGHTLY_PACKAGE_URL}"
|
||||
fi
|
||||
echo "filed backlog issue for suite ${SUITE}"
|
||||
if [ -n "${FROM_URL}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_SOURCE}; from ${FROM_URL}"
|
||||
elif [ -n "${FROM_VERSION}" ]; then
|
||||
PACKAGE_SOURCE="${PACKAGE_SOURCE}; from version ${FROM_VERSION}"
|
||||
fi
|
||||
python3 auto-testing/scripts/issue_manager.py handle \
|
||||
--repo rustfs/backlog \
|
||||
--suite "${SUITE}" --category "${SUITE}" --suite-label "${SUITE_LABEL}" \
|
||||
--outcome "${{ steps.test.outcome }}" \
|
||||
--report "${FUNCTIONAL_ARTIFACTS_DIR}/cases.md" \
|
||||
--report-file "${REPORT_FILE}" \
|
||||
--log "${LOG_FILE}" \
|
||||
--run-url "${RUN_URL}" \
|
||||
--run-id "${GITHUB_RUN_ID}" \
|
||||
--attempt "${GITHUB_RUN_ATTEMPT}" \
|
||||
--commit "${GITHUB_SHA}" \
|
||||
--trigger "${{ github.event_name }}" \
|
||||
--package-source "${PACKAGE_SOURCE}" \
|
||||
--date "$(date -u +%Y-%m-%d)"
|
||||
|
||||
- name: Upload report and logs
|
||||
if: ${{ always() && steps.evidence.outcome == 'success' }}
|
||||
|
||||
@@ -779,7 +779,7 @@ fn free_version_physical_topology_generation(api: &ECStore) -> String {
|
||||
rustfs_utils::crypto::hex(hasher.finalize().as_slice())
|
||||
}
|
||||
|
||||
fn free_version_remote_tuple_matches(candidate: &ObjectInfo, expected: &ObjectInfo) -> std::io::Result<bool> {
|
||||
pub(crate) fn free_version_remote_tuple_matches(candidate: &ObjectInfo, expected: &ObjectInfo) -> std::io::Result<bool> {
|
||||
if candidate.transitioned_object.tier != expected.transitioned_object.tier
|
||||
|| candidate.transitioned_object.name != expected.transitioned_object.name
|
||||
{
|
||||
|
||||
@@ -577,6 +577,26 @@ fn heal_control_auth_may_need_replay_scope_refresh(err: &Error) -> bool {
|
||||
)
|
||||
}
|
||||
|
||||
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
|
||||
enum HealControlRetryAction {
|
||||
Reconnect,
|
||||
RefreshReplayScope,
|
||||
}
|
||||
|
||||
fn heal_control_retry_action(
|
||||
err: &Error,
|
||||
reconnect_attempted: bool,
|
||||
replay_scope_refresh_attempted: bool,
|
||||
) -> Option<HealControlRetryAction> {
|
||||
if !replay_scope_refresh_attempted && heal_control_auth_may_need_replay_scope_refresh(err) {
|
||||
return Some(HealControlRetryAction::RefreshReplayScope);
|
||||
}
|
||||
if !reconnect_attempted && PeerRestClient::is_network_like_error(err) {
|
||||
return Some(HealControlRetryAction::Reconnect);
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
fn decode_remote_version_state_capability(expected_member: &str, result: &[u8]) -> Result<Uuid> {
|
||||
let (topology_member, process_epoch) = rustfs_protos::decode_remote_version_state_capability(result).map_err(Error::other)?;
|
||||
if topology_member != expected_member {
|
||||
@@ -1753,34 +1773,41 @@ impl PeerRestClient {
|
||||
return Err(Error::other("heal control command exceeds size limit"));
|
||||
}
|
||||
let capability_probe = rustfs_protos::is_heal_control_capability_probe(&command);
|
||||
let result = self
|
||||
.heal_control_once(version, &topology_fingerprint, &command, capability_probe)
|
||||
.await;
|
||||
if result
|
||||
.as_ref()
|
||||
.err()
|
||||
.is_some_and(heal_control_auth_may_need_replay_scope_refresh)
|
||||
{
|
||||
self.prepare_heal_control_auth_retry().await;
|
||||
return self
|
||||
.finalize_result(
|
||||
self.heal_control_once(version, &topology_fingerprint, &command, capability_probe)
|
||||
.await,
|
||||
)
|
||||
let mut reconnect_attempted = false;
|
||||
let mut replay_scope_refresh_attempted = false;
|
||||
loop {
|
||||
let result = self
|
||||
.heal_control_once(version, &topology_fingerprint, &command, capability_probe)
|
||||
.await;
|
||||
let Some(action) = result
|
||||
.as_ref()
|
||||
.err()
|
||||
.and_then(|err| heal_control_retry_action(err, reconnect_attempted, replay_scope_refresh_attempted))
|
||||
else {
|
||||
return self.finalize_result(result).await;
|
||||
};
|
||||
match action {
|
||||
HealControlRetryAction::Reconnect => reconnect_attempted = true,
|
||||
HealControlRetryAction::RefreshReplayScope => replay_scope_refresh_attempted = true,
|
||||
}
|
||||
self.prepare_heal_control_retry(action).await;
|
||||
}
|
||||
self.finalize_result(result).await
|
||||
}
|
||||
|
||||
async fn prepare_heal_control_auth_retry(&self) {
|
||||
if let Err(err) = clear_peer_replay_state_for_addr(&self.grid_host) {
|
||||
async fn prepare_heal_control_retry(&self, action: HealControlRetryAction) {
|
||||
if action == HealControlRetryAction::RefreshReplayScope
|
||||
&& let Err(err) = clear_peer_replay_state_for_addr(&self.grid_host)
|
||||
{
|
||||
debug!(
|
||||
peer = %self.grid_host,
|
||||
error = %err,
|
||||
"could not clear heal control replay state before retry"
|
||||
);
|
||||
}
|
||||
self.evict_connection().await;
|
||||
// A restart can leave both the local offline gate and the peer replay
|
||||
// epoch stale. Clear the gate on either recovery step so the next
|
||||
// bounded attempt reaches a fresh channel instead of fast-failing.
|
||||
self.prepare_retry().await;
|
||||
}
|
||||
|
||||
async fn heal_control_once(
|
||||
@@ -3867,6 +3894,51 @@ mod tests {
|
||||
)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn heal_control_retry_plan_allows_one_reconnect_and_one_epoch_refresh() {
|
||||
let offline = Error::RemoteClientUnavailable("peer http://127.0.0.1:9000 is temporarily offline".to_string());
|
||||
let stale_epoch = Error::from(tonic::Status::unauthenticated("No valid auth token"));
|
||||
|
||||
assert_eq!(heal_control_retry_action(&offline, false, false), Some(HealControlRetryAction::Reconnect));
|
||||
assert_eq!(heal_control_retry_action(&offline, true, false), None);
|
||||
assert_eq!(
|
||||
heal_control_retry_action(&stale_epoch, true, false),
|
||||
Some(HealControlRetryAction::RefreshReplayScope),
|
||||
"a reconnect may expose the restarted peer's stale replay epoch"
|
||||
);
|
||||
assert_eq!(heal_control_retry_action(&stale_epoch, true, true), None);
|
||||
assert_eq!(
|
||||
heal_control_retry_action(&stale_epoch, false, false),
|
||||
Some(HealControlRetryAction::RefreshReplayScope)
|
||||
);
|
||||
assert_eq!(
|
||||
heal_control_retry_action(&offline, false, true),
|
||||
Some(HealControlRetryAction::Reconnect),
|
||||
"an epoch refresh may be followed by one bounded reconnect"
|
||||
);
|
||||
assert_eq!(heal_control_retry_action(&offline, true, true), None);
|
||||
assert_eq!(
|
||||
heal_control_retry_action(&Error::from(tonic::Status::permission_denied("bad signature")), false, false),
|
||||
None,
|
||||
"authorization failures must never be retried"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn heal_control_epoch_refresh_clears_offline_gate() {
|
||||
let client = test_peer_client();
|
||||
client.offline.store(true, Ordering::Release);
|
||||
|
||||
client
|
||||
.prepare_heal_control_retry(HealControlRetryAction::RefreshReplayScope)
|
||||
.await;
|
||||
|
||||
assert!(
|
||||
!client.offline.load(Ordering::Acquire),
|
||||
"epoch refresh must not leave the following attempt behind the offline gate"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn peer_rest_client_network_classifier_keeps_slow_peers_online() {
|
||||
// The per-RPC channel deadline (RUSTFS_INTERNODE_RPC_TIMEOUT, 30s)
|
||||
|
||||
@@ -7470,6 +7470,20 @@ impl PoolMeta {
|
||||
.is_some_and(is_decommission_suspended)
|
||||
}
|
||||
|
||||
pub(crate) fn has_active_decommission_capacity_reservation(&self, idx: usize) -> bool {
|
||||
self.pools
|
||||
.get(idx)
|
||||
.and_then(|pool| pool.decommission.as_ref())
|
||||
.is_some_and(|info| {
|
||||
info.has_decommission_state()
|
||||
&& is_decommission_active(info.complete, info.failed, info.canceled)
|
||||
&& info
|
||||
.capacity_reservation
|
||||
.as_ref()
|
||||
.is_some_and(DecommissionCapacityReservation::active)
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn scanner_pause_backlog_pool_writable(&self, idx: usize) -> bool {
|
||||
self.pools.get(idx).is_some_and(|pool| {
|
||||
!pool
|
||||
@@ -17595,6 +17609,17 @@ impl ECStore {
|
||||
}
|
||||
self.persist_decommission_durable_ilm_receipt(source_pool_idx, target_pool_idx, &receipt)
|
||||
.await?;
|
||||
self.decommission_durable_ilm_receipt_path_for_test(source_pool_idx, source_path, record)
|
||||
.await
|
||||
}
|
||||
|
||||
#[cfg(all(test, feature = "test-util"))]
|
||||
pub(crate) async fn decommission_durable_ilm_receipt_path_for_test(
|
||||
&self,
|
||||
source_pool_idx: usize,
|
||||
source_path: &str,
|
||||
record: &ValidatedDurableIlmRecord,
|
||||
) -> Result<String> {
|
||||
let run_token = self.durable_ilm_receipt_run_token(source_pool_idx).await?;
|
||||
Ok(decommission_durable_ilm_receipt_path(&run_token, source_path, record.id_kind, &record.id))
|
||||
}
|
||||
|
||||
@@ -41,6 +41,14 @@ struct DanglingDeleteGraceError {
|
||||
grace_secs: i64,
|
||||
}
|
||||
|
||||
/// Marks a conditional-file write that failed before its publication rename.
|
||||
/// Callers may choose another owner only while this marker is preserved; every
|
||||
/// unmarked error remains commit-ambiguous and must fail closed.
|
||||
#[derive(Debug)]
|
||||
struct ConditionalFileNotCommittedError {
|
||||
source: io::Error,
|
||||
}
|
||||
|
||||
// DiskError == StorageErr
|
||||
#[derive(Debug, thiserror::Error)]
|
||||
pub enum DiskError {
|
||||
@@ -220,6 +228,18 @@ impl std::fmt::Display for DanglingDeleteGraceError {
|
||||
|
||||
impl StdError for DanglingDeleteGraceError {}
|
||||
|
||||
impl std::fmt::Display for ConditionalFileNotCommittedError {
|
||||
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
|
||||
self.source.fmt(f)
|
||||
}
|
||||
}
|
||||
|
||||
impl StdError for ConditionalFileNotCommittedError {
|
||||
fn source(&self) -> Option<&(dyn StdError + 'static)> {
|
||||
Some(&self.source)
|
||||
}
|
||||
}
|
||||
|
||||
fn classify_internode_missing_error(error: &InternodeHttpError) -> Option<DiskError> {
|
||||
if error.is_remote_file_not_found() {
|
||||
return Some(DiskError::FileNotFound);
|
||||
@@ -293,6 +313,22 @@ impl DiskError {
|
||||
})
|
||||
}
|
||||
|
||||
pub(crate) fn conditional_file_not_committed(source: io::Error) -> io::Error {
|
||||
io::Error::new(source.kind(), ConditionalFileNotCommittedError { source })
|
||||
}
|
||||
|
||||
/// Whether a local conditional-file replacement failed before the target
|
||||
/// publication rename and therefore cannot have committed new owner bytes.
|
||||
pub fn is_conditional_file_not_committed(&self) -> bool {
|
||||
matches!(
|
||||
self,
|
||||
DiskError::Io(io_error)
|
||||
if io_error
|
||||
.get_ref()
|
||||
.is_some_and(|source| source.downcast_ref::<ConditionalFileNotCommittedError>().is_some())
|
||||
)
|
||||
}
|
||||
|
||||
pub fn is_dangling_delete_grace(&self) -> bool {
|
||||
matches!(self, DiskError::Io(io_error) if Self::io_error_is_dangling_delete_grace(io_error))
|
||||
}
|
||||
@@ -627,6 +663,9 @@ impl From<tokio::task::JoinError> for DiskError {
|
||||
impl Clone for DiskError {
|
||||
fn clone(&self) -> Self {
|
||||
match self {
|
||||
DiskError::Io(io_error) if self.is_conditional_file_not_committed() => DiskError::Io(
|
||||
DiskError::conditional_file_not_committed(io::Error::new(io_error.kind(), io_error.to_string())),
|
||||
),
|
||||
DiskError::Io(io_error) => DiskError::Io(
|
||||
rustfs_rio::clone_internode_http_io_error(io_error)
|
||||
.and_then(std::io::Error::into_inner)
|
||||
@@ -820,6 +859,21 @@ mod tests {
|
||||
use super::*;
|
||||
use std::collections::HashMap;
|
||||
|
||||
#[test]
|
||||
fn conditional_file_not_committed_marker_is_explicit_and_clone_safe() {
|
||||
let marked = DiskError::from(DiskError::conditional_file_not_committed(io::Error::new(
|
||||
io::ErrorKind::PermissionDenied,
|
||||
"staging rejected",
|
||||
)));
|
||||
assert!(marked.is_conditional_file_not_committed());
|
||||
assert!(marked.is_conditional_file_not_committed());
|
||||
assert!(!DiskError::Timeout.is_conditional_file_not_committed());
|
||||
assert!(
|
||||
!DiskError::Io(io::Error::new(io::ErrorKind::PermissionDenied, "rename rejected"))
|
||||
.is_conditional_file_not_committed()
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn terminal_read_error_preserves_kind_and_disk_classification() {
|
||||
let timeout = terminal_read_error_to_io(DiskError::Timeout);
|
||||
|
||||
@@ -8957,7 +8957,8 @@ impl DiskAPI for LocalDisk {
|
||||
.truncate(false)
|
||||
.read(true)
|
||||
.write(true)
|
||||
.open(&lock_path)?;
|
||||
.open(&lock_path)
|
||||
.map_err(DiskError::conditional_file_not_committed)?;
|
||||
flock(&lock, FlockOperation::NonBlockingLockExclusive).map_err(std::io::Error::from)?;
|
||||
let result = (|| {
|
||||
let current = match std::fs::read(&file_path) {
|
||||
@@ -9012,10 +9013,15 @@ impl DiskAPI for LocalDisk {
|
||||
.ok_or_else(|| std::io::Error::new(ErrorKind::InvalidInput, "conditional file has no parent"))?;
|
||||
let temporary = parent.join(format!(".{}.{}.tmp", path.replace('/', "_"), Uuid::new_v4()));
|
||||
let write_result = (|| -> std::io::Result<()> {
|
||||
let mut staged = std::fs::OpenOptions::new().create_new(true).write(true).open(&temporary)?;
|
||||
staged.write_all(&replacement)?;
|
||||
let not_committed = DiskError::conditional_file_not_committed;
|
||||
let mut staged = std::fs::OpenOptions::new()
|
||||
.create_new(true)
|
||||
.write(true)
|
||||
.open(&temporary)
|
||||
.map_err(not_committed)?;
|
||||
staged.write_all(&replacement).map_err(not_committed)?;
|
||||
if sync_metadata {
|
||||
staged.sync_all()?;
|
||||
staged.sync_all().map_err(not_committed)?;
|
||||
}
|
||||
std::fs::rename(&temporary, &file_path)?;
|
||||
Ok(())
|
||||
@@ -22650,6 +22656,10 @@ mod test {
|
||||
.await
|
||||
.expect_err("directory fsync failure must fail the CAS update");
|
||||
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
|
||||
assert!(
|
||||
!err.is_conditional_file_not_committed(),
|
||||
"an error after publication rename must remain commit-ambiguous"
|
||||
);
|
||||
assert_eq!(
|
||||
disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.await
|
||||
|
||||
@@ -18604,71 +18604,75 @@ mod put_object_tmp_cleanup_tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial(capacity_dirty_scope)]
|
||||
async fn cancelled_rename_keeps_namespace_lock_until_publication() {
|
||||
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
|
||||
let bucket = "put-commit-lock-cancelled-rename";
|
||||
let object = "commit-lock-cancelled-rename-object";
|
||||
for disk in &disk_stores {
|
||||
disk.make_volume(bucket).await.expect("bucket volume should be created");
|
||||
}
|
||||
|
||||
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
|
||||
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
|
||||
let first_store = Arc::clone(&set_disks);
|
||||
let first = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
|
||||
first_store
|
||||
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
|
||||
.await
|
||||
.expect("first PUT should pause during the authoritative rename");
|
||||
|
||||
let second_namespace_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
|
||||
let second_store = Arc::clone(&set_disks);
|
||||
let second = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![b'2'; TEST_OBJECT_SIZE]);
|
||||
second_store
|
||||
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
second_namespace_barrier.release_and_wait_until_namespace_pending().await;
|
||||
|
||||
first.abort();
|
||||
assert!(
|
||||
first
|
||||
.await
|
||||
.expect_err("the first request should be cancelled while rename is parked")
|
||||
.is_cancelled()
|
||||
);
|
||||
tokio::task::yield_now().await;
|
||||
assert!(
|
||||
!second.is_finished(),
|
||||
"the second writer must remain blocked by the cancelled commit owner"
|
||||
);
|
||||
|
||||
rename_barrier.release();
|
||||
drop(rename_barrier);
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
while rename_tasks.running() != 0 {
|
||||
tokio::task::yield_now().await;
|
||||
temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("false"))], async {
|
||||
let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
|
||||
let bucket = "put-commit-lock-cancelled-rename";
|
||||
let object = "commit-lock-cancelled-rename-object";
|
||||
for disk in &disk_stores {
|
||||
disk.make_volume(bucket).await.expect("bucket volume should be created");
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("the cancelled owner's rename fanout should drain");
|
||||
second
|
||||
.await
|
||||
.expect("second overwrite task should join")
|
||||
.expect("second overwrite should commit after the cancelled owner reaches publication");
|
||||
|
||||
let mut reader = set_disks
|
||||
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
let rename_tasks = rename_fanout_barrier::observe_tasks(object);
|
||||
let rename_barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
|
||||
let first_store = Arc::clone(&set_disks);
|
||||
let first = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
|
||||
first_store
|
||||
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
tokio::time::timeout(Duration::from_secs(30), rename_barrier.wait_until_paused())
|
||||
.await
|
||||
.expect("first PUT should pause during the authoritative rename");
|
||||
|
||||
let second_namespace_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
|
||||
let second_store = Arc::clone(&set_disks);
|
||||
let second = tokio::spawn(async move {
|
||||
let mut reader = PutObjReader::from_vec(vec![b'2'; TEST_OBJECT_SIZE]);
|
||||
second_store
|
||||
.put_object(bucket, object, &mut reader, &ObjectOptions::default())
|
||||
.await
|
||||
});
|
||||
second_namespace_barrier.release_and_wait_until_namespace_pending().await;
|
||||
|
||||
first.abort();
|
||||
assert!(
|
||||
first
|
||||
.await
|
||||
.expect_err("the first request should be cancelled while rename is parked")
|
||||
.is_cancelled()
|
||||
);
|
||||
tokio::task::yield_now().await;
|
||||
assert!(
|
||||
!second.is_finished(),
|
||||
"the second writer must remain blocked by the cancelled commit owner"
|
||||
);
|
||||
|
||||
rename_barrier.release();
|
||||
drop(rename_barrier);
|
||||
tokio::time::timeout(Duration::from_secs(30), async {
|
||||
while rename_tasks.running() != 0 {
|
||||
tokio::task::yield_now().await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("the latest overwrite should be readable");
|
||||
let mut body = Vec::new();
|
||||
reader.stream.read_to_end(&mut body).await.expect("latest body should drain");
|
||||
assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
|
||||
.expect("the cancelled owner's rename fanout should drain");
|
||||
second
|
||||
.await
|
||||
.expect("second overwrite task should join")
|
||||
.expect("second overwrite should commit after the cancelled owner reaches publication");
|
||||
|
||||
let mut reader = set_disks
|
||||
.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
|
||||
.await
|
||||
.expect("the latest overwrite should be readable");
|
||||
let mut body = Vec::new();
|
||||
reader.stream.read_to_end(&mut body).await.expect("latest body should drain");
|
||||
assert_eq!(body, vec![b'2'; TEST_OBJECT_SIZE]);
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -9643,7 +9643,13 @@ mod tests {
|
||||
com::save_config(store.pools[1].clone(), &manual_task_path, manual_task_bytes.clone())
|
||||
.await
|
||||
.expect("target task rewrite should invalidate cached metadata before the quorum check");
|
||||
let target_task_set = store.pools[1].get_disks_by_key(&manual_task_path);
|
||||
let manual_task_record =
|
||||
validate_durable_ilm_record(&manual_task_path, &manual_task_bytes).expect("manual task should validate");
|
||||
let manual_task_receipt_path = store
|
||||
.decommission_durable_ilm_receipt_path_for_test(0, &manual_task_path, &manual_task_record)
|
||||
.await
|
||||
.expect("manual task receipt path should resolve");
|
||||
let target_task_set = store.pools[1].get_disks_by_key(&manual_task_receipt_path);
|
||||
let original_target_task_disks = {
|
||||
let mut disks = target_task_set.disks.write().await;
|
||||
let original = disks.clone();
|
||||
|
||||
@@ -20,6 +20,7 @@ fn to_filemeta_err(err: Error) -> rustfs_filemeta::Error {
|
||||
err.narrow_to_filemeta().unwrap_or_else(rustfs_filemeta::Error::other)
|
||||
}
|
||||
|
||||
use crate::bucket::lifecycle::bucket_lifecycle_ops::free_version_remote_tuple_matches;
|
||||
use crate::bucket::metadata_sys::{
|
||||
get_versioning_config, has_authoritative_never_versioned_state, has_authoritative_never_versioned_state_in,
|
||||
};
|
||||
@@ -70,7 +71,7 @@ use tokio::io::duplex;
|
||||
use tokio::sync::broadcast::{self};
|
||||
use tokio::sync::mpsc::{self, Receiver, Sender};
|
||||
use tokio::sync::{OnceCell, RwLock};
|
||||
use tokio::task::JoinSet;
|
||||
use tokio::task::{JoinHandle, JoinSet};
|
||||
use tokio_util::sync::CancellationToken;
|
||||
use tracing::{Instrument, debug, error, info, warn};
|
||||
use uuid::Uuid;
|
||||
@@ -4331,6 +4332,7 @@ impl ECStore {
|
||||
"store list_merged started"
|
||||
);
|
||||
|
||||
let rx = rx.child_token();
|
||||
let mut futures = Vec::new();
|
||||
|
||||
let mut inputs = Vec::new();
|
||||
@@ -4346,16 +4348,10 @@ impl ECStore {
|
||||
}
|
||||
}
|
||||
|
||||
tokio::spawn(
|
||||
async move {
|
||||
if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await {
|
||||
error!("merge_entry_channels err {:?}", err)
|
||||
}
|
||||
}
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
let merge_task = spawn_listing_merge(rx, inputs, sender);
|
||||
|
||||
let results = join_all(futures).await;
|
||||
merge_task.await.map_err(Error::from)??;
|
||||
|
||||
let mut all_at_eof = true;
|
||||
|
||||
@@ -4422,6 +4418,7 @@ impl ECStore {
|
||||
) -> Result<()> {
|
||||
check_list_objs_args(bucket, prefix, &None)?;
|
||||
|
||||
let rx = rx.child_token();
|
||||
let mut futures = Vec::new();
|
||||
let mut inputs = Vec::new();
|
||||
|
||||
@@ -4783,17 +4780,11 @@ impl ECStore {
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
|
||||
tokio::spawn(
|
||||
async move {
|
||||
if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
|
||||
error!("merge_entry_channels err {:?}", err)
|
||||
}
|
||||
}
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
let merge_task = spawn_listing_merge(rx, inputs, merge_tx);
|
||||
|
||||
let walk_started = std::time::Instant::now();
|
||||
let walk_results = join_all(futures).await;
|
||||
merge_task.await.map_err(Error::from)??;
|
||||
let mut errs = Vec::new();
|
||||
for walk_result in walk_results {
|
||||
match walk_result {
|
||||
@@ -5068,6 +5059,130 @@ async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender<MetaCacheEn
|
||||
}
|
||||
}
|
||||
|
||||
/// Each input has already been resolved inside its own erasure set. This is a
|
||||
/// union of version histories, never a quorum vote between unrelated pools.
|
||||
fn merge_object_entry_versions(first: &mut MetaCacheEntry, others: impl Iterator<Item = MetaCacheEntry>) -> Result<()> {
|
||||
let name = first.name.clone();
|
||||
let mut versions: HashMap<(Option<Uuid>, bool), (FileMetaShallowVersion, ObjectInfo)> = HashMap::new();
|
||||
for mut entry in std::iter::once(std::mem::take(first)).chain(others) {
|
||||
let meta = match entry.cached.take() {
|
||||
Some(meta) => meta,
|
||||
None => FileMeta::load(&entry.metadata).map_err(|_| Error::FileCorrupt)?,
|
||||
};
|
||||
if meta.versions.is_empty() {
|
||||
return Err(Error::FileCorrupt);
|
||||
}
|
||||
for version in meta.versions {
|
||||
let parsed = version.parse_version_meta().map_err(|_| Error::FileCorrupt)?;
|
||||
if !parsed.valid() || parsed.version_type != version.header.version_type {
|
||||
return Err(Error::FileCorrupt);
|
||||
}
|
||||
let fi = parsed.into_fileinfo("", &name, true).map_err(|_| Error::FileCorrupt)?;
|
||||
let version_id = fi.version_id.filter(|id| !id.is_nil());
|
||||
if version_id != version.header.version_id.filter(|id| !id.is_nil())
|
||||
|| fi.mod_time != version.header.mod_time
|
||||
|| fi.tier_free_version() != version.header.free_version()
|
||||
{
|
||||
return Err(Error::FileCorrupt);
|
||||
}
|
||||
let info = ObjectInfo::from_file_info(&fi, "", &name, true);
|
||||
let identity = (version_id, version.header.free_version());
|
||||
match versions.entry(identity) {
|
||||
std::collections::hash_map::Entry::Vacant(slot) => {
|
||||
slot.insert((version, info));
|
||||
}
|
||||
std::collections::hash_map::Entry::Occupied(mut slot) => {
|
||||
let (previous, previous_info) = slot.get();
|
||||
// Suspended and unversioned writes replace the one null
|
||||
// slot. Distinct UUID versions never supersede each other.
|
||||
if version_id.is_none() && info.mod_time != previous_info.mod_time {
|
||||
if info.mod_time > previous_info.mod_time {
|
||||
slot.insert((version, info));
|
||||
}
|
||||
continue;
|
||||
}
|
||||
let equivalent = if info.delete_marker && previous_info.delete_marker {
|
||||
super::object::is_equivalent_data_movement_delete_marker(&info, previous_info)
|
||||
} else {
|
||||
crate::data_movement::is_equivalent_data_movement_object_identity(&info, previous_info, true, true)
|
||||
};
|
||||
if !equivalent {
|
||||
return Err(Error::FileCorrupt);
|
||||
}
|
||||
// Equivalent migrated copies can have different coding or
|
||||
// data directories. Choose a stable representation without
|
||||
// making input order part of the S3 version order.
|
||||
if version.meta < previous.meta {
|
||||
slot.insert((version, info));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
let mut live_remote_references = HashMap::<String, HashMap<String, Vec<ObjectInfo>>>::new();
|
||||
for (_, info) in versions.values() {
|
||||
if !info.transitioned_object.free_version && info.transitioned_object.status == rustfs_filemeta::TRANSITION_COMPLETE {
|
||||
live_remote_references
|
||||
.entry(info.transitioned_object.tier.clone())
|
||||
.or_default()
|
||||
.entry(info.transitioned_object.name.clone())
|
||||
.or_default()
|
||||
.push(info.clone());
|
||||
}
|
||||
}
|
||||
// Keep cleanup durable in its source xl.meta, but do not expose it to a
|
||||
// merged recovery walk while another physical pool still owns the tuple.
|
||||
versions.retain(|_, (_, info)| {
|
||||
!info.transitioned_object.free_version
|
||||
|| !live_remote_references
|
||||
.get(info.transitioned_object.tier.as_str())
|
||||
.and_then(|by_name| by_name.get(info.transitioned_object.name.as_str()))
|
||||
.is_some_and(|candidates| {
|
||||
candidates
|
||||
.iter()
|
||||
.any(|live| free_version_remote_tuple_matches(info, live).unwrap_or(false))
|
||||
})
|
||||
});
|
||||
let mut merged = FileMeta::new();
|
||||
merged.versions = versions.into_values().map(|(version, _)| version).collect();
|
||||
merged.versions.sort_by(|a, b| {
|
||||
if a.header.sorts_before(&b.header) {
|
||||
std::cmp::Ordering::Less
|
||||
} else if b.header.sorts_before(&a.header) {
|
||||
std::cmp::Ordering::Greater
|
||||
} else {
|
||||
std::cmp::Ordering::Equal
|
||||
}
|
||||
});
|
||||
let metadata = merged.marshal_msg()?;
|
||||
*first = MetaCacheEntry {
|
||||
name,
|
||||
metadata,
|
||||
cached: Some(merged),
|
||||
reusable: true,
|
||||
};
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `rx` is private to the producers. Cancelling it on a merge error must not
|
||||
/// cancel the request token, which would suppress that error at the API edge.
|
||||
fn spawn_listing_merge(
|
||||
rx: CancellationToken,
|
||||
inputs: Vec<Receiver<MetaCacheEntry>>,
|
||||
sender: Sender<MetaCacheEntry>,
|
||||
) -> JoinHandle<Result<()>> {
|
||||
tokio::spawn(
|
||||
async move {
|
||||
let result = merge_entry_channels(rx.clone(), inputs, sender, 1).await;
|
||||
if result.is_err() {
|
||||
rx.cancel();
|
||||
}
|
||||
result
|
||||
}
|
||||
.instrument(tracing::Span::current()),
|
||||
)
|
||||
}
|
||||
|
||||
async fn merge_entry_channels(
|
||||
rx: CancellationToken,
|
||||
in_channels: Vec<Receiver<MetaCacheEntry>>,
|
||||
@@ -5133,6 +5248,7 @@ async fn merge_entry_channels(
|
||||
// after anything greater has been emitted).
|
||||
let mut last_emitted = String::new();
|
||||
let mut group: Vec<Box<MergeHead>> = Vec::new();
|
||||
let mut object_entries: Vec<MetaCacheEntry> = Vec::new();
|
||||
let mut refill: Vec<usize> = Vec::with_capacity(in_channels.len());
|
||||
|
||||
while let Some(Reverse(first)) = heap.pop() {
|
||||
@@ -5150,7 +5266,7 @@ async fn merge_entry_channels(
|
||||
// Resolve the same-name group to one winner (heads arrive in ascending
|
||||
// channel order):
|
||||
// - prefix dir vs prefix dir: the first (lowest channel) wins;
|
||||
// - object vs object: the later channel wins (legacy authority rule);
|
||||
// - object vs object: merge the independently resolved version stacks;
|
||||
// - object vs prefix dir: same-name means both end with the separator,
|
||||
// i.e. the object is an explicit "directory marker" for the same S3
|
||||
// key — it shadows the prefix dir so the key does not surface as
|
||||
@@ -5168,11 +5284,27 @@ async fn merge_entry_channels(
|
||||
if dir_winner.is_none() {
|
||||
dir_winner = Some(head);
|
||||
}
|
||||
} else if let Some(winner) = object_winner.as_ref() {
|
||||
// Key-only candidates carry no version metadata and cannot
|
||||
// replace a resolved stack or contribute a quorum vote.
|
||||
if head.entry.is_object() {
|
||||
if winner.entry.is_object() {
|
||||
object_entries.push(head.entry);
|
||||
} else {
|
||||
object_winner = Some(head);
|
||||
}
|
||||
}
|
||||
} else {
|
||||
object_winner = Some(head);
|
||||
}
|
||||
}
|
||||
|
||||
if !object_entries.is_empty()
|
||||
&& let Some(winner) = object_winner.as_mut()
|
||||
{
|
||||
merge_object_entry_versions(&mut winner.entry, object_entries.drain(..))?;
|
||||
}
|
||||
|
||||
if let Some(head) = object_winner.or(dir_winner)
|
||||
&& head.entry.name != last_emitted
|
||||
{
|
||||
@@ -5605,6 +5737,7 @@ impl Sets {
|
||||
"sets list_merged started"
|
||||
);
|
||||
|
||||
let rx = rx.child_token();
|
||||
let mut futures = Vec::new();
|
||||
let mut inputs = Vec::new();
|
||||
|
||||
@@ -5617,16 +5750,10 @@ impl Sets {
|
||||
futures.push(async move { set.list_path(rx_clone, opts, send).await });
|
||||
}
|
||||
|
||||
tokio::spawn(
|
||||
async move {
|
||||
if let Err(err) = merge_entry_channels(rx, inputs, sender.clone(), 1).await {
|
||||
error!("merge_entry_channels err {:?}", err);
|
||||
}
|
||||
}
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
let merge_task = spawn_listing_merge(rx, inputs, sender);
|
||||
|
||||
let results = join_all(futures).await;
|
||||
merge_task.await.map_err(Error::from)??;
|
||||
let mut all_at_eof = true;
|
||||
let mut errs = Vec::new();
|
||||
for result in results {
|
||||
@@ -5677,6 +5804,7 @@ impl Sets {
|
||||
) -> Result<()> {
|
||||
check_list_objs_args(bucket, prefix, &None)?;
|
||||
|
||||
let rx = rx.child_token();
|
||||
let mut futures = Vec::new();
|
||||
let mut inputs = Vec::new();
|
||||
|
||||
@@ -6007,17 +6135,11 @@ impl Sets {
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
|
||||
tokio::spawn(
|
||||
async move {
|
||||
if let Err(err) = merge_entry_channels(rx, inputs, merge_tx, 1).await {
|
||||
error!("merge_entry_channels err {:?}", err)
|
||||
}
|
||||
}
|
||||
.instrument(tracing::Span::current()),
|
||||
);
|
||||
let merge_task = spawn_listing_merge(rx, inputs, merge_tx);
|
||||
|
||||
let walk_started = std::time::Instant::now();
|
||||
let walk_results = join_all(futures).await;
|
||||
merge_task.await.map_err(Error::from)??;
|
||||
let mut errs = Vec::new();
|
||||
for walk_result in walk_results {
|
||||
match walk_result {
|
||||
@@ -7016,7 +7138,7 @@ mod test {
|
||||
};
|
||||
use crate::cache_value::metacache_set::{FallbackClaimTracker, TestReaderBehavior, list_path_raw};
|
||||
use crate::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, error::DiskError, new_disk};
|
||||
use crate::error::StorageError;
|
||||
use crate::error::{Result, StorageError};
|
||||
use crate::object_api::ObjectInfo;
|
||||
use rustfs_filemeta::{
|
||||
FileInfo, FileMeta, FileMetaVersion, MetaCacheEntries, MetaCacheEntriesSorted, MetaCacheEntry, MetaDeleteMarker,
|
||||
@@ -7338,6 +7460,47 @@ mod test {
|
||||
}
|
||||
}
|
||||
|
||||
fn test_transitioned_meta_entry(name: &str, remote_object: &str, delete_source: bool) -> MetaCacheEntry {
|
||||
let mut source = FileInfo::new(name, 2, 2);
|
||||
source.volume = "bucket".to_string();
|
||||
source.name = name.to_string();
|
||||
source.version_id = Some(Uuid::from_u128(1));
|
||||
source.versioned = true;
|
||||
source.size = 1;
|
||||
source.mod_time = Some(time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"));
|
||||
source.transition_status = rustfs_filemeta::TRANSITION_COMPLETE.to_string();
|
||||
source.transition_tier = "WARM".to_string();
|
||||
source.transitioned_objname = remote_object.to_string();
|
||||
source.transition_version = Some("remote-version".to_string());
|
||||
source.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact;
|
||||
rustfs_utils::http::metadata_compat::insert_str(
|
||||
&mut source.metadata,
|
||||
rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
|
||||
"00".repeat(32),
|
||||
);
|
||||
|
||||
let mut meta = FileMeta::new();
|
||||
meta.add_version(source.clone())
|
||||
.expect("test metadata should accept transitioned source");
|
||||
if delete_source {
|
||||
let mut delete = FileInfo {
|
||||
name: name.to_string(),
|
||||
version_id: source.version_id,
|
||||
..Default::default()
|
||||
};
|
||||
delete.set_tier_free_version_id(&Uuid::from_u128(2).to_string());
|
||||
meta.delete_version(&delete)
|
||||
.expect("transitioned delete should create a free-version owner");
|
||||
}
|
||||
let metadata = meta.marshal_msg().expect("test transitioned metadata should marshal");
|
||||
MetaCacheEntry {
|
||||
name: name.to_string(),
|
||||
metadata,
|
||||
cached: Some(meta),
|
||||
reusable: false,
|
||||
}
|
||||
}
|
||||
|
||||
fn test_object_with_delete_marker_meta_entry(
|
||||
name: &str,
|
||||
object_mod_time: time::OffsetDateTime,
|
||||
@@ -10399,7 +10562,7 @@ mod test {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_documents_candidate_metadata_authority_risk() {
|
||||
async fn merge_entry_channels_preserves_cross_pool_delete_marker_versions() {
|
||||
let (tx_a, rx_a) = mpsc::channel(4);
|
||||
let (tx_b, rx_b) = mpsc::channel(4);
|
||||
let (tx_c, rx_c) = mpsc::channel(4);
|
||||
@@ -10428,9 +10591,13 @@ mod test {
|
||||
.expect("merged entry should be present");
|
||||
assert_eq!(merged.name, "obj-a");
|
||||
assert!(
|
||||
!merged.is_latest_delete_marker(),
|
||||
"current merge consumes candidate metadata bytes; future index-backed strong modes must live-verify metadata instead"
|
||||
merged.is_latest_delete_marker(),
|
||||
"a newer marker must remain current across independently resolved pools"
|
||||
);
|
||||
let versions = merged.file_info_versions("bucket").expect("merged versions should decode");
|
||||
assert_eq!(versions.versions.len(), 2, "retain the historical object and deduplicate the marker");
|
||||
assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
|
||||
assert!(!versions.versions[1].deleted && !versions.versions[1].is_latest);
|
||||
assert!(
|
||||
matches!(timeout(Duration::from_secs(1), out_rx.recv()).await, Ok(None)),
|
||||
"merge should not emit a duplicate entry for the same key"
|
||||
@@ -10442,6 +10609,276 @@ mod test {
|
||||
.expect("merge task should succeed");
|
||||
}
|
||||
|
||||
fn rewrite_test_version(mut entry: MetaCacheEntry, change: impl FnOnce(&mut FileMetaVersion)) -> MetaCacheEntry {
|
||||
let meta = entry.cached.as_mut().expect("test metadata should be decoded");
|
||||
assert_eq!(meta.versions.len(), 1);
|
||||
let mut version = meta.versions[0].parse_version_meta().expect("test version should decode");
|
||||
change(&mut version);
|
||||
meta.versions[0] = version.try_into().expect("test version should encode");
|
||||
entry.metadata = meta.marshal_msg().expect("test metadata should encode");
|
||||
entry
|
||||
}
|
||||
|
||||
async fn merge_test_object_entries(entries: Vec<MetaCacheEntry>) -> Result<MetaCacheEntry> {
|
||||
let mut inputs = Vec::with_capacity(entries.len());
|
||||
for entry in entries {
|
||||
let (sender, receiver) = mpsc::channel(1);
|
||||
sender.send(entry).await.expect("fixture entry should queue");
|
||||
inputs.push(receiver);
|
||||
}
|
||||
let (sender, mut receiver) = mpsc::channel(1);
|
||||
let task = tokio::spawn(merge_entry_channels(CancellationToken::new(), inputs, sender, 1));
|
||||
let entry = receiver.recv().await;
|
||||
task.await.expect("merge must not panic")?;
|
||||
Ok(entry.expect("a valid same-key group must produce an entry"))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_orders_complete_histories_independently_of_pool_order() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "first", 4, 2)]);
|
||||
let second = rewrite_test_version(
|
||||
test_object_meta_entry_with_erasure_versions("key", &[(time, "second", 4, 2)]),
|
||||
|version| version.object.as_mut().expect("object version").version_id = Some(Uuid::from_u128(2)),
|
||||
);
|
||||
let marker = test_delete_marker_meta_entry("key", time + time::Duration::seconds(1));
|
||||
let inputs = [first, second, marker];
|
||||
let mut expected = None;
|
||||
for order in [[0, 1, 2], [0, 2, 1], [1, 0, 2], [1, 2, 0], [2, 0, 1], [2, 1, 0]] {
|
||||
let entry = merge_test_object_entries(order.map(|index| inputs[index].clone()).to_vec())
|
||||
.await
|
||||
.expect("disjoint version chains should merge");
|
||||
let versions = entry.file_info_versions("bucket").expect("merged versions should decode");
|
||||
assert_eq!(versions.versions.len(), 3);
|
||||
assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
|
||||
assert!(
|
||||
versions.versions[1..]
|
||||
.iter()
|
||||
.all(|version| !version.deleted && !version.is_latest)
|
||||
);
|
||||
assert!(versions.versions.iter().all(|version| version.num_versions == 3));
|
||||
let identities = versions.versions.iter().map(|version| version.version_id).collect::<Vec<_>>();
|
||||
assert!(identities.contains(&Some(Uuid::from_u128(1))));
|
||||
assert!(identities.contains(&Some(Uuid::from_u128(2))));
|
||||
if let Some(expected) = &expected {
|
||||
assert_eq!(&identities, expected, "equal-time versions must have stable pagination order");
|
||||
} else {
|
||||
expected = Some(identities);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_key_only_candidates_do_not_override_version_metadata() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let marker = test_delete_marker_meta_entry("key", time);
|
||||
for entries in [
|
||||
vec![test_meta_entry("key"), marker.clone()],
|
||||
vec![marker.clone(), test_meta_entry("key")],
|
||||
] {
|
||||
let mut merged = merge_test_object_entries(entries)
|
||||
.await
|
||||
.expect("merge a name with resolved metadata");
|
||||
assert!(merged.is_latest_delete_marker());
|
||||
assert_eq!(merged.file_info_versions("bucket").expect("decode marker").versions.len(), 1);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_accepts_equivalent_migrated_coding_and_data_dirs() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let first = test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 4, 2)]);
|
||||
let second = rewrite_test_version(
|
||||
test_object_meta_entry_with_erasure_versions("key", &[(time, "same-etag", 6, 2)]),
|
||||
|version| version.object.as_mut().expect("object version").data_dir = Some(Uuid::from_u128(42)),
|
||||
);
|
||||
let forward = merge_test_object_entries(vec![first.clone(), second.clone()])
|
||||
.await
|
||||
.expect("valid migration copies");
|
||||
let reverse = merge_test_object_entries(vec![second, first])
|
||||
.await
|
||||
.expect("reversed migration copies");
|
||||
assert_eq!(forward.metadata, reverse.metadata, "representation must not depend on channel order");
|
||||
let versions = forward.file_info_versions("bucket").expect("merged metadata should decode");
|
||||
assert_eq!(versions.versions.len(), 1);
|
||||
assert_eq!(versions.versions[0].metadata.get("etag").map(String::as_str), Some("same-etag"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_defers_free_version_while_same_remote_source_is_live() {
|
||||
let live = test_transitioned_meta_entry("key", "remote/shared", false);
|
||||
let free = test_transitioned_meta_entry("key", "remote/shared", true);
|
||||
for inputs in [vec![live.clone(), free.clone()], vec![free.clone(), live.clone()]] {
|
||||
let merged = merge_test_object_entries(inputs)
|
||||
.await
|
||||
.expect("same remote source and cleanup owner should merge");
|
||||
let versions = merged
|
||||
.file_info_versions_with_free_versions("bucket")
|
||||
.expect("merged transition history should decode");
|
||||
assert_eq!(versions.versions.len(), 1);
|
||||
assert!(versions.free_versions.is_empty(), "a live remote reference must defer cleanup discovery");
|
||||
}
|
||||
|
||||
let unrelated = test_transitioned_meta_entry("key", "remote/other", false);
|
||||
let merged = merge_test_object_entries(vec![free, unrelated])
|
||||
.await
|
||||
.expect("unrelated remote references should merge");
|
||||
let versions = merged
|
||||
.file_info_versions_with_free_versions("bucket")
|
||||
.expect("merged transition history should decode");
|
||||
assert_eq!(versions.versions.len(), 1);
|
||||
assert_eq!(versions.free_versions.len(), 1, "an unrelated source must not suppress cleanup");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_rejects_conflicting_version_identity_and_metadata() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "original", 4, 2)]);
|
||||
for key in [
|
||||
"etag",
|
||||
"x-amz-tagging",
|
||||
"x-amz-object-lock-mode",
|
||||
"x-amz-object-lock-retain-until-date",
|
||||
] {
|
||||
let changed = rewrite_test_version(original.clone(), |version| {
|
||||
version
|
||||
.object
|
||||
.as_mut()
|
||||
.expect("object version")
|
||||
.meta_user
|
||||
.insert(key.to_string(), "changed".to_string());
|
||||
});
|
||||
for pair in [[original.clone(), changed.clone()], [changed, original.clone()]] {
|
||||
let err = merge_test_object_entries(pair.to_vec())
|
||||
.await
|
||||
.expect_err("conflicting copies must fail");
|
||||
assert_eq!(err, StorageError::FileCorrupt, "conflict in {key} must not become arbitrary metadata");
|
||||
}
|
||||
}
|
||||
let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time), |version| {
|
||||
version.delete_marker.as_mut().expect("delete marker").version_id = Some(Uuid::from_u128(1));
|
||||
});
|
||||
assert_eq!(
|
||||
merge_test_object_entries(vec![original, marker])
|
||||
.await
|
||||
.expect_err("UUID type conflict"),
|
||||
StorageError::FileCorrupt
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_reconciles_null_overwrite_without_losing_uuid_history() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let history = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]);
|
||||
let old_null = rewrite_test_version(history.clone(), |version| {
|
||||
version.object.as_mut().expect("null object").version_id = None;
|
||||
});
|
||||
let marker = rewrite_test_version(test_delete_marker_meta_entry("key", time + time::Duration::seconds(1)), |version| {
|
||||
version.delete_marker.as_mut().expect("null marker").version_id = Some(Uuid::nil());
|
||||
});
|
||||
for inputs in [
|
||||
vec![old_null.clone(), marker.clone(), history.clone()],
|
||||
vec![history, marker, old_null],
|
||||
] {
|
||||
let entry = merge_test_object_entries(inputs)
|
||||
.await
|
||||
.expect("new null slot should replace old null slot");
|
||||
let versions = entry.file_info_versions("bucket").expect("null versions should decode");
|
||||
assert_eq!(versions.versions.len(), 2);
|
||||
assert!(versions.versions[0].deleted && versions.versions[0].is_latest);
|
||||
assert!(versions.versions[0].version_id.is_none_or(|id| id.is_nil()));
|
||||
assert_eq!(versions.versions[1].version_id, Some(Uuid::from_u128(1)));
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_rejects_corrupt_version_headers_and_empty_stacks() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let original = test_object_meta_entry_with_erasure_versions("key", &[(time, "etag", 4, 2)]);
|
||||
for empty in [false, true] {
|
||||
let mut corrupt = original.clone();
|
||||
let meta = corrupt.cached.as_mut().expect("fixture metadata");
|
||||
if empty {
|
||||
meta.versions.clear();
|
||||
} else {
|
||||
meta.versions[0].header.version_id = Some(Uuid::from_u128(99));
|
||||
}
|
||||
corrupt.metadata = meta.marshal_msg().expect("encode corrupt fixture");
|
||||
assert_eq!(
|
||||
merge_test_object_entries(vec![original.clone(), corrupt])
|
||||
.await
|
||||
.expect_err("corrupt candidate must fail"),
|
||||
StorageError::FileCorrupt
|
||||
);
|
||||
}
|
||||
let mut malformed = original.clone();
|
||||
malformed.cached = None;
|
||||
malformed.metadata = vec![0xff];
|
||||
assert_eq!(
|
||||
merge_test_object_entries(vec![original, malformed])
|
||||
.await
|
||||
.expect_err("malformed metadata must fail"),
|
||||
StorageError::FileCorrupt
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_does_not_combine_subquorum_markers_across_erasure_sets() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let old = test_object_meta_entry_with_erasure_versions("key", &[(time, "history", 4, 2)]);
|
||||
let marked = test_object_with_delete_marker_meta_entry("key", time, time + time::Duration::seconds(1));
|
||||
let mut inputs = Vec::new();
|
||||
for marker_copies in [1, 2] {
|
||||
let resolver = list_metadata_resolution_params("bucket".to_string(), 3, 3, true, 0);
|
||||
let copies = (0..3)
|
||||
.map(|index| Some(if index < marker_copies { marked.clone() } else { old.clone() }))
|
||||
.collect();
|
||||
let entry = resolve_listing_entries(MetaCacheEntries(copies), resolver, false)
|
||||
.expect("each set independently retains its quorum-backed history");
|
||||
inputs.push(entry);
|
||||
}
|
||||
let merged = merge_test_object_entries(inputs).await.expect("merge resolved histories");
|
||||
let versions = merged.file_info_versions("bucket").expect("decode merged history");
|
||||
assert_eq!(
|
||||
versions.versions.len(),
|
||||
1,
|
||||
"three marker copies across two EC domains do not form a quorum"
|
||||
);
|
||||
assert!(!versions.versions[0].deleted);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn listing_merge_preserves_error_after_partial_output_without_cancelling_request() {
|
||||
let time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp");
|
||||
let (first_tx, first_rx) = mpsc::channel(2);
|
||||
let (second_tx, second_rx) = mpsc::channel(1);
|
||||
first_tx.send(test_meta_entry("a/")).await.expect("queue preceding prefix");
|
||||
first_tx
|
||||
.send(test_object_meta_entry_with_erasure_versions("b", &[(time, "one", 4, 2)]))
|
||||
.await
|
||||
.expect("queue first copy");
|
||||
second_tx
|
||||
.send(test_object_meta_entry_with_erasure_versions("b", &[(time, "two", 4, 2)]))
|
||||
.await
|
||||
.expect("queue conflicting copy");
|
||||
drop(first_tx);
|
||||
drop(second_tx);
|
||||
let request = CancellationToken::new();
|
||||
let workers = request.child_token();
|
||||
let (sender, mut receiver) = mpsc::channel(1);
|
||||
let task = super::spawn_listing_merge(workers.clone(), vec![first_rx, second_rx], sender);
|
||||
assert_eq!(receiver.recv().await.expect("preceding result should arrive").name, "a/");
|
||||
assert!(receiver.recv().await.is_none());
|
||||
assert_eq!(
|
||||
task.await
|
||||
.expect("merge task must not panic")
|
||||
.expect_err("conflict must propagate"),
|
||||
StorageError::FileCorrupt
|
||||
);
|
||||
assert!(workers.is_cancelled(), "failed merge must stop the disk producers");
|
||||
assert!(!request.is_cancelled(), "the API must still observe the merge error");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn merge_entry_channels_handles_single_channel() {
|
||||
let (tx, rx) = mpsc::channel(4);
|
||||
|
||||
@@ -2013,7 +2013,7 @@ fn effective_object_actual_size(info: &ObjectInfo) -> Option<i64> {
|
||||
info.get_actual_size().ok()
|
||||
}
|
||||
|
||||
fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool {
|
||||
pub(super) fn is_equivalent_data_movement_delete_marker(source: &ObjectInfo, target: &ObjectInfo) -> bool {
|
||||
is_data_movement_delete_marker(source)
|
||||
&& is_data_movement_delete_marker(target)
|
||||
&& source.version_id == target.version_id
|
||||
@@ -4791,7 +4791,13 @@ impl ECStore {
|
||||
return Ok(ObjectInfo::default());
|
||||
}
|
||||
|
||||
let gopts = delete_pool_lookup_opts(&opts, true);
|
||||
let creates_latest_marker = should_create_delete_marker_for_missing_object(&opts);
|
||||
let mut gopts = delete_pool_lookup_opts(&opts, true);
|
||||
if creates_latest_marker {
|
||||
// An unwritable source still owns its current version. Hiding it
|
||||
// during lookup would turn a rejected write into a new-pool marker.
|
||||
gopts.skip_rebalancing = false;
|
||||
}
|
||||
|
||||
if opts.data_movement {
|
||||
let existing_pool_info = self.get_pool_info_existing_with_opts(bucket, object, &gopts).await;
|
||||
@@ -4917,7 +4923,12 @@ impl ECStore {
|
||||
}
|
||||
|
||||
// Determine which pool contains it
|
||||
let (mut pinfo, errs) = match self.get_pool_info_existing_with_opts(bucket, object, &gopts).await {
|
||||
let existing_pool_info = if creates_latest_marker {
|
||||
self.get_pool_info_for_delete_marker(bucket, object, &gopts).await
|
||||
} else {
|
||||
self.get_pool_info_existing_with_opts(bucket, object, &gopts).await
|
||||
};
|
||||
let (mut pinfo, errs) = match existing_pool_info {
|
||||
Ok(res) => res,
|
||||
Err(err) if is_err_read_quorum(&err) => return Err(StorageError::ErasureWriteQuorum),
|
||||
Err(err) if is_err_object_not_found(&err) && should_create_delete_marker_for_missing_object(&opts) => {
|
||||
@@ -4954,7 +4965,18 @@ impl ECStore {
|
||||
}
|
||||
};
|
||||
|
||||
if pinfo.object_info.delete_marker && opts.version_id.is_none() {
|
||||
if creates_latest_marker && self.is_suspended(pinfo.index).await {
|
||||
let has_active_reservation = self
|
||||
.pool_meta
|
||||
.read()
|
||||
.await
|
||||
.has_active_decommission_capacity_reservation(pinfo.index);
|
||||
if has_active_reservation {
|
||||
pinfo.index = self.get_pool_idx_no_lock(bucket, object, 0).await?;
|
||||
}
|
||||
}
|
||||
|
||||
if pinfo.object_info.delete_marker && opts.version_id.is_none() && !creates_latest_marker {
|
||||
pinfo.object_info.name = decode_dir_object(object);
|
||||
return Ok(pinfo.object_info);
|
||||
}
|
||||
@@ -4976,7 +4998,13 @@ impl ECStore {
|
||||
}
|
||||
|
||||
for pool in self.pools.iter() {
|
||||
if creates_latest_marker && pool.pool_idx != pinfo.index {
|
||||
continue;
|
||||
}
|
||||
if self.is_suspended(pool.pool_idx).await || self.is_pool_rebalancing(pool.pool_idx).await {
|
||||
if creates_latest_marker {
|
||||
return Err(StorageError::SlowDown);
|
||||
}
|
||||
continue;
|
||||
}
|
||||
|
||||
@@ -5001,7 +5029,7 @@ impl ECStore {
|
||||
return Ok(obj);
|
||||
}
|
||||
Err(err) => {
|
||||
if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) {
|
||||
if creates_latest_marker || (!is_err_object_not_found(&err) && !is_err_version_not_found(&err)) {
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
@@ -8291,6 +8319,546 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
async fn multipool_version_test_store(bucket: &str) -> (Vec<tempfile::TempDir>, Arc<ECStore>) {
|
||||
let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
|
||||
let (mut dirs, first_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
|
||||
let (second_dirs, second_set) = make_local_set_disks_with_ctx(4, 2, Arc::clone(&ctx)).await;
|
||||
dirs.extend(second_dirs);
|
||||
let store = Arc::new(new_prepared_reader_test_store_with_ctx(&[first_set, second_set], ctx).await);
|
||||
crate::bucket::metadata_sys::init_bucket_metadata_sys(Arc::clone(&store), Vec::new()).await;
|
||||
store
|
||||
.handle_make_bucket(
|
||||
bucket,
|
||||
&MakeBucketOptions {
|
||||
versioning_enabled: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("create the versioned bucket in both pools");
|
||||
(dirs, store)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipool_delete_marker_stays_with_existing_versions() {
|
||||
let bucket = "multipool-marker-routing";
|
||||
let object = "history.bin";
|
||||
let (_dirs, store) = multipool_version_test_store(bucket).await;
|
||||
|
||||
let mut expected_versions = Vec::new();
|
||||
for value in 1..=3_u8 {
|
||||
let written = store.pools[1]
|
||||
.put_object(
|
||||
bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![value; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
user_defined: HashMap::from([
|
||||
(rustfs_utils::http::AMZ_OBJECT_TAGGING.to_string(), format!("generation={value}")),
|
||||
("x-amz-meta-generation".to_string(), value.to_string()),
|
||||
]),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("write a historical version deterministically to pool 1");
|
||||
expected_versions.push(written.version_id.expect("versioned PUT must acknowledge a UUID"));
|
||||
}
|
||||
let marker = store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("delete the current version");
|
||||
assert!(marker.delete_marker);
|
||||
let marker_id = marker.version_id.expect("DELETE must acknowledge a marker UUID");
|
||||
assert!(!expected_versions.contains(&marker_id));
|
||||
|
||||
let local = store.pools[1]
|
||||
.clone()
|
||||
.inner_list_object_versions(bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("list the object-owning pool");
|
||||
assert_eq!(local.objects.len(), 4, "the marker must be committed beside the three existing versions");
|
||||
assert_eq!(local.objects[0].version_id, Some(marker_id));
|
||||
assert!(local.objects[0].delete_marker && local.objects[0].is_latest);
|
||||
let versions = store
|
||||
.clone()
|
||||
.inner_list_object_versions(bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("list all pools");
|
||||
assert_eq!(versions.objects.len(), 4);
|
||||
assert_eq!(versions.objects.iter().filter(|version| version.is_latest).count(), 1);
|
||||
for (index, version_id) in expected_versions.iter().copied().enumerate() {
|
||||
assert!(
|
||||
versions
|
||||
.objects
|
||||
.iter()
|
||||
.any(|version| version.version_id == Some(version_id) && !version.is_latest)
|
||||
);
|
||||
let mut reader = store
|
||||
.handle_get_object_reader(
|
||||
bucket,
|
||||
object,
|
||||
None,
|
||||
HeaderMap::new(),
|
||||
&ObjectOptions {
|
||||
version_id: Some(version_id.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("historical version should remain readable");
|
||||
let mut payload = Vec::new();
|
||||
reader
|
||||
.stream
|
||||
.read_to_end(&mut payload)
|
||||
.await
|
||||
.expect("read all historical bytes");
|
||||
let value = u8::try_from(index + 1).expect("fixture generation fits u8");
|
||||
assert_eq!(payload, vec![value; 4097]);
|
||||
let listed = versions
|
||||
.objects
|
||||
.iter()
|
||||
.find(|version| version.version_id == Some(version_id))
|
||||
.expect("listed historical version");
|
||||
assert_eq!(listed.user_tags.as_str(), format!("generation={value}"));
|
||||
assert_eq!(listed.user_defined.get("x-amz-meta-generation"), Some(&value.to_string()));
|
||||
}
|
||||
|
||||
let repeated = store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("repeat simple DELETE");
|
||||
assert!(repeated.delete_marker);
|
||||
assert_ne!(
|
||||
repeated.version_id,
|
||||
Some(marker_id),
|
||||
"each enabled-versioning DELETE creates a new marker"
|
||||
);
|
||||
store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: repeated.version_id.map(|id| id.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("remove the newest marker by identity");
|
||||
store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(marker_id.to_string()),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("remove the original marker by identity");
|
||||
let current = store
|
||||
.get_object_info(bucket, object, &ObjectOptions::default())
|
||||
.await
|
||||
.expect("previous version becomes current");
|
||||
assert_eq!(current.version_id, expected_versions.last().copied());
|
||||
assert!(!current.delete_marker);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipool_existing_split_history_lists_and_paginates_without_metadata_writes() {
|
||||
for marker_pool in 0..2 {
|
||||
let bucket = format!("multipool-split-history-{marker_pool}");
|
||||
let object = "history.bin";
|
||||
let (dirs, store) = multipool_version_test_store(&bucket).await;
|
||||
let mut expected = Vec::new();
|
||||
for value in 1..=3_u8 {
|
||||
let version = store.pools[1 - marker_pool]
|
||||
.put_object(
|
||||
&bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![value; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed history through the owning pool's normal write path");
|
||||
expected.push(version.version_id);
|
||||
}
|
||||
let marker = store.pools[marker_pool]
|
||||
.delete_object(
|
||||
&bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("reproduce the previously committed split marker with a normal pool DELETE");
|
||||
assert!(marker.delete_marker);
|
||||
expected.push(marker.version_id);
|
||||
expected.reverse();
|
||||
let mut before = Vec::new();
|
||||
for dir in &dirs {
|
||||
before.push(
|
||||
tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta"))
|
||||
.await
|
||||
.expect("snapshot persisted version metadata"),
|
||||
);
|
||||
}
|
||||
for max_keys in [1, 2, 4, 10] {
|
||||
let mut key_marker = None;
|
||||
let mut version_marker = None;
|
||||
let mut listed = Vec::new();
|
||||
let mut completed = false;
|
||||
for _ in 0..6 {
|
||||
let page = store
|
||||
.clone()
|
||||
.inner_list_object_versions(&bucket, object, key_marker.clone(), version_marker.clone(), None, max_keys)
|
||||
.await
|
||||
.expect("read a complete merged version page");
|
||||
listed.extend(page.objects);
|
||||
if !page.is_truncated {
|
||||
completed = true;
|
||||
break;
|
||||
}
|
||||
assert_ne!(
|
||||
(&page.next_marker, &page.next_version_idmarker),
|
||||
(&key_marker, &version_marker),
|
||||
"version cursor must advance"
|
||||
);
|
||||
key_marker = page.next_marker;
|
||||
version_marker = page.next_version_idmarker;
|
||||
}
|
||||
assert!(completed, "pagination must terminate");
|
||||
assert_eq!(listed.iter().map(|version| version.version_id).collect::<Vec<_>>(), expected);
|
||||
assert!(listed[0].delete_marker && listed[0].is_latest);
|
||||
assert!(listed[1..].iter().all(|version| !version.delete_marker && !version.is_latest));
|
||||
}
|
||||
let visible = store
|
||||
.clone()
|
||||
.list_objects_generic(&bucket, "", None, None, 10, false)
|
||||
.await
|
||||
.expect("list current objects");
|
||||
assert!(visible.objects.is_empty(), "the global current marker hides the object");
|
||||
for (dir, before) in dirs.iter().zip(before) {
|
||||
assert_eq!(
|
||||
tokio::fs::read(dir.path().join(&bucket).join(object).join("xl.meta"))
|
||||
.await
|
||||
.expect("read unchanged metadata"),
|
||||
before
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipool_conflicting_version_metadata_fails_the_listing_request() {
|
||||
let bucket = "multipool-version-conflict";
|
||||
let (_dirs, store) = multipool_version_test_store(bucket).await;
|
||||
store.pools[0]
|
||||
.put_object(
|
||||
bucket,
|
||||
"a.bin",
|
||||
&mut PutObjReader::from_vec(b"preceding result".to_vec()),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed an entry before the conflict");
|
||||
let version_id = Uuid::new_v4();
|
||||
let mod_time = OffsetDateTime::now_utc();
|
||||
let mut etags = Vec::new();
|
||||
for (pool_idx, value) in [(0, 1), (1, 2)] {
|
||||
let written = store.pools[pool_idx]
|
||||
.put_object(
|
||||
bucket,
|
||||
"z.bin",
|
||||
&mut PutObjReader::from_vec(vec![value; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
version_id: Some(version_id.to_string()),
|
||||
mod_time: Some(mod_time),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("persist independent conflicting copies");
|
||||
assert_eq!(written.version_id, Some(version_id));
|
||||
etags.push(written.etag);
|
||||
}
|
||||
assert_ne!(etags[0], etags[1], "fixture must contain a semantic conflict");
|
||||
let err = store
|
||||
.clone()
|
||||
.inner_list_object_versions(bucket, "", None, None, None, 10)
|
||||
.await
|
||||
.expect_err("partial output must not hide the merge error");
|
||||
assert_eq!(err, StorageError::FileCorrupt);
|
||||
|
||||
let cancellation = tokio_util::sync::CancellationToken::new();
|
||||
let (sender, mut receiver) = tokio::sync::mpsc::channel(1);
|
||||
let walk = store
|
||||
.clone()
|
||||
.walk(cancellation.clone(), bucket, "", sender, WalkOptions::default());
|
||||
let drain = async { while receiver.recv().await.is_some() {} };
|
||||
let (walk_result, ()) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(walk, drain) })
|
||||
.await
|
||||
.expect("bounded walk output must drain and terminate on a merge error");
|
||||
assert_eq!(
|
||||
walk_result.expect_err("walk must report the same metadata conflict"),
|
||||
StorageError::FileCorrupt
|
||||
);
|
||||
assert!(!cancellation.is_cancelled(), "worker failure must not cancel the caller's request");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipool_marker_rejects_unwritable_owner_without_falling_back() {
|
||||
let bucket = "multipool-unwritable-owner";
|
||||
let object = "history.bin";
|
||||
let (_dirs, store) = multipool_version_test_store(bucket).await;
|
||||
store.pools[1]
|
||||
.put_object(
|
||||
bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![1; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed the nonzero owner");
|
||||
*store.pool_meta.write().await = PoolMeta {
|
||||
pools: vec![prepared_pool_test_status(0, false), prepared_pool_test_status(1, true)],
|
||||
..Default::default()
|
||||
};
|
||||
let err = store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect_err("suspended owner must reject marker creation");
|
||||
assert_eq!(err, StorageError::SlowDown);
|
||||
*store.pool_meta.write().await = PoolMeta::default();
|
||||
|
||||
let mut rebalancing = crate::services::rebalance::RebalanceStats {
|
||||
participating: true,
|
||||
..Default::default()
|
||||
};
|
||||
rebalancing.info.status = crate::services::rebalance::RebalStatus::Started;
|
||||
*store.rebalance_meta.write().await = Some(crate::services::rebalance::RebalanceMeta {
|
||||
pool_stats: vec![crate::services::rebalance::RebalanceStats::default(), rebalancing],
|
||||
..Default::default()
|
||||
});
|
||||
let error = store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
*store.rebalance_meta.write().await = None;
|
||||
assert_eq!(
|
||||
error.expect_err("rebalance must not hide the owner during marker lookup"),
|
||||
StorageError::SlowDown
|
||||
);
|
||||
|
||||
// Isolate object quorum from the bucket metadata preflight: disabling
|
||||
// a whole pool can otherwise fail before object ownership is looked up.
|
||||
let quorum_bucket = RUSTFS_META_BUCKET;
|
||||
store.pools[1]
|
||||
.put_object(
|
||||
quorum_bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![1; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed the object-quorum fixture");
|
||||
for pool in &store.pools {
|
||||
pool.put_object(
|
||||
quorum_bucket,
|
||||
"split.bin",
|
||||
&mut PutObjReader::from_vec(vec![2; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed another key with history in both pools");
|
||||
}
|
||||
|
||||
let owner = &store.pools[1].disk_set[0];
|
||||
let healthy = owner.disks.read().await.clone();
|
||||
for disk in owner.disks.write().await.iter_mut().skip(1) {
|
||||
*disk = None;
|
||||
}
|
||||
let error = store
|
||||
.delete_object(
|
||||
quorum_bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
let split_error = store
|
||||
.delete_object(
|
||||
quorum_bucket,
|
||||
"split.bin",
|
||||
ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
*owner.disks.write().await = healthy;
|
||||
assert_eq!(
|
||||
error.expect_err("subquorum owner must not become a new-pool marker"),
|
||||
StorageError::ErasureWriteQuorum
|
||||
);
|
||||
assert_eq!(
|
||||
split_error.expect_err("a readable older pool does not prove the global current version"),
|
||||
StorageError::ErasureWriteQuorum
|
||||
);
|
||||
let split = store.pools[0]
|
||||
.clone()
|
||||
.inner_list_object_versions(quorum_bucket, "split.bin", None, None, None, 10)
|
||||
.await
|
||||
.expect("inspect the readable older pool");
|
||||
assert_eq!(split.objects.len(), 1);
|
||||
assert!(!split.objects[0].delete_marker);
|
||||
let other = store.pools[0]
|
||||
.clone()
|
||||
.inner_list_object_versions(quorum_bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("inspect the other pool");
|
||||
assert!(other.objects.is_empty());
|
||||
let history = store.pools[1]
|
||||
.clone()
|
||||
.inner_list_object_versions(quorum_bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("inspect the restored owner");
|
||||
assert_eq!(history.objects.len(), 1);
|
||||
assert!(!history.objects[0].delete_marker);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn multipool_suspended_and_batch_markers_keep_the_null_slot_on_the_owner() {
|
||||
let bucket = "multipool-suspended-markers";
|
||||
let object = "history.bin";
|
||||
let (_dirs, store) = multipool_version_test_store(bucket).await;
|
||||
let original = store.pools[1]
|
||||
.put_object(
|
||||
bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![1; 4097]),
|
||||
&ObjectOptions {
|
||||
versioned: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("seed the UUID history in pool 1");
|
||||
store
|
||||
.update_bucket_metadata_config(
|
||||
bucket,
|
||||
crate::bucket::metadata::BUCKET_VERSIONING_CONFIG,
|
||||
b"<VersioningConfiguration><Status>Suspended</Status></VersioningConfiguration>".to_vec(),
|
||||
)
|
||||
.await
|
||||
.expect("persist suspended bucket versioning");
|
||||
let null = store
|
||||
.put_object(
|
||||
bucket,
|
||||
object,
|
||||
&mut PutObjReader::from_vec(vec![2; 4097]),
|
||||
&ObjectOptions {
|
||||
version_suspended: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("write the null version beside the history");
|
||||
assert!(null.version_id.is_none_or(|id| id.is_nil()));
|
||||
for _ in 0..2 {
|
||||
let marker = store
|
||||
.delete_object(
|
||||
bucket,
|
||||
object,
|
||||
ObjectOptions {
|
||||
version_suspended: true,
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("replace the null slot with a marker");
|
||||
assert!(marker.delete_marker);
|
||||
assert!(marker.version_id.is_none_or(|id| id.is_nil()));
|
||||
}
|
||||
let (deleted, errors) = store
|
||||
.delete_objects(
|
||||
bucket,
|
||||
vec![ObjectToDelete {
|
||||
object_name: object.to_string(),
|
||||
..Default::default()
|
||||
}],
|
||||
ObjectOptions::default(),
|
||||
)
|
||||
.await;
|
||||
assert!(errors.iter().all(Option::is_none), "batch DELETE should succeed: {errors:?}");
|
||||
assert_eq!(deleted.len(), 1);
|
||||
assert!(deleted[0].delete_marker);
|
||||
let versions = store.pools[1]
|
||||
.clone()
|
||||
.inner_list_object_versions(bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("inspect the owner after batch DELETE");
|
||||
assert_eq!(versions.objects.len(), 2, "only one null marker and the UUID history remain");
|
||||
assert!(versions.objects[0].delete_marker && versions.objects[0].is_latest);
|
||||
assert_eq!(versions.objects[1].version_id, original.version_id);
|
||||
let other = store.pools[0]
|
||||
.clone()
|
||||
.inner_list_object_versions(bucket, object, None, None, None, 10)
|
||||
.await
|
||||
.expect("inspect the unused pool");
|
||||
assert!(other.objects.is_empty());
|
||||
}
|
||||
|
||||
async fn assert_prepared_reader_blocks_writer(store: &ECStore, bucket: &str, object: &str) {
|
||||
assert_pool_writer_is_blocked(store, 0, bucket, object).await;
|
||||
}
|
||||
|
||||
@@ -611,7 +611,18 @@ impl ECStore {
|
||||
object: &str,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
|
||||
self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await
|
||||
self.internal_get_pool_info_existing_with_opts(bucket, object, opts, false)
|
||||
.await
|
||||
}
|
||||
|
||||
pub(super) async fn get_pool_info_for_delete_marker(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
opts: &ObjectOptions,
|
||||
) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
|
||||
self.internal_get_pool_info_existing_with_opts(bucket, object, opts, true)
|
||||
.await
|
||||
}
|
||||
|
||||
async fn internal_get_pool_info_existing_with_opts(
|
||||
@@ -619,6 +630,7 @@ impl ECStore {
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
opts: &ObjectOptions,
|
||||
require_all_pool_reads: bool,
|
||||
) -> Result<(PoolObjInfo, Vec<PoolErr>)> {
|
||||
let mut futures = Vec::new();
|
||||
for pool in self.pools.iter() {
|
||||
@@ -647,6 +659,15 @@ impl ECStore {
|
||||
});
|
||||
}
|
||||
Err(e) => {
|
||||
// A readable older pool cannot prove ownership of the
|
||||
// current version while another pool is unreadable. Check
|
||||
// both raw and object-scoped quorum errors before sorting.
|
||||
if require_all_pool_reads && !is_err_object_not_found(&e) && !is_err_version_not_found(&e) {
|
||||
return Err(match e {
|
||||
Error::ErasureReadQuorum | Error::InsufficientReadQuorum(_, _) => Error::ErasureWriteQuorum,
|
||||
err => err,
|
||||
});
|
||||
}
|
||||
ress.push(PoolObjInfo {
|
||||
index,
|
||||
err: Some(e),
|
||||
@@ -656,6 +677,34 @@ impl ECStore {
|
||||
}
|
||||
}
|
||||
|
||||
if require_all_pool_reads {
|
||||
let suspended_pools = {
|
||||
let pool_meta = self.pool_meta.read().await;
|
||||
(0..self.pools.len())
|
||||
.map(|idx| pool_meta.is_suspended(idx))
|
||||
.collect::<Vec<_>>()
|
||||
};
|
||||
let candidates = ress
|
||||
.iter()
|
||||
.map(|pinfo| LatestObjectInfoCandidate {
|
||||
info: pinfo.err.is_none().then(|| pinfo.object_info.clone()),
|
||||
idx: pinfo.index,
|
||||
err: pinfo.err.clone(),
|
||||
})
|
||||
.collect();
|
||||
let (object_info, index) =
|
||||
resolve_latest_object_info_candidates_with_pool_state(candidates, &suspended_pools, bucket, object, opts)?;
|
||||
let pools_with_object = self.pools_with_object(&ress, opts).await;
|
||||
return Ok((
|
||||
PoolObjInfo {
|
||||
index,
|
||||
object_info,
|
||||
err: None,
|
||||
},
|
||||
pools_with_object,
|
||||
));
|
||||
}
|
||||
|
||||
ress.sort_by(|a, b| {
|
||||
let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
|
||||
let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH);
|
||||
|
||||
@@ -520,32 +520,50 @@ impl RootHealRecovery {
|
||||
let _guard = self.mutation.lock().await;
|
||||
let disks = self.disks().await?;
|
||||
let existing = Self::find(&disks, &request.id).await?;
|
||||
let (disk, expected) = match existing {
|
||||
Some((disk, bytes)) => (disk, Some(bytes)),
|
||||
None => {
|
||||
let disk = disks
|
||||
.first()
|
||||
.cloned()
|
||||
.ok_or_else(|| Error::Other("No local disk available for root heal shutdown recovery".to_string()))?;
|
||||
(disk, None)
|
||||
}
|
||||
};
|
||||
if request.options.no_lock {
|
||||
return Err(Error::Other("Administrator root heal cannot skip namespace locking".to_string()));
|
||||
}
|
||||
let bytes = serde_json::to_vec(&RootHealIntent::from_request(request))
|
||||
.map_err(|error| Error::Other(format!("Serialize root heal recovery record: {error}")))?;
|
||||
match EcstoreDiskAPI::compare_and_update_file(
|
||||
disk.as_ref(),
|
||||
RUSTFS_META_BUCKET,
|
||||
&intent_path(&request.id)?,
|
||||
expected,
|
||||
Some(bytes.into()),
|
||||
)
|
||||
.await?
|
||||
{
|
||||
EcstoreConditionalFileUpdate::Updated => Ok(()),
|
||||
_ => Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))),
|
||||
let path = intent_path(&request.id)?;
|
||||
if let Some((disk, expected)) = existing {
|
||||
return match EcstoreDiskAPI::compare_and_update_file(
|
||||
disk.as_ref(),
|
||||
RUSTFS_META_BUCKET,
|
||||
&path,
|
||||
Some(expected),
|
||||
Some(bytes.into()),
|
||||
)
|
||||
.await?
|
||||
{
|
||||
EcstoreConditionalFileUpdate::Updated => Ok(()),
|
||||
_ => Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))),
|
||||
};
|
||||
}
|
||||
|
||||
if disks.is_empty() {
|
||||
return Err(Error::Other("No local disk available for root heal shutdown recovery".to_string()));
|
||||
}
|
||||
let mut last_not_committed = None;
|
||||
for disk in &disks {
|
||||
match EcstoreDiskAPI::compare_and_update_file(
|
||||
disk.as_ref(),
|
||||
RUSTFS_META_BUCKET,
|
||||
&path,
|
||||
None,
|
||||
Some(bytes.clone().into()),
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(EcstoreConditionalFileUpdate::Updated) => return Ok(()),
|
||||
Ok(_) => return Err(Error::Other(format!("Root heal recovery record changed for {}", request.id))),
|
||||
Err(error) if error.is_conditional_file_not_committed() => last_not_committed = Some(error),
|
||||
Err(error) => return Err(Error::Disk(error)),
|
||||
}
|
||||
}
|
||||
match last_not_committed {
|
||||
Some(error) => Err(Error::Disk(error)),
|
||||
None => Err(Error::Other("No local disk accepted the root heal recovery record".to_string())),
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -17,6 +17,44 @@ use super::*;
|
||||
use crate::heal::RUSTFS_META_BUCKET;
|
||||
use std::collections::HashSet;
|
||||
|
||||
#[cfg(unix)]
|
||||
struct RestoreDirectoryMode {
|
||||
path: std::path::PathBuf,
|
||||
mode: u32,
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
impl RestoreDirectoryMode {
|
||||
fn read_only(path: std::path::PathBuf) -> Self {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
let mode = std::fs::metadata(&path)
|
||||
.expect("metadata directory mode")
|
||||
.permissions()
|
||||
.mode();
|
||||
std::fs::set_permissions(&path, std::fs::Permissions::from_mode(0o555)).expect("make metadata directory read-only");
|
||||
Self { path, mode }
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
impl Drop for RestoreDirectoryMode {
|
||||
fn drop(&mut self) {
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
|
||||
let _ = std::fs::set_permissions(&self.path, std::fs::Permissions::from_mode(self.mode));
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
fn ordered_recovery_disks(first: DiskStore, second: DiskStore) -> (DiskStore, DiskStore) {
|
||||
if first.endpoint().to_string() <= second.endpoint().to_string() {
|
||||
(first, second)
|
||||
} else {
|
||||
(second, first)
|
||||
}
|
||||
}
|
||||
|
||||
async fn recovery_disk() -> (TempDir, DiskStore) {
|
||||
let temp = TempDir::new().expect("temporary root recovery disk");
|
||||
let endpoint = Endpoint::try_from(temp.path().to_string_lossy().as_ref()).expect("disk endpoint");
|
||||
@@ -78,6 +116,94 @@ fn completed_admin_status(heal_type: &HealType, completed_at: SystemTime) -> Com
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn root_recovery_new_intent_skips_prepublication_read_only_owner() {
|
||||
let (first_temp, first_disk) = recovery_disk().await;
|
||||
let (second_temp, second_disk) = recovery_disk().await;
|
||||
let first_endpoint = first_disk.endpoint().to_string();
|
||||
let (read_only_disk, writable_disk) = ordered_recovery_disks(first_disk, second_disk);
|
||||
let read_only_root = if read_only_disk.endpoint().to_string() == first_endpoint {
|
||||
first_temp.path()
|
||||
} else {
|
||||
second_temp.path()
|
||||
};
|
||||
let _restore = RestoreDirectoryMode::read_only(read_only_root.join(RUSTFS_META_BUCKET));
|
||||
let manager = recovery_manager(vec![read_only_disk.clone(), writable_disk.clone()]);
|
||||
let mut request = admin_request(HealType::Object {
|
||||
bucket: "bucket".to_string(),
|
||||
object: "object".to_string(),
|
||||
version_id: None,
|
||||
});
|
||||
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(request.clone())
|
||||
.await
|
||||
.expect("a writable local disk should own the admin heal intent");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
let path = format!("root-heal-{}.json", request.id);
|
||||
assert!(matches!(
|
||||
read_only_disk.read_all(RUSTFS_META_BUCKET, &path).await,
|
||||
Err(DiskError::FileNotFound)
|
||||
));
|
||||
assert!(writable_disk.read_all(RUSTFS_META_BUCKET, &path).await.is_ok());
|
||||
|
||||
request.retry_attempts = 1;
|
||||
manager
|
||||
.root_recovery
|
||||
.persist(&request)
|
||||
.await
|
||||
.expect("an existing fallback owner should remain updateable");
|
||||
let pending = manager.root_recovery.pending().await.expect("read the single durable owner");
|
||||
assert_eq!(pending.len(), 1);
|
||||
assert_eq!(pending[0].id, request.id);
|
||||
assert_eq!(pending[0].retry_attempts, 1);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn root_recovery_existing_owner_never_migrates_after_write_rejection() {
|
||||
let (first_temp, first_disk) = recovery_disk().await;
|
||||
let (second_temp, second_disk) = recovery_disk().await;
|
||||
let first_endpoint = first_disk.endpoint().to_string();
|
||||
let (owner_disk, alternate_disk) = ordered_recovery_disks(first_disk, second_disk);
|
||||
let owner_root = if owner_disk.endpoint().to_string() == first_endpoint {
|
||||
first_temp.path()
|
||||
} else {
|
||||
second_temp.path()
|
||||
};
|
||||
let manager = recovery_manager(vec![owner_disk.clone(), alternate_disk.clone()]);
|
||||
let mut request = root_request();
|
||||
manager
|
||||
.root_recovery
|
||||
.persist(&request)
|
||||
.await
|
||||
.expect("create the canonical owner");
|
||||
let path = format!("root-heal-{}.json", request.id);
|
||||
let committed = owner_disk
|
||||
.read_all(RUSTFS_META_BUCKET, &path)
|
||||
.await
|
||||
.expect("canonical owner bytes");
|
||||
let _restore = RestoreDirectoryMode::read_only(owner_root.join(RUSTFS_META_BUCKET));
|
||||
|
||||
request.retry_attempts = 1;
|
||||
assert!(
|
||||
manager.root_recovery.persist(&request).await.is_err(),
|
||||
"an existing owner write rejection must fail closed"
|
||||
);
|
||||
assert_eq!(
|
||||
owner_disk
|
||||
.read_all(RUSTFS_META_BUCKET, &path)
|
||||
.await
|
||||
.expect("original owner remains"),
|
||||
committed
|
||||
);
|
||||
assert!(matches!(
|
||||
alternate_disk.read_all(RUSTFS_META_BUCKET, &path).await,
|
||||
Err(DiskError::FileNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
async fn active_root(manager: &HealManager, request: HealRequest) -> Arc<HealTask> {
|
||||
let task = Arc::new(HealTask::from_request(request, manager.storage.clone()));
|
||||
*task.status.write().await = HealTaskStatus::Running;
|
||||
|
||||
+189
-27
@@ -2133,7 +2133,7 @@ impl DefaultObjectUsecase {
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
fn build_reader_blob<R>(
|
||||
async fn build_reader_blob<R>(
|
||||
reader: R,
|
||||
response_content_length: i64,
|
||||
request_id: &str,
|
||||
@@ -2144,10 +2144,12 @@ impl DefaultObjectUsecase {
|
||||
key: &str,
|
||||
lifecycle: GetObjectBodyLifecycle,
|
||||
resume: Option<GetObjectResumeControl<R>>,
|
||||
) -> StreamingBlob
|
||||
) -> S3Result<StreamingBlob>
|
||||
where
|
||||
R: AsyncRead + Send + Sync + Unpin + 'static,
|
||||
{
|
||||
use tokio::io::AsyncReadExt as _;
|
||||
|
||||
let streaming_blob_start = rustfs_io_metrics::get_stage_metrics_enabled().then(std::time::Instant::now);
|
||||
let expected = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX);
|
||||
let tuned_stream_buffer_size =
|
||||
@@ -2163,7 +2165,7 @@ impl DefaultObjectUsecase {
|
||||
);
|
||||
}
|
||||
let handoff_start = get_stage_metrics_enabled.then(std::time::Instant::now);
|
||||
let reader = GetObjectStreamingReader::new(
|
||||
let mut reader = GetObjectStreamingReader::new(
|
||||
reader,
|
||||
bucket,
|
||||
key,
|
||||
@@ -2174,6 +2176,17 @@ impl DefaultObjectUsecase {
|
||||
lifecycle,
|
||||
resume,
|
||||
);
|
||||
let mut prefix = [0_u8; 1];
|
||||
let prefix_len = if expected == 0 {
|
||||
0
|
||||
} else {
|
||||
reader
|
||||
.read_exact(&mut prefix)
|
||||
.await
|
||||
.map_err(|error| map_get_object_reader_error(StorageError::from(error)))?;
|
||||
1
|
||||
};
|
||||
let reader = std::io::Cursor::new(prefix).take(prefix_len).chain(reader);
|
||||
let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source)
|
||||
.with_diagnostics(bucket, key, request_id);
|
||||
let blob = StreamingBlob::new(stream);
|
||||
@@ -2187,7 +2200,7 @@ impl DefaultObjectUsecase {
|
||||
);
|
||||
}
|
||||
record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_STREAMING_BLOB, streaming_blob_start);
|
||||
blob
|
||||
Ok(blob)
|
||||
}
|
||||
|
||||
fn init_get_object_bootstrap(&self, bucket: &str, key: &str, request_id: &str) -> S3Result<GetObjectBootstrap> {
|
||||
@@ -3161,7 +3174,7 @@ impl DefaultObjectUsecase {
|
||||
let (stream_buffer_size, stream_strategy) =
|
||||
Self::select_stream_buffer_strategy(response_content_length, optimal_buffer_size, enable_readahead, has_range);
|
||||
record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_STREAM_STRATEGY, stream_strategy_start);
|
||||
return Ok(Self::build_reader_blob(
|
||||
return Self::build_reader_blob(
|
||||
final_stream,
|
||||
response_content_length,
|
||||
request_id,
|
||||
@@ -3172,7 +3185,8 @@ impl DefaultObjectUsecase {
|
||||
key,
|
||||
lifecycle,
|
||||
resume(info),
|
||||
));
|
||||
)
|
||||
.await;
|
||||
}
|
||||
|
||||
if let Some(buffered_body) = buffered_body {
|
||||
@@ -3239,7 +3253,7 @@ impl DefaultObjectUsecase {
|
||||
let (stream_buffer_size, stream_strategy) =
|
||||
Self::select_stream_buffer_strategy(response_content_length, optimal_buffer_size, enable_readahead, has_range);
|
||||
record_get_object_s3_handler_stage_duration(GET_OBJECT_STAGE_BODY_STREAM_STRATEGY, stream_strategy_start);
|
||||
Ok(Self::build_reader_blob(
|
||||
Self::build_reader_blob(
|
||||
final_stream,
|
||||
response_content_length,
|
||||
request_id,
|
||||
@@ -3250,7 +3264,8 @@ impl DefaultObjectUsecase {
|
||||
key,
|
||||
lifecycle,
|
||||
resume(info),
|
||||
))
|
||||
)
|
||||
.await
|
||||
}
|
||||
|
||||
#[allow(clippy::too_many_arguments)]
|
||||
@@ -5847,8 +5862,11 @@ mod tests {
|
||||
}
|
||||
|
||||
impl AsyncRead for ReadProbeReader {
|
||||
fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
|
||||
self.reads.fetch_add(1, AtomicOrdering::Relaxed);
|
||||
if buf.remaining() > 0 {
|
||||
buf.put_slice(b"x");
|
||||
}
|
||||
Poll::Ready(Ok(()))
|
||||
}
|
||||
}
|
||||
@@ -6448,12 +6466,11 @@ mod tests {
|
||||
)
|
||||
.await
|
||||
.expect("reservation bypass must construct the normal streaming fallback");
|
||||
let chunk = fallback_body
|
||||
.next()
|
||||
.await
|
||||
.expect("fallback stream must yield a body chunk")
|
||||
.expect("fallback stream must not fail");
|
||||
assert_eq!(chunk, Bytes::from_static(b"body"));
|
||||
let mut received = Vec::new();
|
||||
while let Some(chunk) = fallback_body.next().await {
|
||||
received.extend_from_slice(&chunk.expect("fallback stream must not fail"));
|
||||
}
|
||||
assert_eq!(received, b"body");
|
||||
assert!(fallback_reads.load(AtomicOrdering::Relaxed) > 0);
|
||||
assert_eq!(readers.load(AtomicOrdering::Relaxed), 0, "cold-fill materialization must remain unopened");
|
||||
assert_eq!(coordinator.active_session_count_for_test(), 0);
|
||||
@@ -7773,6 +7790,151 @@ mod tests {
|
||||
)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_reader_blob_rejects_quorum_failure_before_handoff() {
|
||||
let result = DefaultObjectUsecase::build_reader_blob(
|
||||
FailAtEndReader::new(
|
||||
b"",
|
||||
Some(std::io::Error::other(StorageError::InsufficientReadQuorum(
|
||||
"test-bucket".to_string(),
|
||||
"unavailable-object".to_string(),
|
||||
))),
|
||||
),
|
||||
5,
|
||||
"req-preheader-quorum",
|
||||
None,
|
||||
64,
|
||||
GetObjectStreamStrategy::Standard,
|
||||
"test-bucket",
|
||||
"unavailable-object",
|
||||
GetObjectBodyLifecycle::disabled(),
|
||||
None,
|
||||
)
|
||||
.await;
|
||||
|
||||
let error = result.expect_err("a read quorum failure before the first byte must reject response construction");
|
||||
assert_eq!(error.code(), &S3ErrorCode::Custom("SlowDownRead".into()));
|
||||
assert_eq!(error.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
|
||||
assert_eq!(error.message(), Some("Resource requested is unreadable, please reduce your request rate"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_reader_blob_preserves_primed_byte_for_full_and_range() {
|
||||
for (request_id, content_range) in [("req-preheader-full", None), ("req-preheader-range", Some("bytes 10-14/100"))] {
|
||||
let reads = Arc::new(AtomicUsize::new(0));
|
||||
let mut body = DefaultObjectUsecase::build_reader_blob(
|
||||
DataProbeReader {
|
||||
reads: Arc::clone(&reads),
|
||||
data: std::io::Cursor::new(b"hello".to_vec()),
|
||||
},
|
||||
5,
|
||||
request_id,
|
||||
content_range,
|
||||
64,
|
||||
GetObjectStreamStrategy::Standard,
|
||||
"test-bucket",
|
||||
"test-object",
|
||||
GetObjectBodyLifecycle::disabled(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("the first byte should be available before response handoff");
|
||||
|
||||
assert_eq!(reads.load(AtomicOrdering::Relaxed), 1, "response construction must prime one byte");
|
||||
let mut received = Vec::new();
|
||||
while let Some(chunk) = body.next().await {
|
||||
received.extend_from_slice(&chunk.expect("the primed body should remain readable"));
|
||||
}
|
||||
assert_eq!(received, b"hello", "the primed byte must be delivered exactly once");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_reader_blob_does_not_poll_empty_object() {
|
||||
let reads = Arc::new(AtomicUsize::new(0));
|
||||
let mut body = DefaultObjectUsecase::build_reader_blob(
|
||||
ReadProbeReader {
|
||||
reads: Arc::clone(&reads),
|
||||
},
|
||||
0,
|
||||
"req-empty-object",
|
||||
None,
|
||||
64,
|
||||
GetObjectStreamStrategy::Standard,
|
||||
"test-bucket",
|
||||
"empty-object",
|
||||
GetObjectBodyLifecycle::disabled(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("an empty response should not require a storage read");
|
||||
|
||||
assert_eq!(reads.load(AtomicOrdering::Relaxed), 0);
|
||||
assert!(body.next().await.is_none());
|
||||
assert_eq!(reads.load(AtomicOrdering::Relaxed), 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_reader_blob_leaves_later_failure_in_body_stream() {
|
||||
let mut body = DefaultObjectUsecase::build_reader_blob(
|
||||
FailAtEndReader::new(b"h", Some(std::io::Error::other("failure after handoff"))),
|
||||
5,
|
||||
"req-postheader-failure",
|
||||
None,
|
||||
64,
|
||||
GetObjectStreamStrategy::Standard,
|
||||
"test-bucket",
|
||||
"later-failure-object",
|
||||
GetObjectBodyLifecycle::disabled(),
|
||||
None,
|
||||
)
|
||||
.await
|
||||
.expect("the available first byte should allow response handoff");
|
||||
|
||||
let first = body
|
||||
.next()
|
||||
.await
|
||||
.expect("the primed byte must be present")
|
||||
.expect("the primed byte must be successful");
|
||||
assert_eq!(first, Bytes::from_static(b"h"));
|
||||
let error = body
|
||||
.next()
|
||||
.await
|
||||
.expect("the later read must produce a body result")
|
||||
.expect_err("a failure after the first byte must stay in the body stream");
|
||||
assert!(error.to_string().contains("failure after handoff"));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_reader_blob_resume_offset_includes_primed_byte() {
|
||||
let reopen_count = Arc::new(AtomicUsize::new(0));
|
||||
let control = counting_resume_control(Arc::clone(&reopen_count), |emitted| {
|
||||
assert_eq!(emitted, 1, "resume must start after the byte consumed before response handoff");
|
||||
Ok(FailAtEndReader::new(b"ello", None))
|
||||
});
|
||||
let mut body = DefaultObjectUsecase::build_reader_blob(
|
||||
FailAtEndReader::new(b"h", Some(relocation_read_error())),
|
||||
5,
|
||||
"req-preheader-resume-offset",
|
||||
None,
|
||||
64,
|
||||
GetObjectStreamStrategy::Standard,
|
||||
"test-bucket",
|
||||
"relocated-object",
|
||||
GetObjectBodyLifecycle::disabled(),
|
||||
Some(control),
|
||||
)
|
||||
.await
|
||||
.expect("the first byte should permit response handoff before relocation");
|
||||
|
||||
let mut received = Vec::new();
|
||||
while let Some(chunk) = body.next().await {
|
||||
received.extend_from_slice(&chunk.expect("resume should complete the body"));
|
||||
}
|
||||
assert_eq!(received, b"hello");
|
||||
assert_eq!(reopen_count.load(Ordering::Relaxed), 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn get_object_streaming_reader_resumes_after_relocation_error() {
|
||||
use tokio::io::AsyncReadExt;
|
||||
@@ -9041,7 +9203,7 @@ mod tests {
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_get_object_body_keeps_large_objects_on_streaming_path_without_preread() {
|
||||
async fn build_get_object_body_primes_large_stream_before_handoff() {
|
||||
let reads = Arc::new(AtomicUsize::new(0));
|
||||
let reader = ReadProbeReader {
|
||||
reads: Arc::clone(&reads),
|
||||
@@ -9074,13 +9236,13 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
reads.load(AtomicOrdering::Relaxed),
|
||||
0,
|
||||
"large-object response construction should not pre-read object data"
|
||||
1,
|
||||
"large-object response construction should prime exactly one byte"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn build_get_object_body_keeps_large_encrypted_objects_on_streaming_path_without_preread() {
|
||||
async fn build_get_object_body_primes_large_encrypted_stream_before_handoff() {
|
||||
let reads = Arc::new(AtomicUsize::new(0));
|
||||
let reader = ReadProbeReader {
|
||||
reads: Arc::clone(&reads),
|
||||
@@ -9113,8 +9275,8 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
reads.load(AtomicOrdering::Relaxed),
|
||||
0,
|
||||
"large encrypted object response construction should not pre-read object data"
|
||||
1,
|
||||
"large encrypted object response construction should prime exactly one byte"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -9284,8 +9446,8 @@ mod tests {
|
||||
assert_eq!(fill, rustfs_object_data_cache::ObjectDataCacheFillResult::SkippedSizeMismatch);
|
||||
assert_eq!(
|
||||
reads.load(AtomicOrdering::Relaxed),
|
||||
0,
|
||||
"size-mismatched rejected fill should construct the fallback stream without pre-reading"
|
||||
1,
|
||||
"size-mismatched rejected fill should prime the fallback stream before handoff"
|
||||
);
|
||||
assert!(
|
||||
matches!(lookup_after_mismatch, rustfs_object_data_cache::ObjectDataCacheLookup::Miss),
|
||||
@@ -9993,8 +10155,8 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
reads.load(AtomicOrdering::Relaxed),
|
||||
0,
|
||||
"too-large materialize-fill candidate must not pre-read the fallback reader"
|
||||
1,
|
||||
"too-large materialize-fill candidate must prime the streaming fallback"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -10032,8 +10194,8 @@ mod tests {
|
||||
|
||||
assert_eq!(
|
||||
reads.load(AtomicOrdering::Relaxed),
|
||||
0,
|
||||
"default GetObject response construction should not pre-read small plain object data"
|
||||
1,
|
||||
"default GetObject response construction should prime exactly one byte"
|
||||
);
|
||||
}
|
||||
|
||||
|
||||
@@ -465,7 +465,7 @@ mod bucket_default_sse_lookup_tests {
|
||||
let err = classify_bucket_default_sse_lookup("bucket", Err(StorageError::ErasureReadQuorum))
|
||||
.expect_err("an unreadable metadata subsystem must never degrade to plaintext");
|
||||
|
||||
assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
|
||||
assert_eq!(err.code(), &S3ErrorCode::Custom("SlowDownRead".into()));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
+25
-6
@@ -20,6 +20,8 @@ use s3s::{S3Error, S3ErrorCode};
|
||||
|
||||
const MAX_VERSIONS_EXCEEDED_CODE: &str = "MaxVersionsExceeded";
|
||||
const MAX_VERSIONS_EXCEEDED_MESSAGE: &str = "You've exceeded the limit on the number of versions you can create on this object";
|
||||
const SLOW_DOWN_READ_CODE: &str = "SlowDownRead";
|
||||
const SLOW_DOWN_READ_MESSAGE: &str = "Resource requested is unreadable, please reduce your request rate";
|
||||
|
||||
/// S3 error code for a request that names a KMS key the KMS does not hold.
|
||||
pub const KMS_KEY_NOT_FOUND_ERROR_CODE: &str = "KMS.NotFoundException";
|
||||
@@ -105,6 +107,7 @@ fn custom_error_status(code: &S3ErrorCode) -> Option<StatusCode> {
|
||||
S3ErrorCode::Custom(custom) if &**custom == KMS_KEY_NOT_FOUND_ERROR_CODE || &**custom == MAX_VERSIONS_EXCEEDED_CODE => {
|
||||
Some(StatusCode::BAD_REQUEST)
|
||||
}
|
||||
S3ErrorCode::Custom(custom) if &**custom == SLOW_DOWN_READ_CODE => Some(StatusCode::SERVICE_UNAVAILABLE),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
@@ -403,6 +406,7 @@ impl ApiError {
|
||||
S3ErrorCode::Custom(code) if &**code == MAX_VERSIONS_EXCEEDED_CODE => {
|
||||
MAX_VERSIONS_EXCEEDED_MESSAGE.to_string()
|
||||
}
|
||||
S3ErrorCode::Custom(code) if &**code == SLOW_DOWN_READ_CODE => SLOW_DOWN_READ_MESSAGE.to_string(),
|
||||
_ => code.as_str().to_string(),
|
||||
}
|
||||
}
|
||||
@@ -577,10 +581,10 @@ impl From<StorageError> for ApiError {
|
||||
| StorageError::FaultyRemoteDisk
|
||||
| StorageError::DiskNotFound
|
||||
| StorageError::TooManyOpenFiles => S3ErrorCode::ServiceUnavailable,
|
||||
StorageError::ErasureReadQuorum
|
||||
| StorageError::InsufficientReadQuorum(_, _)
|
||||
| StorageError::ErasureWriteQuorum
|
||||
| StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
|
||||
StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => {
|
||||
S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())
|
||||
}
|
||||
StorageError::ErasureWriteQuorum | StorageError::InsufficientWriteQuorum(_, _) => S3ErrorCode::ServiceUnavailable,
|
||||
StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable,
|
||||
StorageError::QuotaExceeded { .. } => S3ErrorCode::InvalidRequest,
|
||||
StorageError::MaxVersionsExceeded => S3ErrorCode::Custom(MAX_VERSIONS_EXCEEDED_CODE.into()),
|
||||
@@ -1288,10 +1292,10 @@ mod tests {
|
||||
(StorageError::FaultyRemoteDisk, S3ErrorCode::ServiceUnavailable),
|
||||
(StorageError::DiskNotFound, S3ErrorCode::ServiceUnavailable),
|
||||
(StorageError::TooManyOpenFiles, S3ErrorCode::ServiceUnavailable),
|
||||
(StorageError::ErasureReadQuorum, S3ErrorCode::ServiceUnavailable),
|
||||
(StorageError::ErasureReadQuorum, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into())),
|
||||
(
|
||||
StorageError::InsufficientReadQuorum("test".into(), "test".into()),
|
||||
S3ErrorCode::ServiceUnavailable,
|
||||
S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()),
|
||||
),
|
||||
(StorageError::ErasureWriteQuorum, S3ErrorCode::ServiceUnavailable),
|
||||
(
|
||||
@@ -1429,6 +1433,21 @@ mod tests {
|
||||
assert_eq!(s3_error.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_quorum_failure_matches_minio_slow_down_read_response() {
|
||||
for error in [
|
||||
StorageError::ErasureReadQuorum,
|
||||
StorageError::InsufficientReadQuorum("bucket".into(), "object".into()),
|
||||
] {
|
||||
let api_error = ApiError::from(error);
|
||||
assert_eq!(api_error.code, S3ErrorCode::Custom(SLOW_DOWN_READ_CODE.into()));
|
||||
assert_eq!(api_error.message, SLOW_DOWN_READ_MESSAGE);
|
||||
|
||||
let s3_error: S3Error = api_error.into();
|
||||
assert_eq!(s3_error.status_code(), Some(StatusCode::SERVICE_UNAVAILABLE));
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn quota_exceeded_preserves_existing_s3_error_contract() {
|
||||
let api_error: ApiError = StorageError::QuotaExceeded { current: 5, limit: 10 }.into();
|
||||
|
||||
@@ -14,7 +14,6 @@
|
||||
|
||||
use crate::server::RPC_PREFIX;
|
||||
use crate::storage::request_context::spawn_traced;
|
||||
use crate::storage::storage_api::DiskError;
|
||||
use crate::storage::storage_api::rpc_consumer::http_service::{
|
||||
DEFAULT_READ_BUFFER_SIZE, DeleteOptions, DiskStore, NS_SCANNER_PROTOCOL_VERSION, NsScannerCapabilityResponse,
|
||||
PUT_FILE_AUTH_TRAILER_LEN, PUT_FILE_AUTH_V1, PUT_FILE_CAPABILITY_VERSION, PutFileCapabilityResponse, StorageDiskRpcExt as _,
|
||||
@@ -31,6 +30,7 @@ use crate::storage::storage_api::rpc_consumer::http_service::{
|
||||
};
|
||||
use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
|
||||
use crate::storage::storage_api::tonic_rpc_auth_failure_reason;
|
||||
use crate::storage::storage_api::{DiskError, FileReader};
|
||||
use bytes::{Bytes, BytesMut};
|
||||
use futures_util::{Stream, StreamExt, TryStreamExt, stream};
|
||||
use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Uri};
|
||||
@@ -67,6 +67,8 @@ const LOG_SUBSYSTEM_NAMESPACE_SCANNER: &str = "namespace_scanner";
|
||||
const LOG_SUBSYSTEM_ROUTING: &str = "routing";
|
||||
const EVENT_RPC_REQUEST_REJECTED: &str = "rpc_request_rejected";
|
||||
const EVENT_RPC_REQUEST_FAILED: &str = "rpc_request_failed";
|
||||
const RUSTFS_META_BUCKET: &str = ".rustfs.sys";
|
||||
const MIGRATING_META_BUCKET: &str = ".minio.sys";
|
||||
const EVENT_RPC_BACKGROUND_TASK_FAILED: &str = "rpc_background_task_failed";
|
||||
const RPC_OPERATION_UNKNOWN: &str = "unknown";
|
||||
const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream";
|
||||
@@ -633,34 +635,32 @@ async fn handle_read_file(req: Request<Incoming>) -> Response<Body> {
|
||||
return response_with_status(StatusCode::BAD_REQUEST, "disk not found");
|
||||
};
|
||||
|
||||
let file = match disk
|
||||
.read_file_stream(&query.volume, &query.path, query.offset, query.length)
|
||||
.await
|
||||
{
|
||||
Ok(file) => file,
|
||||
Err(e) => {
|
||||
let message = format!("read file err {e}");
|
||||
error!(
|
||||
event = EVENT_RPC_REQUEST_FAILED,
|
||||
component = LOG_COMPONENT_INTERNODE_RPC,
|
||||
subsystem = LOG_SUBSYSTEM_FILE_TRANSFER,
|
||||
operation = INTERNODE_OPERATION_READ_FILE_STREAM,
|
||||
result = "failed",
|
||||
status_code = StatusCode::INTERNAL_SERVER_ERROR.as_u16(),
|
||||
rpc_path = req.uri().path(),
|
||||
method = %req.method(),
|
||||
reason = "read_file_failed",
|
||||
disk = %query.disk,
|
||||
volume = %query.volume,
|
||||
path = %query.path,
|
||||
offset = query.offset,
|
||||
length = query.length,
|
||||
error = %e,
|
||||
"internode rpc request failed"
|
||||
);
|
||||
return response_with_disk_error(&e, message);
|
||||
}
|
||||
};
|
||||
let file =
|
||||
match read_file_stream_with_legacy_meta_fallback(&disk, &query.volume, &query.path, query.offset, query.length).await {
|
||||
Ok(file) => file,
|
||||
Err(e) => {
|
||||
let message = format!("read file err {e}");
|
||||
error!(
|
||||
event = EVENT_RPC_REQUEST_FAILED,
|
||||
component = LOG_COMPONENT_INTERNODE_RPC,
|
||||
subsystem = LOG_SUBSYSTEM_FILE_TRANSFER,
|
||||
operation = INTERNODE_OPERATION_READ_FILE_STREAM,
|
||||
result = "failed",
|
||||
status_code = StatusCode::INTERNAL_SERVER_ERROR.as_u16(),
|
||||
rpc_path = req.uri().path(),
|
||||
method = %req.method(),
|
||||
reason = "read_file_failed",
|
||||
disk = %query.disk,
|
||||
volume = %query.volume,
|
||||
path = %query.path,
|
||||
offset = query.offset,
|
||||
length = query.length,
|
||||
error = %e,
|
||||
"internode rpc request failed"
|
||||
);
|
||||
return response_with_disk_error(&e, message);
|
||||
}
|
||||
};
|
||||
|
||||
runtime_sources::current_internode_metrics().record_incoming_request_for_operation_and_backend(
|
||||
INTERNODE_OPERATION_READ_FILE_STREAM,
|
||||
@@ -800,28 +800,30 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
|
||||
let log_disk_id = args.disk_id.clone();
|
||||
let log_skip_total_timeout = args.skip_total_timeout;
|
||||
let body = walk_dir_response_body(propagate_completion_errors, move |mut writer| async move {
|
||||
disk.walk_dir(args, &mut writer).await.map_err(|e| {
|
||||
warn!(
|
||||
event = EVENT_RPC_BACKGROUND_TASK_FAILED,
|
||||
component = LOG_COMPONENT_INTERNODE_RPC,
|
||||
subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK,
|
||||
operation = INTERNODE_OPERATION_WALK_DIR,
|
||||
result = "failed",
|
||||
disk = %log_disk,
|
||||
bucket = %log_bucket,
|
||||
base_dir = %log_base_dir,
|
||||
recursive = log_recursive,
|
||||
report_notfound = log_report_notfound,
|
||||
filter_prefix = ?log_filter_prefix,
|
||||
forward_to = ?log_forward_to,
|
||||
limit = log_limit,
|
||||
disk_id = %log_disk_id,
|
||||
skip_total_timeout = log_skip_total_timeout,
|
||||
error = %e,
|
||||
"internode rpc background task failed"
|
||||
);
|
||||
io::Error::other("remote walk_dir failed")
|
||||
})
|
||||
walk_dir_with_legacy_meta_fallback(&disk, args, &mut writer)
|
||||
.await
|
||||
.map_err(|e| {
|
||||
warn!(
|
||||
event = EVENT_RPC_BACKGROUND_TASK_FAILED,
|
||||
component = LOG_COMPONENT_INTERNODE_RPC,
|
||||
subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK,
|
||||
operation = INTERNODE_OPERATION_WALK_DIR,
|
||||
result = "failed",
|
||||
disk = %log_disk,
|
||||
bucket = %log_bucket,
|
||||
base_dir = %log_base_dir,
|
||||
recursive = log_recursive,
|
||||
report_notfound = log_report_notfound,
|
||||
filter_prefix = ?log_filter_prefix,
|
||||
forward_to = ?log_forward_to,
|
||||
limit = log_limit,
|
||||
disk_id = %log_disk_id,
|
||||
skip_total_timeout = log_skip_total_timeout,
|
||||
error = %e,
|
||||
"internode rpc background task failed"
|
||||
);
|
||||
io::Error::other("remote walk_dir failed")
|
||||
})
|
||||
});
|
||||
|
||||
runtime_sources::current_internode_metrics()
|
||||
@@ -833,6 +835,54 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
|
||||
.expect("failed to build walk dir response")
|
||||
}
|
||||
|
||||
fn legacy_meta_bucket_alias(volume: &str) -> Option<String> {
|
||||
if volume == MIGRATING_META_BUCKET {
|
||||
return Some(RUSTFS_META_BUCKET.to_string());
|
||||
}
|
||||
volume
|
||||
.strip_prefix(MIGRATING_META_BUCKET)
|
||||
.filter(|rest| rest.starts_with('/'))
|
||||
.map(|rest| format!("{RUSTFS_META_BUCKET}{rest}"))
|
||||
}
|
||||
|
||||
fn legacy_meta_alias_can_retry(error: &DiskError, volume: &str) -> bool {
|
||||
matches!(error, DiskError::FileNotFound | DiskError::VolumeNotFound) && legacy_meta_bucket_alias(volume).is_some()
|
||||
}
|
||||
|
||||
async fn read_file_stream_with_legacy_meta_fallback(
|
||||
disk: &DiskStore,
|
||||
volume: &str,
|
||||
path: &str,
|
||||
offset: usize,
|
||||
length: usize,
|
||||
) -> Result<FileReader, DiskError> {
|
||||
match disk.read_file_stream(volume, path, offset, length).await {
|
||||
Ok(file) => Ok(file),
|
||||
Err(error) if legacy_meta_alias_can_retry(&error, volume) => {
|
||||
let alias = legacy_meta_bucket_alias(volume).expect("legacy meta alias checked before retry");
|
||||
disk.read_file_stream(&alias, path, offset, length).await
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
|
||||
async fn walk_dir_with_legacy_meta_fallback<W: tokio::io::AsyncWrite + Unpin + Send>(
|
||||
disk: &DiskStore,
|
||||
args: WalkDirOptions,
|
||||
writer: &mut W,
|
||||
) -> Result<(), DiskError> {
|
||||
let original_bucket = args.bucket.clone();
|
||||
match disk.walk_dir(args.clone(), writer).await {
|
||||
Ok(()) => Ok(()),
|
||||
Err(error) if legacy_meta_alias_can_retry(&error, &original_bucket) => {
|
||||
let mut retry_args = args;
|
||||
retry_args.bucket = legacy_meta_bucket_alias(&original_bucket).expect("legacy meta alias checked before retry");
|
||||
disk.walk_dir(retry_args, writer).await
|
||||
}
|
||||
Err(error) => Err(error),
|
||||
}
|
||||
}
|
||||
|
||||
async fn handle_ns_scanner(req: Request<Incoming>) -> Response<Body> {
|
||||
let query = match parse_query::<NsScannerQuery>(&req) {
|
||||
Ok(query) => query,
|
||||
@@ -1699,13 +1749,14 @@ mod tests {
|
||||
NS_SCANNER_SESSION_SEQUENCE_QUERY, NS_SCANNER_TIER_REGISTRY_GENERATION_QUERY, NsScannerCapabilityResponse,
|
||||
NsScannerQuery, PUT_FILE_AUTH_STREAM_PATH, PUT_FILE_CAPABILITY_PATH, PUT_FILE_STREAM_PATH, PutFileQuery,
|
||||
READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, append_walk_dir_completion,
|
||||
internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, ns_scanner_response_body,
|
||||
ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce, put_file_capability_response,
|
||||
put_file_server_epoch_accepted, put_file_server_epoch_matches, put_file_stage_error_message, put_file_target_lock,
|
||||
read_file_body_stream, read_file_stream_buffer_size, remote_scanner_claim_rejection, response_with_disk_error,
|
||||
supports_walk_dir_stream_completion, validate_walk_dir_completion_request, verify_internode_rpc_signature,
|
||||
verify_ns_scanner_body_digest, verify_walk_dir_body_digest, walk_dir_response_body, write_authenticated_put_file,
|
||||
write_body_chunks_to_writer, write_put_file_body_chunks_to_writer,
|
||||
internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, legacy_meta_bucket_alias,
|
||||
ns_scanner_response_body, ns_scanner_server_epoch_matches, put_body_size_mismatch, put_file_auth_nonce,
|
||||
put_file_capability_response, put_file_server_epoch_accepted, put_file_server_epoch_matches,
|
||||
put_file_stage_error_message, put_file_target_lock, read_file_body_stream, read_file_stream_buffer_size,
|
||||
remote_scanner_claim_rejection, response_with_disk_error, supports_walk_dir_stream_completion,
|
||||
validate_walk_dir_completion_request, verify_internode_rpc_signature, verify_ns_scanner_body_digest,
|
||||
verify_walk_dir_body_digest, walk_dir_response_body, write_authenticated_put_file, write_body_chunks_to_writer,
|
||||
write_put_file_body_chunks_to_writer,
|
||||
};
|
||||
use crate::storage::storage_api::ecstore_rpc::{build_put_file_auth_trailer, gen_signature_headers};
|
||||
use crate::storage::storage_api::rpc_consumer::http_service::{DiskAPI as _, DiskOption, DiskStore, Endpoint, new_disk};
|
||||
@@ -3013,4 +3064,15 @@ mod tests {
|
||||
let response = response_with_disk_error(&DiskError::DiskAccessDenied, "permission denied");
|
||||
assert!(response.headers().get(rustfs_rio::INTERNODE_DISK_ERROR_HEADER).is_none());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn legacy_meta_bucket_alias_maps_only_legacy_system_metadata() {
|
||||
assert_eq!(legacy_meta_bucket_alias(".minio.sys").as_deref(), Some(".rustfs.sys"));
|
||||
assert_eq!(
|
||||
legacy_meta_bucket_alias(".minio.sys/config/iam").as_deref(),
|
||||
Some(".rustfs.sys/config/iam")
|
||||
);
|
||||
assert_eq!(legacy_meta_bucket_alias(".minio.sys-lookalike"), None);
|
||||
assert_eq!(legacy_meta_bucket_alias("user-bucket"), None);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -46,6 +46,10 @@ REQUIRED_GATES = {
|
||||
}
|
||||
|
||||
|
||||
def is_sha(value: Any) -> bool:
|
||||
return isinstance(value, str) and len(value) == 40 and all(char in "0123456789abcdef" for char in value)
|
||||
|
||||
|
||||
def command(*parts: str) -> list[str]:
|
||||
return list(parts)
|
||||
|
||||
@@ -74,6 +78,14 @@ def load_registry() -> dict[str, Any]:
|
||||
return registry
|
||||
|
||||
|
||||
def read_json_object(path: Path) -> dict[str, Any]:
|
||||
with path.open() as stream:
|
||||
payload = json.load(stream)
|
||||
if not isinstance(payload, dict):
|
||||
raise ValueError(f"expected JSON object: {path}")
|
||||
return payload
|
||||
|
||||
|
||||
def validate_registry(registry: dict[str, Any]) -> None:
|
||||
gates = {item["gate"] for item in registry.get("release_requirements", [])}
|
||||
missing = sorted(REQUIRED_GATES - gates)
|
||||
@@ -604,6 +616,89 @@ def build_status(plan: dict[str, Any], run_root: Path) -> dict[str, Any]:
|
||||
}
|
||||
|
||||
|
||||
def descriptor_gate_names(payload: dict[str, Any]) -> list[str]:
|
||||
gates = payload.get("gates")
|
||||
if not isinstance(gates, dict):
|
||||
return []
|
||||
return sorted(gate for gate in gates if isinstance(gate, str))
|
||||
|
||||
|
||||
def build_descriptor_ledger(descriptor_paths: list[Path], revision: str) -> dict[str, Any]:
|
||||
entries = []
|
||||
same_head_gates: set[str] = set()
|
||||
old_head_gates: set[str] = set()
|
||||
duplicate_gates: dict[str, list[str]] = {}
|
||||
gate_sources: dict[str, list[str]] = {}
|
||||
for raw_path in descriptor_paths:
|
||||
path = raw_path.resolve()
|
||||
entry: dict[str, Any] = {
|
||||
"path": str(raw_path),
|
||||
"file_name": path.name,
|
||||
}
|
||||
try:
|
||||
if not path.is_file():
|
||||
raise ValueError("descriptor is missing")
|
||||
if path.stat().st_size <= 0:
|
||||
raise ValueError("descriptor is empty")
|
||||
payload = read_json_object(path)
|
||||
evidence = payload.get("evidence")
|
||||
descriptor_revision = payload.get("source_revision")
|
||||
gates = descriptor_gate_names(payload)
|
||||
if evidence != "measured":
|
||||
classification = "case-level only"
|
||||
elif not is_sha(descriptor_revision):
|
||||
classification = "invalid"
|
||||
elif not gates:
|
||||
classification = "case-level only"
|
||||
elif descriptor_revision == revision:
|
||||
classification = "same-head verified"
|
||||
same_head_gates.update(gates)
|
||||
else:
|
||||
classification = "old-head measured, drift-readable"
|
||||
old_head_gates.update(gates)
|
||||
for gate in gates:
|
||||
gate_sources.setdefault(gate, []).append(path.name)
|
||||
entry.update({
|
||||
"status": "present",
|
||||
"classification": classification,
|
||||
"evidence": evidence,
|
||||
"source_revision": descriptor_revision,
|
||||
"gates": gates,
|
||||
})
|
||||
except (ValueError, OSError, json.JSONDecodeError) as error:
|
||||
entry.update({
|
||||
"status": "invalid",
|
||||
"classification": "invalid",
|
||||
"error": str(error),
|
||||
"gates": [],
|
||||
})
|
||||
entries.append(entry)
|
||||
for gate, sources in sorted(gate_sources.items()):
|
||||
if len(sources) > 1:
|
||||
duplicate_gates[gate] = sorted(sources)
|
||||
measured_gates = same_head_gates | old_head_gates
|
||||
return {
|
||||
"schema": 1,
|
||||
"kind": "scanner-heal-descriptor-ledger",
|
||||
"source_revision": revision,
|
||||
"release_approved": False,
|
||||
"entries": entries,
|
||||
"same_head_verified_gates": sorted(same_head_gates),
|
||||
"old_head_measured_gates": sorted(old_head_gates - same_head_gates),
|
||||
"missing_measured_gates": sorted(REQUIRED_GATES - measured_gates),
|
||||
"missing_current_head_gates": sorted(REQUIRED_GATES - same_head_gates),
|
||||
"duplicate_gates": duplicate_gates,
|
||||
"totals": {
|
||||
"descriptors": len(entries),
|
||||
"same_head_verified_gates": len(same_head_gates),
|
||||
"old_head_measured_gates": len(old_head_gates - same_head_gates),
|
||||
"missing_measured_gates": len(REQUIRED_GATES - measured_gates),
|
||||
"missing_current_head_gates": len(REQUIRED_GATES - same_head_gates),
|
||||
"invalid_descriptors": sum(1 for entry in entries if entry["status"] == "invalid"),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def run_preflight(plan: dict[str, Any]) -> int:
|
||||
commands = iter_preflight_commands(plan)
|
||||
if not commands:
|
||||
@@ -672,6 +767,7 @@ def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
||||
parser.add_argument("--format", choices=("text", "json"), default="text")
|
||||
parser.add_argument("--run-preflight", action="store_true")
|
||||
parser.add_argument("--status-root", type=Path)
|
||||
parser.add_argument("--descriptor-ledger", type=Path, nargs="+")
|
||||
parser.add_argument("--self-test", action="store_true")
|
||||
return parser.parse_args(argv)
|
||||
|
||||
@@ -687,9 +783,15 @@ def main(argv: list[str] | None = None) -> int:
|
||||
phases = set(args.phase or ["all"])
|
||||
if "all" in phases and len(phases) > 1:
|
||||
raise ValueError("--phase all cannot be combined with another phase")
|
||||
if args.status_root is not None and (args.write_plan or args.run_preflight):
|
||||
raise ValueError("--status-root cannot be combined with --write-plan or --run-preflight")
|
||||
plan = build_plan(registry, source_revision(args.source_revision), phases)
|
||||
if args.status_root is not None and (args.write_plan or args.run_preflight or args.descriptor_ledger):
|
||||
raise ValueError("--status-root cannot be combined with --write-plan, --run-preflight, or --descriptor-ledger")
|
||||
if args.descriptor_ledger and (args.write_plan or args.run_preflight):
|
||||
raise ValueError("--descriptor-ledger cannot be combined with --write-plan or --run-preflight")
|
||||
revision = source_revision(args.source_revision)
|
||||
if args.descriptor_ledger:
|
||||
print(json.dumps(build_descriptor_ledger(args.descriptor_ledger, revision), indent=2, sort_keys=True))
|
||||
return 0
|
||||
plan = build_plan(registry, revision, phases)
|
||||
if args.status_root is not None:
|
||||
status = build_status(plan, args.status_root)
|
||||
print(json.dumps(status, indent=2, sort_keys=True))
|
||||
|
||||
@@ -108,6 +108,60 @@ assert status["release_approved"] is False
|
||||
assert status["artifact_totals"]["missing"] == 0
|
||||
PY
|
||||
|
||||
"${RUSTFS_PYTHON_BIN:-python3}" - "$TMP_DIR" <<'PY'
|
||||
import json
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
root = pathlib.Path(sys.argv[1])
|
||||
same_head = root / "same-head.json"
|
||||
old_head = root / "old-head.json"
|
||||
case_level = root / "case-level.json"
|
||||
same_head.write_text(json.dumps({
|
||||
"schema": 1,
|
||||
"evidence": "measured",
|
||||
"source_revision": "a" * 40,
|
||||
"gates": {"G01": {}, "G02": {}},
|
||||
}) + "\n")
|
||||
old_head.write_text(json.dumps({
|
||||
"schema": 1,
|
||||
"evidence": "measured",
|
||||
"source_revision": "b" * 40,
|
||||
"gates": {"G03": {}, "G09": {}},
|
||||
}) + "\n")
|
||||
case_level.write_text(json.dumps({
|
||||
"schema": 1,
|
||||
"evidence": "case",
|
||||
"source_revision": "a" * 40,
|
||||
"gates": {"G14": {}},
|
||||
}) + "\n")
|
||||
PY
|
||||
|
||||
"${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" \
|
||||
--source-revision aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa \
|
||||
--descriptor-ledger "$TMP_DIR/same-head.json" "$TMP_DIR/old-head.json" "$TMP_DIR/case-level.json" \
|
||||
"$TMP_DIR/missing.json" >"$TMP_DIR/ledger.json"
|
||||
|
||||
"${RUSTFS_PYTHON_BIN:-python3}" - "$TMP_DIR/ledger.json" <<'PY'
|
||||
import json
|
||||
import pathlib
|
||||
import sys
|
||||
|
||||
ledger = json.loads(pathlib.Path(sys.argv[1]).read_text())
|
||||
assert ledger["kind"] == "scanner-heal-descriptor-ledger"
|
||||
assert ledger["release_approved"] is False
|
||||
assert ledger["same_head_verified_gates"] == ["G01", "G02"]
|
||||
assert ledger["old_head_measured_gates"] == ["G03", "G09"]
|
||||
assert "G14" in ledger["missing_measured_gates"]
|
||||
assert "G03" in ledger["missing_current_head_gates"]
|
||||
assert ledger["totals"]["invalid_descriptors"] == 1
|
||||
classifications = {entry["file_name"]: entry["classification"] for entry in ledger["entries"]}
|
||||
assert classifications["same-head.json"] == "same-head verified"
|
||||
assert classifications["old-head.json"] == "old-head measured, drift-readable"
|
||||
assert classifications["case-level.json"] == "case-level only"
|
||||
assert classifications["missing.json"] == "invalid"
|
||||
PY
|
||||
|
||||
if "${RUSTFS_PYTHON_BIN:-python3}" "$RUNNER" \
|
||||
--phase performance \
|
||||
--run-preflight >/dev/null 2>"$TMP_DIR/no-preflight.err"; then
|
||||
|
||||
@@ -140,9 +140,9 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
|
||||
self.assertNotIn(" continue-on-error: true", self.steps[name])
|
||||
self.assertIn(" if: ${{ always() && steps.evidence.outcome == 'success' }}", self.steps["Generate report"])
|
||||
self.assertNotIn("/tmp/rustfs-security", self.source)
|
||||
for name in ("Upload functional report to dashboard", "File failure issue in rustfs/backlog"):
|
||||
report = next(line for line in self.steps[name] if line.strip().startswith("REPORT_FILE:"))
|
||||
self.assertIn("${{ env.SECURITY_ARTIFACTS_DIR }}/report.md", report)
|
||||
for name in ("Upload functional report to dashboard", "Manage backlog issues (dedup / label / auto-close)"):
|
||||
expected = "${{ env.SECURITY_ARTIFACTS_DIR }}/report.md" if name.startswith("Upload") else '--report-file "${SECURITY_ARTIFACTS_DIR}/report.md"'
|
||||
self.assertIn(expected, "\n".join(self.steps[name]))
|
||||
for name in ("Upload functional report to dashboard", "Upload report and logs"):
|
||||
self.assertIn(" if: ${{ always() && steps.evidence.outcome == 'success' }}", self.steps[name])
|
||||
artifact_settings = yaml_block(self.steps["Upload report and logs"], "with", 8)
|
||||
@@ -268,10 +268,12 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase):
|
||||
gh.chmod(0o755)
|
||||
body = self.directory / "issue-body.md"
|
||||
self.env.update(PATH=f"{fake_bin}{os.pathsep}{os.environ['PATH']}", CAPTURE_BODY=str(body))
|
||||
result = self.run_step("File failure issue in rustfs/backlog")
|
||||
result = self.run_step("Manage backlog issues (dedup / label / auto-close)")
|
||||
self.assertEqual(result.returncode, 0, result.stderr)
|
||||
self.assertNotIn("OLD RUN REPORT", body.read_text())
|
||||
self.assertIn("https://github.com/rustfs/rustfs/actions/runs/314159", body.read_text())
|
||||
# The manager lives in the private auto-testing checkout; without it
|
||||
# the step must skip without publishing anything.
|
||||
self.assertIn("issue_manager.py not found", result.stdout + result.stderr)
|
||||
self.assertFalse(body.exists())
|
||||
|
||||
def test_all_ten_suites_hold_the_shared_lock_for_manual_and_chain_runs(self) -> None:
|
||||
for suite in ("upgrade", "s3-compat", "kms", "tier", "storage", "heal", "pool-expand", "security", "replication", "performance"):
|
||||
@@ -351,7 +353,7 @@ fi
|
||||
job = yaml_block(replication.splitlines(), "replication-test", 2)
|
||||
self.assertFalse(any(line.startswith(" continue-on-error:") for line in job))
|
||||
self.steps = named_steps(job)
|
||||
handoff = "Continue functional chain (next: Performance)"
|
||||
handoff = "Continue functional chain (next: Fault tolerance)"
|
||||
self.assertIn(" if: ${{ always() && github.event_name == 'repository_dispatch' }}", self.steps[handoff])
|
||||
self.assertFalse(any(line.strip().startswith("continue-on-error:") for line in self.steps[handoff]))
|
||||
self.assertIn(" if: always()", self.steps["Cleanup environment (after)"])
|
||||
@@ -371,11 +373,11 @@ fi
|
||||
self.assertEqual(forwarded.returncode == 0, bool(token) and failed_attempts < 3, forwarded.stderr)
|
||||
calls = dispatches.read_text().splitlines() if dispatches.exists() else []
|
||||
self.assertEqual(calls, [
|
||||
"api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-performance -F client_payload[from_suite]=replication",
|
||||
"api --method POST repos/rustfs/rustfs/dispatches -f event_type=rustfs-chain-fault-tolerance -F client_payload[from_suite]=replication",
|
||||
] * (min(failed_attempts + 1, 3) if token else 0))
|
||||
if failed_attempts == 3:
|
||||
self.assertIn("could not hand off from **replication** to **Performance**", body.read_text())
|
||||
self.assertIn("rustfs-chain-performance", body.read_text())
|
||||
self.assertIn("could not hand off from **replication** to **Fault tolerance**", body.read_text())
|
||||
self.assertIn("rustfs-chain-fault-tolerance", body.read_text())
|
||||
self.assertEqual(executed.read_text().splitlines().count("issue"), 2 if issue_exit else 1)
|
||||
self.assertFalse(Path(body_path.read_text().strip()).exists())
|
||||
|
||||
@@ -587,11 +589,10 @@ class FunctionalEvidenceTests(WorkflowSteps, unittest.TestCase):
|
||||
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)
|
||||
manager = self.run_step("Manage backlog issues (dedup / label / auto-close)")
|
||||
self.assertEqual(manager.returncode, 0, manager.stderr)
|
||||
self.assertIn("issue_manager.py not found", manager.stdout + manager.stderr)
|
||||
self.assertFalse(Path(self.env["CAPTURE_BODY"]).exists())
|
||||
self.assertEqual((existing / "report.md").read_text(), "OLD RUN EVIDENCE")
|
||||
|
||||
def test_reports_use_only_current_complete_suite_evidence(self):
|
||||
|
||||
Reference in New Issue
Block a user