From 17807b05fb174ee23e6c4ef33d997ff3604ea45f Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 9 Sep 2026 14:24:23 +0800 Subject: [PATCH 1/7] fix(ecstore): preserve internode error context when cloning (#7548) --- .../src/heal_erasure_disk_rebuild_test.rs | 2 + crates/ecstore/src/disk/error.rs | 108 +++++++++++++++++- crates/ecstore/src/disk/error_reduce.rs | 72 ++++++++++++ 3 files changed, 181 insertions(+), 1 deletion(-) diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index 1114efda2..a2bb5c425 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -1047,6 +1047,8 @@ mod tests { let mut cluster = RustFSTestClusterEnvironment::new(4).await?; cluster.set_env("RUSTFS_UNSAFE_BYPASS_DISK_CHECK", "true"); cluster.set_env("RUSTFS_HEAL_ENABLED", "true"); + // Capture physical baselines after the PUT rename fanout has drained. + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); // Heal control uses the first lexicographically sorted grid host. // Keep that coordinator distinct from the remote target at index 1. cluster.nodes.sort_by(|left, right| left.url.cmp(&right.url)); diff --git a/crates/ecstore/src/disk/error.rs b/crates/ecstore/src/disk/error.rs index c8db76400..8e551b4af 100644 --- a/crates/ecstore/src/disk/error.rs +++ b/crates/ecstore/src/disk/error.rs @@ -627,7 +627,13 @@ impl From for DiskError { impl Clone for DiskError { fn clone(&self) -> Self { match self { - DiskError::Io(io_error) => DiskError::Io(std::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) + // The helper derives a kind from the source; Clone must retain the original outer kind. + .map(|source| std::io::Error::new(io_error.kind(), source)) + .unwrap_or_else(|| std::io::Error::new(io_error.kind(), io_error.to_string())), + ), DiskError::MaxVersionsExceeded => DiskError::MaxVersionsExceeded, DiskError::Unexpected => DiskError::Unexpected, DiskError::CorruptedFormat => DiskError::CorruptedFormat, @@ -1265,6 +1271,49 @@ mod tests { assert!(!bad_request.is_retryable_internode_write_failure()); } + #[test] + fn test_internode_http_clone_preserves_retryability_status_and_context() { + use http::StatusCode; + use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, ConnectionReset, HttpStatus, Unknown}; + + for (kind, retryable) in [ + (ConnectionRefused, true), + (ConnectionReset, true), + (HttpStatus(StatusCode::TOO_MANY_REQUESTS), true), + (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true), + (HttpStatus(StatusCode::CONFLICT), true), + (Unknown, false), + (HttpStatus(StatusCode::BAD_REQUEST), false), + (HttpStatus(StatusCode::INTERNAL_SERVER_ERROR), false), + ] { + let original = DiskError::from(rustfs_rio::new_test_internode_http_io_error(kind)); + assert_eq!(original.internode_http_error_kind(), Some(kind)); + assert_eq!(original.is_retryable_internode_write_failure(), retryable); + + let cloned = original.clone(); + assert_eq!(cloned, original, "clone must preserve the error bucket for {kind:?}"); + assert_eq!( + cloned.is_retryable_internode_write_failure(), + retryable, + "clone changed retryability for {kind:?}" + ); + assert_eq!(cloned.internode_http_error_kind(), Some(kind)); + if let HttpStatus(status) = kind { + assert!(cloned.is_internode_http_status(status.as_u16())); + } + let DiskError::Io(io_error) = &cloned else { + panic!("unmarked internode error must remain Io: {cloned:?}"); + }; + let source = io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("clone must retain the structured internode error"); + assert_eq!(source.context().method(), "PUT"); + assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream"); + assert_eq!(source.context().operation(), Some(INTERNODE_OPERATION_PUT_FILE_STREAM)); + } + } + #[tokio::test] async fn read_stream_conflict_is_not_a_retryable_put_file_failure() { use tokio::io::{AsyncReadExt, AsyncWriteExt}; @@ -1309,11 +1358,57 @@ mod tests { !error.is_retryable_internode_write_failure(), "read-operation 409 must not trigger put-file retry" ); + let cloned = error.clone(); + let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(error)], &[], 1) + .expect("the read conflict must remain the dominant error"); + for preserved in [&cloned, &reduced] { + assert!( + !preserved.is_retryable_internode_write_failure(), + "cloning or reducing a read conflict must not turn it into a PUT retry" + ); + assert!(preserved.is_internode_http_status(409)); + let DiskError::Io(io_error) = preserved else { + panic!("read conflict must remain Io: {preserved:?}"); + }; + let source = io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("read conflict must retain its request context"); + assert_eq!(source.context().method(), "GET"); + assert_eq!(source.context().target(), "/rustfs/rpc/read_file_stream"); + assert_eq!( + source.context().operation(), + Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_READ_FILE_STREAM) + ); + } }) .await .expect("isolated read-conflict test must finish within its budget"); } + #[test] + fn test_internode_http_clone_preserves_outer_io_kind_and_message() { + let source = rustfs_rio::new_test_internode_http_io_error(InternodeHttpErrorKind::ConnectionReset) + .into_inner() + .expect("the internode helper must provide a typed source"); + let original_io = io::Error::new(io::ErrorKind::InvalidData, source); + let message = original_io.to_string(); + let original = DiskError::from(original_io); + assert_eq!(original.internode_http_error_kind(), Some(InternodeHttpErrorKind::ConnectionReset)); + assert!(original.is_retryable_internode_write_failure()); + + let cloned = original.clone(); + let reduced = crate::disk::error_reduce::reduce_write_quorum_errs(&[Some(original)], &[], 1) + .expect("the wrapped internode error must remain the dominant error"); + for preserved in [&cloned, &reduced] { + let DiskError::Io(io_error) = preserved else { + panic!("the wrapped error must remain Io: {preserved:?}"); + }; + assert_eq!(io_error.kind(), io::ErrorKind::InvalidData); + assert_eq!(io_error.to_string(), message); + } + } + #[test] fn test_internode_missing_errors_preserve_disk_error_types() { let file_missing = DiskError::from(rustfs_rio::new_test_remote_file_not_found_http_io_error()); @@ -1325,6 +1420,17 @@ mod tests { assert_eq!(file_missing, DiskError::FileNotFound); assert_eq!(volume_missing, DiskError::VolumeNotFound); assert!(matches!(unmarked_server_error, DiskError::Io(_))); + for missing in [file_missing, volume_missing] { + assert_eq!(missing.clone(), missing); + assert_eq!( + crate::disk::error_reduce::reduce_write_quorum_errs( + &[Some(missing.clone()), Some(missing.clone()), None], + &[], + 2 + ), + Some(missing) + ); + } } #[test] diff --git a/crates/ecstore/src/disk/error_reduce.rs b/crates/ecstore/src/disk/error_reduce.rs index 05ed1b519..298905459 100644 --- a/crates/ecstore/src/disk/error_reduce.rs +++ b/crates/ecstore/src/disk/error_reduce.rs @@ -226,6 +226,78 @@ mod tests { assert_eq!(res, Some(quorum_err)); } + #[test] + fn test_write_quorum_reduction_preserves_internode_http_identity() { + use http::StatusCode; + use rustfs_rio::InternodeHttpErrorKind::{ConnectionRefused, HttpStatus, Unknown}; + + for (kind, retryable) in [ + (ConnectionRefused, true), + (HttpStatus(StatusCode::SERVICE_UNAVAILABLE), true), + (HttpStatus(StatusCode::CONFLICT), true), + (Unknown, false), + (HttpStatus(StatusCode::BAD_REQUEST), false), + ] { + // Construct both producer errors independently: the reducer owns the first clone. + let first = Error::from(rustfs_rio::new_test_internode_http_io_error(kind)); + let second = Error::from(rustfs_rio::new_test_internode_http_io_error(kind)); + assert_eq!(first.internode_http_error_kind(), Some(kind)); + assert_eq!(second.internode_http_error_kind(), Some(kind)); + assert_eq!(first.is_retryable_internode_write_failure(), retryable); + let errors = [Some(first), Some(second), None]; + let reduced = reduce_write_quorum_errs(&errors, OBJECT_OP_IGNORED_ERRS, 2) + .expect("two equal producer errors must dominate one successful write"); + + assert_eq!(Some(&reduced), errors[0].as_ref()); + assert_eq!( + reduced.is_retryable_internode_write_failure(), + retryable, + "quorum reduction changed retryability for {kind:?}" + ); + assert_eq!(reduced.internode_http_error_kind(), Some(kind)); + if let HttpStatus(status) = kind { + assert!(reduced.is_internode_http_status(status.as_u16())); + } + let Error::Io(io_error) = &reduced else { + panic!("the dominant error must remain Io: {reduced:?}"); + }; + let source = io_error + .get_ref() + .and_then(|source| source.downcast_ref::()) + .expect("quorum reduction must retain the structured internode error"); + assert_eq!(source.context().method(), "PUT"); + assert_eq!(source.context().target(), "/rustfs/rpc/put_file_stream"); + assert_eq!( + source.context().operation(), + Some(rustfs_io_metrics::internode_metrics::INTERNODE_OPERATION_PUT_FILE_STREAM) + ); + } + } + + #[test] + fn test_clone_and_write_quorum_do_not_promote_non_retryable_errors() { + use http::StatusCode; + use rustfs_rio::InternodeHttpErrorKind::{HttpStatus, Unknown}; + + for original in [ + Error::from(rustfs_rio::new_test_internode_http_io_error(Unknown)), + Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::BAD_REQUEST))), + Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::FORBIDDEN))), + Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus(StatusCode::NOT_FOUND))), + Error::from(rustfs_rio::new_test_internode_http_io_error(HttpStatus( + StatusCode::INTERNAL_SERVER_ERROR, + ))), + err_io("internode connection reset: PUT /rustfs/rpc/put_file_stream"), + ] { + assert!(!original.is_retryable_internode_write_failure()); + let cloned = original.clone(); + let reduced = + reduce_write_quorum_errs(&[Some(original)], &[], 1).expect("a non-retryable error must remain an error"); + assert!(!cloned.is_retryable_internode_write_failure()); + assert!(!reduced.is_retryable_internode_write_failure()); + } + } + #[test] fn test_count_errs() { let e1 = err_io("a"); From e549252ac6ef756b59237136a3b3d2348f89d344 Mon Sep 17 00:00:00 2001 From: hector <42570491+majinghe@users.noreply.github.com> Date: Wed, 9 Sep 2026 14:24:44 +0800 Subject: [PATCH 2/7] feat(nightly): make the build branch configurable (NIGHTLY_BRANCH var + dispatch input) (#7557) --- .github/workflows/nightly-gnu.yml | 177 +++++++++++++++++++++++++++++- scripts/test_nightly_candidate.py | 11 +- 2 files changed, 178 insertions(+), 10 deletions(-) diff --git a/.github/workflows/nightly-gnu.yml b/.github/workflows/nightly-gnu.yml index 1864a9715..6d7920f5e 100644 --- a/.github/workflows/nightly-gnu.yml +++ b/.github/workflows/nightly-gnu.yml @@ -19,17 +19,27 @@ on: - cron: "7 0 * * *" timezone: "Asia/Shanghai" workflow_dispatch: + inputs: + branch: + description: 'Branch/ref to build and publish as the nightly (empty = scheduled source, see NIGHTLY_BUILD_REF)' + required: false + default: '' permissions: contents: read +# Scheduled builds follow the NIGHTLY_BRANCH repo variable so the channel can +# be pointed at e.g. `release` for the GA cycle and back to `main` afterwards +# without touching this file. Manual runs take the `branch` input, falling +# back to the branch the run was dispatched from. concurrency: - group: nightly-gnu-build-main-${{ github.event_name }} + group: nightly-gnu-build-${{ github.event_name }}-${{ github.event_name == 'schedule' && (vars.NIGHTLY_BRANCH || 'main') || (inputs.branch || github.ref_name) }} cancel-in-progress: ${{ github.event_name == 'workflow_dispatch' }} env: CARGO_TERM_COLOR: always RUST_BACKTRACE: 1 + NIGHTLY_BUILD_REF: ${{ github.event_name == 'schedule' && (vars.NIGHTLY_BRANCH || 'main') || (inputs.branch || github.ref_name) }} jobs: build: @@ -43,6 +53,7 @@ jobs: uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 with: persist-credentials: false + ref: ${{ env.NIGHTLY_BUILD_REF }} - name: Setup Rust environment uses: ./.github/actions/setup @@ -152,13 +163,104 @@ jobs: fakeroot dpkg-deb --build "${PKG_DIR}" ls -lh "${DEB_FILE}" + echo "deb_date=${DEB_DATE}" >> "${GITHUB_OUTPUT}" echo "deb_file=${DEB_FILE}" >> "${GITHUB_OUTPUT}" + # Same packaging scheme as .github/workflows/package.yml (fpm), but from + # the locally built nightly binary instead of a release artifact, with a + # date-based version that mirrors the DEB. + - name: Build RPM package + id: rpm + shell: bash + env: + DEB_DATE: ${{ steps.deb.outputs.deb_date }} + run: | + set -euo pipefail + + if ! command -v fpm >/dev/null 2>&1; then + SUDO=""; [ "$(id -u)" -ne 0 ] && SUDO="sudo -n" + ${SUDO} apt-get update -qq && ${SUDO} apt-get install -y -qq ruby ruby-dev build-essential rpm >/dev/null + ${SUDO} gem install fpm --no-document >/dev/null + fi + + RPM_FILE="rustfs-nightly-${DEB_DATE}.rpm" + RPM_VERSION="0" + RPM_RELEASE="0.nightly.${DEB_DATE//-/.}" + + echo "Building RPM: ${RPM_FILE} (version ${RPM_VERSION}-${RPM_RELEASE})" + + # fpm wants the config file to exist before packaging. + mkdir -p ./tmp-pkg/etc/default + cat > ./tmp-pkg/etc/default/rustfs << 'ENVEOF' + # RustFS Environment Configuration + # See https://rustfs.com/docs/ for more information + # RUSTFS_VOLUMES="" + # RUSTFS_ROOT_USER="" + # RUSTFS_ROOT_PASSWORD="" + ENVEOF + + fpm -s dir -t rpm \ + --name rustfs \ + --version "$RPM_VERSION" \ + --iteration "$RPM_RELEASE" \ + --architecture x86_64 \ + --package "$RPM_FILE" \ + --depends "glibc >= 2.31" \ + --maintainer "RustFS Team " \ + --description "High-performance distributed object storage" \ + --url "https://rustfs.com" \ + --license "Apache-2.0" \ + --after-install <(cat << 'POSTINST' + #!/bin/bash + set -e + if ! getent passwd rustfs > /dev/null 2>&1; then + useradd -r -s /bin/false -d /opt/rustfs rustfs + fi + mkdir -p /opt/rustfs /data/rustfs /var/log/rustfs + chown rustfs:rustfs /opt/rustfs /data/rustfs /var/log/rustfs + if [ -d /run/systemd/system ]; then + systemctl daemon-reload + fi + POSTINST + ) \ + --before-remove <(cat << 'PRERM' + #!/bin/bash + set -e + if [ -d /run/systemd/system ] && systemctl is-active --quiet rustfs; then + systemctl stop rustfs + fi + PRERM + ) \ + --after-remove <(cat << 'POSTRM' + #!/bin/bash + set -e + if [ -d /run/systemd/system ]; then + systemctl daemon-reload + fi + POSTRM + ) \ + --config-files /etc/default/rustfs \ + "rustfs-nightly-${DEB_DATE}/usr/bin/rustfs=/usr/bin/rustfs" \ + ./tmp-pkg/etc/default/rustfs=/etc/default/rustfs \ + deploy/build/rustfs.service=/lib/systemd/system/rustfs.service \ + LICENSE=/usr/share/doc/rustfs/LICENSE \ + README.md=/usr/share/doc/rustfs/README.md + + [[ -f "$RPM_FILE" ]] || { echo "RPM build failed"; exit 1; } + rpm -qpl "$RPM_FILE" | grep -Fx '/usr/bin/rustfs' >/dev/null + stat --printf='%n %s bytes\n' "$RPM_FILE" + echo "rpm_file=$RPM_FILE" >> "$GITHUB_OUTPUT" + - name: Upload DEB artifact uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6 with: name: ${{ steps.deb.outputs.deb_file }} path: ${{ steps.deb.outputs.deb_file }} + - name: Upload RPM artifact + uses: actions/upload-artifact@b7c566a772e6b6bfb58ed0dc250532a479d7789f # v6 + with: + name: ${{ steps.rpm.outputs.rpm_file }} + path: ${{ steps.rpm.outputs.rpm_file }} if-no-files-found: error # Persist the nightly deb on Cloudflare R2 (same channel as package.yml) @@ -187,11 +289,10 @@ jobs: export AWS_SECRET_ACCESS_KEY="$R2_SECRET_ACCESS_KEY" export AWS_DEFAULT_REGION="auto" + # The candidate manifest must describe the tree that was actually + # built. With a ref override (NIGHTLY_BRANCH / dispatch input) that + # is not necessarily GITHUB_SHA, so always advertise HEAD. SOURCE_SHA="$(git rev-parse HEAD)" - if [[ "${SOURCE_SHA}" != "${GITHUB_SHA}" ]]; then - echo "Checkout SHA does not match the nightly build run" >&2 - exit 1 - fi DEB_SHA256="$(sha256sum "${DEB_FILE}" | cut -d ' ' -f 1)" CANDIDATE_KEY="artifacts/rustfs/packages/nightly/runs/${GITHUB_RUN_ID}/${GITHUB_RUN_ATTEMPT}/${DEB_SHA256}/rustfs.deb" CANDIDATE_URL="https://dl.rustfs.com/${CANDIDATE_KEY}" @@ -247,6 +348,70 @@ jobs: path: ${{ steps.publish.outputs.candidate_file }} if-no-files-found: error + # Publish the deb/rpm pair to the auto-testing repo's `assets` branch so + # engineers can download and install the nightly directly. The branch is + # a single-commit orphan rewritten on every build, which keeps the repo + # small while the latest files stay reachable at stable raw URLs. + - name: Publish packages to auto-testing assets + env: + ASSETS_TOKEN: ${{ secrets.PF_TESTING_GH_TOKEN }} + DEB_FILE: ${{ steps.deb.outputs.deb_file }} + RPM_FILE: ${{ steps.rpm.outputs.rpm_file }} + DEB_DATE: ${{ steps.deb.outputs.deb_date }} + BUILD_REF: ${{ env.NIGHTLY_BUILD_REF }} + run: | + set -euo pipefail + + if [ -z "${ASSETS_TOKEN}" ]; then + echo "PF_TESTING_GH_TOKEN is not configured; skipping assets upload" + exit 0 + fi + export GH_TOKEN="${ASSETS_TOKEN}" + + for f in "${DEB_FILE}" "${RPM_FILE}"; do + [ -f "$f" ] || { echo "missing package: $f"; exit 1; } + done + + rm -rf assets-work && mkdir assets-work + if ! gh repo clone rustfs/auto-testing assets-work -- --depth 1 --branch assets --quiet 2>/dev/null; then + echo "assets branch does not exist yet; creating an orphan" + ( cd assets-work && git init -q -b assets ) + fi + cd assets-work + git remote add origin "https://github.com/rustfs/auto-testing.git" 2>/dev/null || \ + git remote set-url origin "https://github.com/rustfs/auto-testing.git" + gh auth setup-git >/dev/null + + mkdir -p nightly + cp "../${DEB_FILE}" "nightly/${DEB_FILE}" + cp "../${RPM_FILE}" "nightly/${RPM_FILE}" + cp "nightly/${DEB_FILE}" nightly/rustfs-nightly-latest.deb + cp "nightly/${RPM_FILE}" nightly/rustfs-nightly-latest.rpm + + SOURCE_SHA="$(git -C .. rev-parse HEAD 2>/dev/null || echo "${GITHUB_SHA}")" + { + echo "# Nightly packages" + echo "" + echo "- Built: ${DEB_DATE} from \`${BUILD_REF}@${SOURCE_SHA:0:12}\`" + echo "- Run: ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" + echo "" + echo '| File | Size | SHA256 |' + echo '|---|---|---|' + for f in "nightly/${DEB_FILE}" "nightly/${RPM_FILE}" nightly/rustfs-nightly-latest.deb nightly/rustfs-nightly-latest.rpm; do + printf '| %s | %s | %s |\n' "$f" "$(du -h "$f" | cut -f1)" "$(sha256sum "$f" | cut -d' ' -f1)" + done + echo "" + echo "Download: replace /blob/ with /raw/ in any file URL, e.g." + echo "\`https://raw.githubusercontent.com/rustfs/auto-testing/assets/nightly/rustfs-nightly-latest.deb\`" + } > BUILD-INFO.md + + git add -A + git -c user.name="rustfs-nightly-bot" -c user.email="support@rustfs.com" \ + commit -q -m "nightly ${DEB_DATE} (${BUILD_REF}@${SOURCE_SHA:0:12})" \ + --allow-empty + git push --force origin assets + echo "✅ Published ${DEB_FILE} and ${RPM_FILE} to rustfs/auto-testing@assets" + # Live-Vault lane for the rustfs-kms suite (rustfs/backlog#1774). # # RUSTFS_KMS_VAULT_TOKEN is the single switch that adds the Vault KV2 and @@ -284,6 +449,7 @@ jobs: uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 with: persist-credentials: false + ref: ${{ env.NIGHTLY_BUILD_REF }} - name: Setup Rust environment uses: ./.github/actions/setup @@ -372,6 +538,7 @@ jobs: uses: actions/checkout@9c091bb21b7c1c1d1991bb908d89e4e9dddfe3e0 # v7 with: persist-credentials: false + ref: ${{ env.NIGHTLY_BUILD_REF }} - name: Setup Rust environment uses: ./.github/actions/setup diff --git a/scripts/test_nightly_candidate.py b/scripts/test_nightly_candidate.py index 3e5a50401..a1dca2405 100644 --- a/scripts/test_nightly_candidate.py +++ b/scripts/test_nightly_candidate.py @@ -163,12 +163,13 @@ SH self.assertFalse(self.store.exists()) self.assertEqual(list(self.root.glob("nightly-awscli.*")), []) - def test_checkout_sha_mismatch_fails_before_upload(self): + def test_manifest_advertises_checked_out_head_even_when_github_sha_differs(self): + # With a ref override (NIGHTLY_BRANCH variable / dispatch `branch` + # input) the checked-out HEAD intentionally differs from GITHUB_SHA; + # the candidate manifest must record the tree that was built. result = self.run_publish(GITHUB_SHA="f" * 40) - self.assertNotEqual(result.returncode, 0) - self.assertIn("Checkout SHA", result.stderr) - self.assertFalse(self.output.exists()) - self.assertFalse((self.root / "aws.log").exists()) + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(self.manifest()["source_sha"], self.sha) def test_same_date_builds_and_reruns_keep_distinct_candidates(self): urls = [] From 7db206466bad4fbb1e109e8c02a7562c8e4b5dd7 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 9 Sep 2026 14:24:57 +0800 Subject: [PATCH 3/7] test(ci): reserve runner capacity for bounded rollback probe (#7560) --- .config/nextest.toml | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/.config/nextest.toml b/.config/nextest.toml index 17c5dc7a5..2e54a7180 100644 --- a/.config/nextest.toml +++ b/.config/nextest.toml @@ -197,6 +197,12 @@ test-group = 'e2e-cluster-nightly' filter = 'package(e2e_test) & (test(/^kms::kms_vault_test::/) | test(/^kms::kms_rekey_sweep_test::/) | test(/^kms::configured_roundtrip_test::test_configured_vault_kms_admin_and_versioned_cleanup$/))' test-group = 'e2e-vault' +# This four-disk, 65-member rollback probe already drives up to 32 concurrent +# durable deletions. Reserve this nextest run's capacity for its progress oracle. +[[profile.default.overrides]] +filter = 'package(rustfs-ecstore) & test(=store::init::tests::dispatch_manifest_rollback_bounded_concurrency_reaches_tail_behind_slow_member)' +threads-required = "num-test-threads" + # --------------------------------------------------------------------------- # ci profile — the strict CI gate (ci.yml `cargo nextest run --profile ci`) # --------------------------------------------------------------------------- From d7c2fc75875d91d6877b69676771ca86d7d2aca3 Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 9 Sep 2026 14:50:28 +0800 Subject: [PATCH 4/7] fix(ecstore): apply read quorum to bucket validation (#7556) * fix(ecstore): apply read quorum to bucket validation * fix(e2e): remove needless borrow in quorum test --- .../src/namespace_lock_quorum_test.rs | 184 +++++++++++ crates/ecstore/src/set_disk/metadata.rs | 49 ++- crates/ecstore/src/set_disk/mod.rs | 1 + crates/ecstore/src/set_disk/ops/bucket.rs | 117 ++++--- crates/ecstore/src/store/bucket.rs | 301 +++++++++++++++++- 5 files changed, 578 insertions(+), 74 deletions(-) diff --git a/crates/e2e_test/src/namespace_lock_quorum_test.rs b/crates/e2e_test/src/namespace_lock_quorum_test.rs index 0af609ecc..d6e7f0f0e 100644 --- a/crates/e2e_test/src/namespace_lock_quorum_test.rs +++ b/crates/e2e_test/src/namespace_lock_quorum_test.rs @@ -25,6 +25,190 @@ const KEY: &str = "thumb/79/concurrent-overwrite.jpg"; type TestResult = Result<(), Box>; +async fn assert_quorum_object_body(client: &Client, bucket: &str, key: &str, expected: &[u8]) -> TestResult { + let body = client + .get_object() + .bucket(bucket) + .key(key) + .send() + .await? + .body + .collect() + .await? + .into_bytes(); + assert_eq!(body.as_ref(), expected, "quorum read returned incorrect contents for {key}"); + Ok(()) +} + +async fn wait_for_quorum_read_admission(clients: &[Client], bucket: &str) -> TestResult { + // SIGKILL can orphan a granted lease. Wait for shared metadata-lock + // admission before asserting the stable quorum boundary; cold bodies + // remain unread throughout this readiness probe. + let deadline = + tokio::time::Instant::now() + rustfs_lock::fast_lock::DEFAULT_LOCK_TIMEOUT + std::time::Duration::from_secs(15); + loop { + let mut ready = true; + for client in clients { + for key in ["warm-small", "warm-large"] { + match client.head_object().bucket(bucket).key(key).send().await { + Ok(_) => {} + Err(error) if error.raw_response().is_some_and(|response| response.status().as_u16() == 503) => { + ready = false; + break; + } + Err(error) => return Err(error.into()), + } + } + if !ready { + break; + } + } + if ready { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(format!("read quorum did not become available after lease convergence for {bucket}").into()); + } + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + } +} + +#[tokio::test] +async fn test_degraded_cluster_read_quorum_follows_erasure_layout() -> TestResult { + crate::common::init_logging(); + + for (node_count, parity) in [(4, 2), (6, 3), (6, 2)] { + let read_quorum = node_count - parity; + let write_quorum = read_quorum + usize::from(read_quorum == parity); + let mut cluster = RustFSTestClusterEnvironment::new(node_count).await?; + cluster.set_env("RUSTFS_STORAGE_CLASS_STANDARD", format!("EC:{parity}")); + // Wait for every seed fanout before removing any physical shard. + cluster.set_env("RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE", "false"); + cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "false"); + cluster.set_env("RUST_LOG", "warn,rustfs_lock=debug"); + cluster.start().await?; + + let clients = cluster + .create_all_clients()? + .into_iter() + .map(|client| { + Client::from_conf( + client + .config() + .to_builder() + .retry_config(aws_sdk_s3::config::retry::RetryConfig::standard().with_max_attempts(1)) + .build(), + ) + }) + .collect::>(); + let bucket = format!("read-quorum-{node_count}-{parity}"); + clients[0].create_bucket().bucket(&bucket).send().await?; + let small = b"read quorum is derived from the erasure layout".to_vec(); + let large = (0..1_048_576) + .map(|index| u8::try_from(index % 251).expect("bounded payload byte")) + .collect::>(); + for (key, body) in [ + ("warm-small", &small), + ("warm-large", &large), + ("cold-small", &small), + ("cold-large", &large), + ("below-quorum", &large), + ] { + clients[node_count - 1] + .put_object() + .bucket(&bucket) + .key(key) + .body(Bytes::copy_from_slice(body).into()) + .send() + .await?; + } + for node in &cluster.nodes { + for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] { + let census = + crate::chaos::census_object_version_on_disk(std::path::Path::new(&node.data_dir), &bucket, key, None)?; + assert!(census.is_complete(), "seed shard must be complete before fault injection: {census:?}"); + assert_eq!(census.data_blocks, Some(read_quorum)); + assert_eq!(census.parity_blocks, Some(parity)); + } + } + for client in &clients { + assert_quorum_object_body(client, &bucket, "warm-small", &small).await?; + assert_quorum_object_body(client, &bucket, "warm-large", &large).await?; + } + + for offline_node in (read_quorum..node_count).rev() { + cluster.stop_node(offline_node)?; + wait_for_quorum_read_admission(&clients[..offline_node], &bucket).await?; + for client in clients.iter().take(offline_node) { + client.head_bucket().bucket(&bucket).send().await?; + assert_quorum_object_body(client, &bucket, "warm-large", &large).await?; + } + } + + // Exercise more than the five-second positive bucket-validation TTL. + // Every sample must succeed; polling must not hide a transient failure. + let validation_deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(6); + loop { + for client in clients.iter().take(read_quorum) { + assert_quorum_object_body(client, &bucket, "warm-small", &small).await?; + assert_quorum_object_body(client, &bucket, "warm-large", &large).await?; + let listing = client.list_objects_v2().bucket(&bucket).send().await?; + for key in ["warm-small", "warm-large", "cold-small", "cold-large", "below-quorum"] { + assert!(listing.contents().iter().any(|entry| entry.key() == Some(key)), "listing omitted {key}"); + } + } + if tokio::time::Instant::now() >= validation_deadline { + break; + } + tokio::time::sleep(std::time::Duration::from_millis(250)).await; + } + for client in clients.iter().take(read_quorum) { + assert_quorum_object_body(client, &bucket, "cold-small", &small).await?; + assert_quorum_object_body(client, &bucket, "cold-large", &large).await?; + } + + let write = clients[0] + .put_object() + .bucket(&bucket) + .key("quorum-write") + .body(Bytes::copy_from_slice(&small).into()) + .send() + .await; + if read_quorum >= write_quorum { + write?; + } else { + let error = write.expect_err("a read quorum must not authorize a write that needs more votes"); + assert_eq!(error.as_service_error().and_then(|error| error.meta().code()), Some("ServiceUnavailable")); + } + + cluster.stop_node(read_quorum - 1)?; + for client in clients.iter().take(read_quorum - 1) { + match client.get_object().bucket(&bucket).key("below-quorum").send().await { + Ok(response) => assert!( + response.body.collect().await.is_err(), + "fewer than {read_quorum} valid fragments must not reconstruct an uncached object" + ), + Err(error) => assert_eq!( + error.as_service_error().and_then(|error| error.meta().code()), + Some("ServiceUnavailable"), + "a quorum loss must not be mistaken for a missing object" + ), + } + } + + for node in 0..read_quorum - 1 { + cluster.stop_node(node)?; + } + cluster.start().await?; + for client in &clients { + assert_quorum_object_body(client, &bucket, "warm-large", &large).await?; + assert_quorum_object_body(client, &bucket, "below-quorum", &large).await?; + } + } + + Ok(()) +} + async fn put_object(client: Client, payload: Vec, writer_id: usize) -> Result<(), String> { client .put_object() diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index 9f6c0d8ac..60ad041da 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -326,14 +326,15 @@ impl SetDisks { let parity_blocks = Self::common_parity(&parities, default_parity_count as i32); if parity_blocks < 0 { - // No parity value reached read quorum. Distinguish two cases: - // enough disks answered with valid-looking metadata that simply - // cannot be reconciled (corrupt/foreign entries — retrying cannot - // help, and heal should see Corrupt, rustfs#5801) versus too few - // healthy answers (a genuine quorum condition where retry may - // succeed once disks recover). + // A consistent layout can require more replies than the initial + // half-set probe. Reaching that probe alone is not corruption; + // only invalid or conflicting healthy replies establish that. let healthy_replies = errs.iter().filter(|err| err.is_none()).count(); - if healthy_replies >= expected_rquorum { + let consistent_parity = parities + .iter() + .find(|&&parity| parity >= 0) + .filter(|&&parity| parities.iter().filter(|&&candidate| candidate == parity).count() == healthy_replies); + if healthy_replies >= expected_rquorum && consistent_parity.is_none() { error!( "object_quorum_from_meta: irreconcilable parity across {healthy_replies} healthy replies (corrupt metadata), errs={errs:?}" ); @@ -1652,6 +1653,40 @@ mod tests { assert_eq!(err, DiskError::FileCorrupt); } + #[test] + fn consistent_parity_below_its_data_shard_quorum_is_not_corruption() { + for (drive_count, parity) in [(6, 2), (8, 2), (12, 4)] { + let data = drive_count - parity; + let mut metas = (1..=drive_count) + .map(|index| { + let mut info = FileInfo::new("bucket/object", data, parity); + info.size = 1024; + info.erasure.index = index; + info + }) + .collect::>(); + let mut errs = vec![Some(DiskError::DiskNotFound); drive_count]; + errs[..data].fill(None); + assert_eq!( + SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect("exact data quorum should resolve"), + (data as i32, data as i32) + ); + + errs[data - 1] = Some(DiskError::DiskNotFound); + assert_eq!( + SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("one fewer shard cannot resolve"), + DiskError::ErasureReadQuorum, + "layout {drive_count}/{parity} has consistent metadata but insufficient shards" + ); + + metas[0].erasure.parity_blocks = usize::MAX; + assert_eq!( + SetDisks::object_quorum_from_meta(&metas, &errs, parity).expect_err("corrupt healthy replies must be rejected"), + DiskError::FileCorrupt + ); + } + } + /// Too few healthy replies remains a genuine quorum condition where a /// retry may succeed once disks recover. #[test] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index d2b63795d..17129c9ae 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -865,6 +865,7 @@ pub(crate) use core::io_primitives::{ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, ren mod ctx; mod metadata; mod ops; +pub(crate) use ops::bucket::BucketInfoQuorum; #[cfg(test)] pub(crate) use ops::hermetic_set_disks_isolated; diff --git a/crates/ecstore/src/set_disk/ops/bucket.rs b/crates/ecstore/src/set_disk/ops/bucket.rs index 50e84ee68..236ed8b56 100644 --- a/crates/ecstore/src/set_disk/ops/bucket.rs +++ b/crates/ecstore/src/set_disk/ops/bucket.rs @@ -21,12 +21,72 @@ use super::super::{ BUCKET_OP_IGNORED_ERRS, BucketInfo, BucketOperations, BucketOptions, DeleteBucketOptions, DiskError, Error, HashMap, - MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_write_quorum_errs, + MakeBucketOptions, Result, SetDisks, is_reserved_or_invalid_bucket, join_all, reduce_read_quorum_errs, + reduce_write_quorum_errs, }; use crate::api::bucket::metadata_sys; use crate::disk::DiskAPI; +#[derive(Clone, Copy)] +pub(crate) enum BucketInfoQuorum { + Read, + Write, +} + impl SetDisks { + pub(crate) async fn stat_bucket_with_quorum(&self, bucket: &str, quorum: BucketInfoQuorum) -> Result { + let disks = self.disk_inventory().await; + let disk_count = disks.len(); + let mut futures = Vec::with_capacity(disk_count); + for disk in disks { + let bucket = bucket.to_string(); + futures.push(async move { + match disk { + Some(disk) => disk.stat_volume(&bucket).await, + None => Err(DiskError::DiskNotFound), + } + }); + } + + let results = join_all(futures).await; + let mut infos = Vec::with_capacity(results.len()); + let mut errs = Vec::with_capacity(results.len()); + for result in results { + match result { + Ok(info) => { + infos.push(Some(info)); + errs.push(None); + } + Err(err) => { + infos.push(None); + errs.push(Some(err)); + } + } + } + + let error = match quorum { + // Bucket mutations use a majority regardless of object storage + // class. A namespace read must intersect that majority; object + // readers still enforce the persisted layout's data-shard quorum. + BucketInfoQuorum::Read => reduce_read_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count.div_ceil(2).max(1)), + BucketInfoQuorum::Write => reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, disk_count / 2 + 1), + }; + if let Some(err) = error { + return Err(err.into()); + } + + infos + .into_iter() + .flatten() + .next() + .map(|info| BucketInfo { + name: info.name, + created: info.created, + ..Default::default() + }) + .ok_or(Error::VolumeNotFound) + } + pub(crate) async fn list_bucket_for_scanner(&self, _opts: &BucketOptions) -> Result<(Vec, bool)> { let disks = self.disk_inventory().await; let write_quorum = (disks.len() / 2) + 1; @@ -131,59 +191,12 @@ impl BucketOperations for SetDisks { #[tracing::instrument(skip(self))] async fn get_bucket_info(&self, bucket: &str, _opts: &BucketOptions) -> Result { - let disks = self.disk_inventory().await; - let write_quorum = (disks.len() / 2) + 1; - - let mut futures = Vec::with_capacity(disks.len()); - for disk in disks { - let bucket = bucket.to_string(); - futures.push(async move { - match disk { - Some(disk) => disk.stat_volume(&bucket).await, - None => Err(DiskError::DiskNotFound), - } - }); - } - - let results = join_all(futures).await; - let mut infos = Vec::with_capacity(results.len()); - let mut errs = Vec::with_capacity(results.len()); - for result in results { - match result { - Ok(info) => { - infos.push(Some(info)); - errs.push(None); - } - Err(err) => { - infos.push(None); - errs.push(Some(err)); - } - } - } - - if let Some(err) = reduce_write_quorum_errs(&errs, BUCKET_OP_IGNORED_ERRS, write_quorum) { - return Err(err.into()); - } - - let mut versioning = false; - let mut object_locking = false; + let mut info = self.stat_bucket_with_quorum(bucket, BucketInfoQuorum::Write).await?; if let Ok(sys) = metadata_sys::get(bucket).await { - versioning = sys.versioning(); - object_locking = sys.object_locking(); + info.versioning = sys.versioning(); + info.object_locking = sys.object_locking(); } - - infos - .into_iter() - .flatten() - .next() - .map(|info| BucketInfo { - name: info.name, - created: info.created, - versioning, - object_locking, - ..Default::default() - }) - .ok_or(Error::VolumeNotFound) + Ok(info) } #[tracing::instrument(skip(self))] diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs index c648255ad..71c175b97 100644 --- a/crates/ecstore/src/store/bucket.rs +++ b/crates/ecstore/src/store/bucket.rs @@ -19,7 +19,7 @@ use crate::bucket::{ }; use crate::error::is_err_bucket_not_found; use crate::runtime::sources as runtime_sources; -use crate::set_disk::get_lock_acquire_timeout; +use crate::set_disk::{BucketInfoQuorum, get_lock_acquire_timeout}; use crate::storage_api_contracts::bucket::{BUCKET_LIFECYCLE_LOCK_OBJECT, SRBucketDeleteOp}; use crate::storage_api_contracts::namespace::NamespaceLocking as _; use futures::stream::{self, StreamExt}; @@ -772,17 +772,30 @@ impl ECStore { #[instrument(skip(self))] pub(crate) async fn get_bucket_info_from_sets(&self, bucket: &str, opts: &BucketOptions) -> Result { + self.get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Write) + .await + } + + async fn get_bucket_info_from_sets_with_quorum( + &self, + bucket: &str, + opts: &BucketOptions, + quorum: BucketInfoQuorum, + ) -> Result { // One host may participate in several pools after expansion. Resolve the // namespace against each erasure set so disks from different pools can // never be combined into one bucket quorum. // Bucket validation is request-path IO. Keep the previous peer fanout's // latency shape by probing every set concurrently; scanner listings use // a separate bounded path below because they run continuously. - let mut scoped_results = - futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move { - (pool_index, set_index, set.get_bucket_info(bucket, opts).await) - })) - .await; + let mut scoped_results = futures::future::join_all(self.bucket_sets().map(|(pool_index, set_index, set)| async move { + let result = match quorum { + BucketInfoQuorum::Read => set.stat_bucket_with_quorum(bucket, quorum).await, + BucketInfoQuorum::Write => set.get_bucket_info(bucket, opts).await, + }; + (pool_index, set_index, result) + })) + .await; scoped_results.sort_unstable_by_key(|(pool_index, set_index, _)| (*pool_index, *set_index)); let mut first_info = None; @@ -806,7 +819,11 @@ impl ECStore { #[instrument(skip(self))] pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { - let mut info = self.get_bucket_info_from_sets(bucket, opts).await?; + let mut info = match self.get_bucket_info_from_sets(bucket, opts).await { + Ok(info) => info, + Err(Error::ErasureWriteQuorum) => return self.get_bucket_info_at_read_quorum(bucket, opts).await, + Err(err) => return Err(err), + }; if let Ok(sys) = metadata_sys::get_in(&self.ctx, bucket).await { if should_override_created_from_metadata(sys.created) { @@ -819,6 +836,35 @@ impl ECStore { Ok(info) } + async fn get_bucket_info_at_read_quorum(&self, bucket: &str, opts: &BucketOptions) -> Result { + // Lock order: bucket lifecycle -> internal metadata object read locks. + // Keep create/delete from changing the namespace while a read quorum + // confirms both physical presence and persisted bucket metadata. + let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?; + await_bucket_namespace_operation(Some(&guard), bucket, "bucket read quorum validation", async { + let mut info = self + .get_bucket_info_from_sets_with_quorum(bucket, opts, BucketInfoQuorum::Read) + .await?; + let (metadata, persisted) = metadata_sys::get_config_from_disk_with_presence_in(&self.ctx, bucket).await?; + if !persisted { + // A minority of directories left by failed creation is not an + // authoritative bucket. Never turn fabricated defaults into + // permission to serve degraded reads. + return Err(Error::ErasureReadQuorum); + } + if metadata.name != bucket { + return Err(Error::FileCorrupt); + } + if should_override_created_from_metadata(metadata.created) { + info.created = Some(metadata.created); + } + info.versioning = metadata.versioning(); + info.object_locking = metadata.object_locking(); + Ok(info) + }) + .await + } + #[instrument(skip(self))] pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result> { // TODO(backlog): support cached bucket listing via opts.cached @@ -1049,7 +1095,7 @@ mod tests { run_physical_bucket_deletion, scan_metadata_less_residue, scan_metadata_less_residue_with_budget, should_override_created_from_metadata, validate_table_bucket_delete_allowed, }; - use crate::bucket::metadata::table_bucket_catalog_metadata_prefix; + use crate::bucket::metadata::{BucketMetadata, table_bucket_catalog_metadata_prefix}; use crate::bucket::metadata_sys; use crate::cluster::rpc::peer_s3_client::install_delete_bucket_empty_scan_barrier; use crate::disk::{BUCKET_META_PREFIX, DiskAPI, RUSTFS_META_BUCKET, STORAGE_FORMAT_FILE}; @@ -1076,6 +1122,7 @@ mod tests { use std::sync::atomic::{AtomicBool, Ordering}; use std::time::{Duration, SystemTime}; use time::OffsetDateTime; + use tokio::io::AsyncReadExt; use tokio::sync::{Notify, OnceCell}; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -1359,11 +1406,18 @@ mod tests { } async fn setup_multi_pool_bucket_test_env() -> (tempfile::TempDir, Arc) { + setup_bucket_quorum_test_env(&[4, 4], None).await + } + + async fn setup_bucket_quorum_test_env( + drives_per_pool: &[usize], + standard_parity: Option, + ) -> (tempfile::TempDir, Arc) { let temp_dir = tempfile::tempdir().expect("multi-pool bucket test directory should be created"); let mut pools = Vec::new(); - for pool_index in 0..2 { + for (pool_index, &drive_count) in drives_per_pool.iter().enumerate() { let mut endpoints = Vec::new(); - for disk_index in 0..4 { + for disk_index in 0..drive_count { let disk_path = temp_dir.path().join(format!("pool{pool_index}-disk{disk_index}")); tokio::fs::create_dir_all(&disk_path) .await @@ -1378,7 +1432,7 @@ mod tests { pools.push(PoolEndpoints { legacy: false, set_count: 1, - drives_per_set: 4, + drives_per_set: drive_count, endpoints: Endpoints::from(endpoints), cmd_line: format!("bucket-test-pool-{pool_index}"), platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), @@ -1399,9 +1453,12 @@ mod tests { ) .await .expect("multi-pool ECStore should initialize"); - let storage_class = - crate::config::storageclass::lookup_config_for_pools_without_env(&rustfs_config::server_config::KVS::new(), &[4, 4]) - .expect("multi-pool storage class should match both four-disk pools"); + let mut storage_class_kvs = rustfs_config::server_config::KVS::new(); + if let Some(parity) = standard_parity { + storage_class_kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), format!("EC:{parity}")); + } + let storage_class = crate::config::storageclass::lookup_config_for_pools_without_env(&storage_class_kvs, drives_per_pool) + .expect("storage class should match every test erasure set"); for pool in &ecstore.pools { for set in &pool.disk_set { set.set_test_storage_class_config(storage_class.clone()); @@ -2067,6 +2124,218 @@ mod tests { } } + #[tokio::test] + #[serial] + async fn bucket_info_read_quorum_tracks_erasure_layout() { + for (drive_count, parity) in [(2, 1), (3, 1), (4, 2), (5, 2), (6, 3), (8, 4), (6, 2), (12, 6)] { + let (_temp_dir, store) = setup_bucket_quorum_test_env(&[drive_count], Some(parity)).await; + metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("read-quorum-{drive_count}-{parity}"); + let object = "uncached-object"; + let body = b"erasure read quorum must follow the persisted layout".repeat(32_768); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("healthy namespace should accept bucket creation"); + store + .put_object(&bucket, object, &mut PutObjReader::from_vec(body.clone()), &ObjectOptions::default()) + .await + .expect("healthy erasure set should accept the seed object"); + let set = &store.pools[0].disk_set[0]; + let lock = set + .new_ns_lock(&bucket, object) + .await + .expect("seed namespace lock should resolve"); + drop( + lock.get_write_lock(Duration::from_secs(30)) + .await + .expect("seed physical fanout must finish before taking disks offline"), + ); + if (drive_count, parity) == (6, 3) { + let mut kvs = rustfs_config::server_config::KVS::new(); + kvs.insert(crate::config::storageclass::CLASS_STANDARD.to_string(), "EC:2".to_string()); + set.set_test_storage_class_config( + crate::config::storageclass::lookup_config_for_pools_without_env(&kvs, &[drive_count]) + .expect("a later storage-class change must not raise old objects' read quorum"), + ); + } + + let offline_indexes = (0..parity).collect::>(); + let offline = take_set_disks_offline(&store, set, &offline_indexes).await; + let info = store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .expect("bucket validation must admit the object's exact read quorum"); + assert_eq!(info.name, bucket); + + let mut reader = store + .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default()) + .await + .expect("the persisted layout should remain readable at its exact data-shard quorum"); + let mut restored = Vec::new(); + reader + .stream + .read_to_end(&mut restored) + .await + .expect("quorum read should reconstruct the body"); + assert_eq!(restored, body, "layout {drive_count}/{parity} must retain exact object contents"); + drop(reader); + + if drive_count - parity == drive_count / 2 { + let error = store + .get_bucket_info_from_sets(&bucket, &BucketOptions::default()) + .await + .expect_err("bucket mutations must retain their majority namespace check"); + assert_eq!(error, StorageError::ErasureWriteQuorum); + } + + let below_quorum = take_set_disks_offline(&store, set, &[parity]).await; + let read = store + .get_object_reader(&bucket, object, None, Default::default(), &ObjectOptions::default()) + .await; + match read { + Ok(mut reader) => assert!( + reader.stream.read_to_end(&mut Vec::new()).await.is_err(), + "layout {drive_count}/{parity} must reject fewer than its data-shard quorum" + ), + Err(error) => assert!( + matches!(error, StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _)), + "a missing shard must report read quorum loss, got {error}" + ), + } + restore_set_disks(&store, set, below_quorum).await; + restore_set_disks(&store, set, offline).await; + } + } + + #[tokio::test] + #[serial] + async fn bucket_info_read_quorum_is_scoped_to_each_erasure_set() { + let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4, 6], None).await; + metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = "read-quorum-mixed-pools"; + store + .make_bucket(bucket, &MakeBucketOptions::default()) + .await + .expect("healthy pools should accept bucket creation"); + + let first_set = &store.pools[0].disk_set[0]; + let second_set = &store.pools[1].disk_set[0]; + let first_offline = take_set_disks_offline(&store, first_set, &[0, 1]).await; + let second_offline = take_set_disks_offline(&store, second_set, &[0, 1, 2]).await; + store + .get_bucket_info(bucket, &BucketOptions::default()) + .await + .expect("each set independently satisfies its namespace read quorum"); + + for (set, extra_disk) in [(first_set, 2), (second_set, 3)] { + let extra_offline = take_set_disks_offline(&store, set, &[extra_disk]).await; + assert_eq!( + store + .get_bucket_info(bucket, &BucketOptions::default()) + .await + .expect_err("another pool must not subsidize a set below its read quorum"), + StorageError::ErasureReadQuorum + ); + restore_set_disks(&store, set, extra_offline).await; + } + restore_set_disks(&store, first_set, first_offline).await; + restore_set_disks(&store, second_set, second_offline).await; + } + + #[tokio::test] + #[serial] + async fn bucket_info_read_quorum_requires_authoritative_metadata() { + for state in ["missing", "corrupt", "foreign", "incarnation"] { + let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await; + metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("read-quorum-{state}-metadata"); + let mut metadata = if state == "missing" { + store + .make_bucket_on_sets(&bucket, &MakeBucketOptions::default()) + .await + .expect("simulate directories left before bucket metadata is published"); + BucketMetadata::new(&bucket) + } else { + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("healthy bucket should publish metadata"); + metadata_sys::get_in(&store.ctx, &bucket) + .await + .expect("seed metadata should be cached") + .as_ref() + .clone() + }; + let path = metadata.save_file_path(); + match state { + "corrupt" => crate::config::com::save_config(store.clone(), &path, b"corrupt".to_vec()) + .await + .expect("persist corrupt metadata while the cached copy remains valid"), + "foreign" => { + metadata.name = "different-bucket".to_string(); + let mut encoded = vec![1, 0, 1, 0]; + encoded.extend(metadata.marshal_msg().expect("foreign metadata should encode")); + crate::config::com::save_config(store.clone(), &path, encoded) + .await + .expect("persist metadata for a different bucket at the requested path"); + } + "incarnation" => crate::bucket::metadata::save_bucket_incarnation(store.clone(), &bucket, Uuid::new_v4()) + .await + .expect("persist a different bucket generation"), + _ => {} + } + + let set = &store.pools[0].disk_set[0]; + let offline = take_set_disks_offline(&store, set, &[0, 1]).await; + let error = store + .get_bucket_info(&bucket, &BucketOptions::default()) + .await + .expect_err("read admission must not trust residual directories or cached metadata"); + match state { + "missing" => assert_eq!(error, StorageError::ErasureReadQuorum), + "foreign" => assert_eq!(error, StorageError::FileCorrupt), + "incarnation" => assert!(error.to_string().contains("sidecar does not match bucket metadata")), + "corrupt" => assert!(error.to_string().contains("format invalid"), "unexpected corruption error: {error}"), + _ => unreachable!(), + } + restore_set_disks(&store, set, offline).await; + } + } + + #[tokio::test] + #[serial] + async fn bucket_info_read_quorum_accepts_persisted_legacy_metadata() { + let (_temp_dir, store) = setup_bucket_quorum_test_env(&[4], None).await; + metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = "interop"; + store + .make_bucket_on_sets(bucket, &MakeBucketOptions::default()) + .await + .expect("legacy bucket directories should exist"); + let hex = include_str!("../../tests/fixtures/minio/bucket_metadata.blob.hex") + .split_whitespace() + .collect::(); + let body = (0..hex.len()) + .step_by(2) + .map(|index| u8::from_str_radix(&hex[index..index + 2], 16).expect("pinned MinIO metadata fixture")) + .collect(); + crate::config::com::save_config(store.clone(), &BucketMetadata::new(bucket).save_file_path(), body) + .await + .expect("legacy metadata should be persisted without an incarnation sidecar"); + + let set = &store.pools[0].disk_set[0]; + let offline = take_set_disks_offline(&store, set, &[0, 1]).await; + let info = store + .get_bucket_info(bucket, &BucketOptions::default()) + .await + .expect("persisted MinIO metadata should authorize reads at the namespace read quorum"); + assert_eq!(info.name, bucket); + assert!(info.versioning); + assert!(info.object_locking); + restore_set_disks(&store, set, offline).await; + } + #[tokio::test] #[serial] async fn bucket_namespace_reads_report_missing_when_every_set_is_absent() { @@ -2090,6 +2359,7 @@ mod tests { #[serial] async fn bucket_namespace_reads_fail_closed_when_any_set_loses_quorum() { let (_temp_dir, ecstore) = setup_multi_pool_bucket_test_env().await; + metadata_sys::init_bucket_metadata_sys(ecstore.clone(), Vec::new()).await; let bucket = format!("degraded-expansion-{}", Uuid::new_v4().simple()); ecstore.pools[0].disk_set[0] .make_bucket(&bucket, &MakeBucketOptions::default()) @@ -2097,6 +2367,7 @@ mod tests { .expect("bucket should be created in the original pool only"); ecstore.pools[1].disk_set[0].disks.write().await[0] = None; ecstore.pools[1].disk_set[0].disks.write().await[1] = None; + ecstore.pools[1].disk_set[0].disks.write().await[2] = None; let list_err = ecstore .list_bucket(&BucketOptions::default()) @@ -2108,7 +2379,7 @@ mod tests { .get_bucket_info(&bucket, &BucketOptions::default()) .await .expect_err("bucket validation must fail when an expansion pool is unavailable"); - assert_eq!(info_err, StorageError::ErasureWriteQuorum); + assert_eq!(info_err, StorageError::ErasureReadQuorum); } #[tokio::test] From 35b5cfcf8d023ee4e8fcc6806566578990594ab2 Mon Sep 17 00:00:00 2001 From: JaySon Date: Wed, 9 Sep 2026 16:39:25 +0800 Subject: [PATCH 5/7] docs: fix invalid docker-buildx.sh usage example in README and README_ZH (#7570) docs: fix invalid docker-buildx.sh usage in README and README_ZH Replace the docker-buildx.sh --build-arg RELEASE=latest example, which the script never accepted as a CLI flag (--build-arg is only used internally for docker buildx build), with the supported invocations: - bare ./docker-buildx.sh for the default local build - ./docker-buildx.sh -p linux/amd64 for a single-platform local build Also update the surrounding comments to reflect single-platform local builds and add the multi-arch example comment accordingly. --- README.md | 5 ++++- README_ZH.md | 5 ++++- 2 files changed, 8 insertions(+), 2 deletions(-) diff --git a/README.md b/README.md index 11301393b..a70c36047 100644 --- a/README.md +++ b/README.md @@ -211,7 +211,10 @@ For developers who want to build RustFS Docker images from source with multi-arc ```bash # Build multi-architecture images locally -./docker-buildx.sh --build-arg RELEASE=latest +./docker-buildx.sh + +# Build a single-platform image locally +./docker-buildx.sh -p linux/amd64 # Build and push to registry ./docker-buildx.sh --push diff --git a/README_ZH.md b/README_ZH.md index baf6dab80..2951a5ddf 100644 --- a/README_ZH.md +++ b/README_ZH.md @@ -150,7 +150,10 @@ docker compose -f docker-compose-simple.yml up -d ```bash # 在本地构建多架构镜像 -./docker-buildx.sh --build-arg RELEASE=latest +./docker-buildx.sh + +# 在本地构建单平台镜像 +./docker-buildx.sh -p linux/amd64 # 构建并推送到仓库 ./docker-buildx.sh --push From e546ae9c621b14fbcf130d19dccda730d7eaef4f Mon Sep 17 00:00:00 2001 From: hector <42570491+majinghe@users.noreply.github.com> Date: Wed, 9 Sep 2026 16:39:59 +0800 Subject: [PATCH 6/7] =?UTF-8?q?fix(nightly):=20push=20assets=20with=20plai?= =?UTF-8?q?n=20git=20=E2=80=94=20the=20build=20fleet=20has=20no=20gh=20CLI?= =?UTF-8?q?=20(#7572)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The sm-standard-4 runners used by the nightly build have no gh binary (functional-chain workflows run elsewhere, on the jumpbox). The assets publish step died on 'gh: command not found' at the credential-helper setup, and the preceding clone failure had been masked by 2>/dev/null, misleading the step into the orphan path. Swap clone and remote setup to plain git with the token embedded in the URL; push semantics are unchanged. --- .github/workflows/nightly-gnu.yml | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/.github/workflows/nightly-gnu.yml b/.github/workflows/nightly-gnu.yml index 6d7920f5e..8f1480084 100644 --- a/.github/workflows/nightly-gnu.yml +++ b/.github/workflows/nightly-gnu.yml @@ -372,15 +372,18 @@ jobs: [ -f "$f" ] || { echo "missing package: $f"; exit 1; } done + # Plain git + token URL: the build fleet has no gh CLI. + ASSETS_URL="https://x-access-token:${ASSETS_TOKEN}@github.com/rustfs/auto-testing.git" rm -rf assets-work && mkdir assets-work - if ! gh repo clone rustfs/auto-testing assets-work -- --depth 1 --branch assets --quiet 2>/dev/null; then + if git clone -q --depth 1 --branch assets "${ASSETS_URL}" assets-work 2>/dev/null; then + echo "assets branch cloned" + else echo "assets branch does not exist yet; creating an orphan" - ( cd assets-work && git init -q -b assets ) + git -C assets-work init -q -b assets fi cd assets-work - git remote add origin "https://github.com/rustfs/auto-testing.git" 2>/dev/null || \ - git remote set-url origin "https://github.com/rustfs/auto-testing.git" - gh auth setup-git >/dev/null + git remote add origin "${ASSETS_URL}" 2>/dev/null || \ + git remote set-url origin "${ASSETS_URL}" mkdir -p nightly cp "../${DEB_FILE}" "nightly/${DEB_FILE}" From 27d66c159fff13283eadb17d6c7d45d07722d2b7 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Wed, 9 Sep 2026 19:05:26 +0800 Subject: [PATCH 7/7] fix(kms): prevent transient health failures from latching status (#7578) Keep backend health checks from overwriting the running service lifecycle state so subsequent admin checks and probe-based readiness can recover without a restart. Add a regression test that fails on the original implementation after backend recovery and verifies the service instance and version remain unchanged. Validation: 39 focused resilience, lifecycle, concurrency, and service manager tests passed; one existing live AWS test remained ignored. cargo fmt --all --check and git diff --check passed. Thanks to @stevapple for reporting the issue and providing a detailed diagnosis and reproduction. Fixes #7554 --- crates/kms/src/service_manager.rs | 33 ++++----------------- crates/kms/tests/behavior_resilience.rs | 38 +++++++++++++++++++++++++ 2 files changed, 43 insertions(+), 28 deletions(-) diff --git a/crates/kms/src/service_manager.rs b/crates/kms/src/service_manager.rs index 31c62bce0..67ab5b17c 100644 --- a/crates/kms/src/service_manager.rs +++ b/crates/kms/src/service_manager.rs @@ -577,13 +577,16 @@ impl KmsServiceManager { Some(service_version.probe_worker.as_ref()?.status()) } - /// Health check for the KMS service + /// Check backend health without changing the service lifecycle state. + /// + /// A transient backend failure leaves the published service available for + /// subsequent checks and operations. Readiness uses the background probe + /// to evaluate backend availability independently of lifecycle state. pub async fn health_check(&self) -> Result { let checked_state = self.state.load_full(); match checked_state.current_service.as_ref() { Some(service_version) => { let manager = service_version.manager.clone(); - let checked_version = service_version.version; // Perform health check on the backend match manager.health_check().await { Ok(healthy) => { @@ -594,8 +597,6 @@ impl KmsServiceManager { } Err(e) => { error!("KMS health check error: {}", e); - let _guard = self.lifecycle_mutex.lock().await; - self.mark_health_error_if_current(checked_version, &e); Err(e) } } @@ -739,17 +740,6 @@ impl KmsServiceManager { task: std::sync::Mutex::new(Some(task)), })) } - - fn mark_health_error_if_current(&self, checked_version: u64, error: &KmsError) { - let current = self.state.load_full(); - if current.current_service.as_ref().map(|version| version.version) == Some(checked_version) { - self.state.store(Arc::new(RuntimeState { - config: current.config.clone(), - status: KmsServiceStatus::Error(format!("Health check failed: {error}")), - current_service: current.current_service.clone(), - })); - } - } } impl Default for KmsServiceManager { @@ -1004,19 +994,6 @@ mod tests { assert!(manager.get_service_version().await.expect("restarted version") > first_version); } - #[tokio::test] - async fn stale_health_failure_cannot_poison_new_service_status() { - let manager = KmsServiceManager::new(); - manager.configure(static_config("key-a", 0x11)).await.expect("configure"); - manager.start().await.expect("start"); - let old_version = manager.get_service_version().await.expect("old version"); - manager.restart().await.expect("restart"); - - manager.mark_health_error_if_current(old_version, &KmsError::backend_error("stale failure")); - - assert_eq!(manager.get_status().await, KmsServiceStatus::Running); - } - #[tokio::test] async fn forbidden_local_master_key_change_preserves_running_config_and_service() { use crate::types::{CreateKeyRequest, KeyUsage}; diff --git a/crates/kms/tests/behavior_resilience.rs b/crates/kms/tests/behavior_resilience.rs index 536257978..77d8de17d 100644 --- a/crates/kms/tests/behavior_resilience.rs +++ b/crates/kms/tests/behavior_resilience.rs @@ -75,6 +75,44 @@ fn unreachable_vault_config() -> KmsConfig { } } +#[tokio::test] +async fn transient_health_failure_does_not_latch_the_service_status() { + let kms = TestKms::local().await; + let manager = kms.manager(); + let service = manager.get_encryption_service().await.expect("running service"); + let version = manager.get_service_version().await.expect("running version"); + assert!(manager.health_check().await.expect("initial backend health")); + + // Move only this test's keys out of reach, then restore the same backend. + let key_dir = kms.key_dir().expect("local key directory"); + let outage = tempfile::TempDir::new().expect("temporary outage directory"); + let hidden_keys = outage.path().join("keys"); + tokio::fs::rename(&key_dir, &hidden_keys) + .await + .expect("make backend unavailable"); + let failure = manager.health_check().await; + let outage_status = manager.get_status().await; + tokio::fs::rename(&hidden_keys, &key_dir).await.expect("restore backend"); + + assert!(failure.is_err(), "the outage must surface as a health-check error"); + assert!(manager.health_check().await.expect("backend recovers without restart")); + assert!(Arc::ptr_eq( + &service, + &manager.get_encryption_service().await.expect("service survives the outage") + )); + assert_eq!(manager.get_service_version().await, Some(version)); + assert_eq!( + manager.get_status().await, + KmsServiceStatus::Running, + "a recovered backend must not leave service-status and readiness latched in Error" + ); + assert_eq!( + outage_status, + KmsServiceStatus::Running, + "backend health does not change the running service's lifecycle state" + ); +} + #[tokio::test] async fn starting_against_an_unreachable_backend_fails_without_publishing_a_service() { let manager = KmsServiceManager::new();