feat(ci): add fault-tolerance degradation suite to the functional chain

Adds a fault-tolerance suite that verifies read/write behavior under drive and node loss against the erasure-coding contract and snapshots health-endpoint responses at every degradation tier. Scenarios are derived from product source (default_parity_count, erasure set sizing): - A: single-node 4 drives (EC:2, read quorum 2): hide 1/2/3 drives - B: multi-node 4x1 (one set of 4, EC:2): stop 1/2/3 nodes - C: multi-node 4x4 (one set of 16, EC:4, read quorum 12): 1 node down lands exactly on the read-quorum boundary; 2 nodes down breaks it - C2: multi-node 4x4 with RUSTFS_STORAGE_CLASS_STANDARD=EC:8 (read quorum 8, write quorum 9, lock majority 9): 2 nodes down puts reads inside the reported divergence window (read quorum met while the lock majority is broken) By default a reads-refused-despite-met-read-quorum observation is reported as known-divergence without failing the suite; the strict input escalates it. Chain order becomes: upgrade -> s3 -> kms -> tier -> storage -> heal -> pool -> security -> replication -> fault-tolerance -> performance. Covers the findings of the 2026-09 external degradation report (reads at read quorum, health-endpoint readiness truthfulness, degradedReasons capture) as automated regression probes.
fix(heal): select writable recovery intent owner (#7668 )
2026-09-11 21:39:27 +00:00 · 2026-09-11 22:45:56 +08:00 · 2026-09-11 22:38:23 +08:00 · 2026-09-11 22:37:48 +08:00 · 2026-09-11 22:36:15 +08:00
8 changed files with 607 additions and 51 deletions
@@ -0,0 +1,276 @@
+# 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: File failure issue in rustfs/backlog
+        if: ${{ failure() && steps.evidence.outcome == 'success' }}
+        env:
+          GH_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }}
+          SUITE: fault-tolerance
+          SUITE_LABEL: Fault-Tolerance
+        run: |
+          set -euo pipefail
+          TITLE="[functional][${SUITE}] ${SUITE_LABEL} suite failed (run ${GITHUB_RUN_ID})"
+          RUN_URL="${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}"
+          EXISTING="$(gh issue list -R rustfs/backlog \
+            --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"
+            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 [ -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}"
+          fi
+          echo "filed backlog issue for suite ${SUITE}"
+
+      - 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);
@@ -329,7 +329,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 +342,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 +352,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}" \
@@ -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)
@@ -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.clone().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
@@ -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;
Author	SHA1	Message	Date
xiaomage	e55c9ffece	feat(ci): add fault-tolerance degradation suite to the functional chain Adds a fault-tolerance suite that verifies read/write behavior under drive and node loss against the erasure-coding contract and snapshots health-endpoint responses at every degradation tier. Scenarios are derived from product source (default_parity_count, erasure set sizing): - A: single-node 4 drives (EC:2, read quorum 2): hide 1/2/3 drives - B: multi-node 4x1 (one set of 4, EC:2): stop 1/2/3 nodes - C: multi-node 4x4 (one set of 16, EC:4, read quorum 12): 1 node down lands exactly on the read-quorum boundary; 2 nodes down breaks it - C2: multi-node 4x4 with RUSTFS_STORAGE_CLASS_STANDARD=EC:8 (read quorum 8, write quorum 9, lock majority 9): 2 nodes down puts reads inside the reported divergence window (read quorum met while the lock majority is broken) By default a reads-refused-despite-met-read-quorum observation is reported as known-divergence without failing the suite; the strict input escalates it. Chain order becomes: upgrade -> s3 -> kms -> tier -> storage -> heal -> pool -> security -> replication -> fault-tolerance -> performance. Covers the findings of the 2026-09 external degradation report (reads at read quorum, health-endpoint readiness truthfulness, degradedReasons capture) as automated regression probes.	2026-09-11 22:45:56 +08:00
cxymds	c478a392e7	fix(heal): select writable recovery intent owner (#7668 )	2026-09-11 22:38:23 +08:00
cxymds	c8ccc1e198	fix(heal): recover stale peer state after restart (#7665 )	2026-09-11 22:37:48 +08:00
cxymds	74ba5c205c	fix(ecstore): preserve multi-pool version histories (#7658 ) * fix(ecstore): preserve multi-pool version histories * fix(ecstore): satisfy clippy in rebalance test * fix(ecstore): preserve decommission delete compatibility	2026-09-11 22:36:15 +08:00