From 6afdec9fe1b0b88007f2e5b445c19bf73d51d3ca Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 20 Oct 2026 04:07:05 +0800 Subject: [PATCH] test(ecstore): exercise heal during committed overwrite cleanup --- .github/workflows/rustfs-validation-tools.yml | 31 +++- crates/ecstore/src/set_disk/ops/object.rs | 160 ++++++++++++++++++ scripts/test_functional_chain.py | 94 ++++++++++ 3 files changed, 281 insertions(+), 4 deletions(-) diff --git a/.github/workflows/rustfs-validation-tools.yml b/.github/workflows/rustfs-validation-tools.yml index 12e0564a8..44dbf76e6 100644 --- a/.github/workflows/rustfs-validation-tools.yml +++ b/.github/workflows/rustfs-validation-tools.yml @@ -160,14 +160,36 @@ jobs: run: | set -euo pipefail test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}" - python3 -B auto-testing/scripts/validation_heal/native_regressions.py build \ - --source candidate --commit "${CANDIDATE_SHA}" --out native-tools + use_race_native=false + if [[ -f auto-testing/scripts/validation_heal/race_native.py ]]; then + support_rc=0 + python3 -B auto-testing/scripts/validation_heal/race_native.py supports \ + --commit "${CANDIDATE_SHA}" || support_rc=$? + case "${support_rc}" in + 0) use_race_native=true ;; + 2) ;; # The original exact-source collector handles legacy candidates. + *) exit "${support_rc}" ;; + esac + fi + if [[ "${use_race_native}" == true ]]; then + python3 -B auto-testing/scripts/validation_heal/race_native.py build \ + --source candidate --commit "${CANDIDATE_SHA}" --testing-commit "${AUTO_TESTING_SHA}" --out native-tools + else + python3 -B auto-testing/scripts/validation_heal/native_regressions.py build \ + --source candidate --commit "${CANDIDATE_SHA}" --out native-tools + fi - name: Run exact native regressions separately from fleet validation run: | set -euo pipefail - python3 -B auto-testing/scripts/validation_heal/native_regressions.py run \ - --directory native-tools --out native-evidence + if [[ -f native-tools/race-native.json ]]; then + test ! -e native-tools/native-regressions.json + python3 -B auto-testing/scripts/validation_heal/race_native.py run \ + --directory native-tools --out native-evidence + else + python3 -B auto-testing/scripts/validation_heal/native_regressions.py run \ + --directory native-tools --out native-evidence + fi - name: Run existing F07 directory fsync failure regressions if: ${{ inputs.candidate_sha == '1aa9be87d18cf941e5020f19e8622c26ba4b10ac' }} @@ -195,6 +217,7 @@ jobs: name: rustfs-validation-native-${{ github.run_id }}-${{ github.run_attempt }} path: | native-tools/ecstore-native-tests + native-tools/ecstore-race-native-tests native-tools/*.json native-tools/*.jsonl native-tools/*.stderr diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ff6e5ebae..79afc4244 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -21108,6 +21108,166 @@ mod put_object_tmp_cleanup_tests { .await; } + #[tokio::test] + async fn deep_heal_preserves_committed_generation_during_old_data_cleanup() { + use rustfs_heal_contracts::heal_channel::{HealOpts, HealScanMode}; + + for (disk_count, parity, read_repair) in [(4, 2, false), (4, 2, true), (16, 4, false), (16, 4, true)] { + let (dirs, disks, set) = super::hermetic_set_disks_support::hermetic_set_disks_for_pool_with_default_parity_isolated( + disk_count, 0, parity, + ) + .await; + let bucket = "native-heal-commit-cleanup"; + let object = format!("generation-{disk_count}-{parity}-{read_repair}.bin"); + for disk in &disks { + disk.make_volume(bucket).await.expect("create actual fixture volume"); + } + let options = ObjectOptions { + write_completion: crate::object_api::WriteCompletion::TailDrained, + shard_integrity_write_mode: Some(crate::object_api::ShardIntegrityWriteMode::Protected), + ..Default::default() + }; + set.put_object(bucket, &object, &mut PutObjReader::from_vec(vec![0x17; TEST_OBJECT_SIZE]), &options) + .await + .expect("acknowledge original generation on every disk"); + let old = disks[0] + .read_version("", bucket, &object, "", &ReadOptions::default()) + .await + .expect("read actual original generation"); + let old_dir = old.data_dir.expect("original external data directory"); + let old_part = dirs[0] + .path() + .join(bucket) + .join(&object) + .join(old_dir.to_string()) + .join("part.1"); + assert!(tokio::fs::try_exists(&old_part).await.expect("original part exists")); + + let barrier = rename_fanout_barrier::arm(&object, 0, rename_fanout_barrier::PHASE_CLEANUP); + let writer_set = Arc::clone(&set); + let writer_object = object.clone(); + let writer_options = options.clone(); + let payload = vec![0xa3; TEST_OBJECT_SIZE]; + let writer_payload = payload.clone(); + let writer = tokio::spawn(async move { + writer_set + .put_object(bucket, &writer_object, &mut PutObjReader::from_vec(writer_payload), &writer_options) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("overwrite commits before its real old-data cleanup pause"); + assert!(!writer.is_finished(), "tail-drained ACK must remain pending at cleanup"); + assert!( + tokio::fs::try_exists(&old_part) + .await + .expect("old source still present at pause") + ); + + let mut committed = Vec::new(); + let mut healthy_metadata = Vec::new(); + for disk in &disks { + let info = disk + .read_version("", bucket, &object, "", &ReadOptions::default()) + .await + .expect("read independently committed physical generation without GET"); + assert_eq!((info.erasure.data_blocks, info.erasure.parity_blocks), (disk_count - parity, parity)); + assert!(info.data.is_none(), "use actual external parts"); + assert_ne!(info.data_dir, Some(old_dir)); + committed.push(info); + healthy_metadata.push( + disk.read_all(bucket, &format!("{object}/xl.meta")) + .await + .expect("committed raw metadata"), + ); + } + let new_dir = committed[0].data_dir.expect("new committed data directory"); + assert!(committed.iter().all(|info| info.data_dir == Some(new_dir))); + let new_part = dirs[0] + .path() + .join(bucket) + .join(&object) + .join(new_dir.to_string()) + .join("part.1"); + let expected_part = tokio::fs::read(&new_part).await.expect("exact new source bytes"); + tokio::fs::remove_file(&new_part) + .await + .expect("remove only the newly committed target shard"); + println!( + "native_g01: geometry={disk_count}/{parity} read_repair={read_repair} old_cleanup_paused new_generation_committed target_part_missing" + ); + + let (result, error) = tokio::time::timeout( + Duration::from_secs(30), + set.heal_object( + bucket, + &object, + "", + &HealOpts { + scan_mode: HealScanMode::Deep, + read_repair, + ..Default::default() + }, + ), + ) + .await + .expect("real locked Heal must progress during old cleanup") + .expect("complete actual deep Heal"); + assert!(error.is_none(), "new generation must remain repairable: {error:?}"); + assert_eq!(result.drives_healed(), Some(1)); + // Shared data-phase repair deliberately does not certify full integrity. + assert_eq!(result.integrity_verified, !read_repair); + assert_eq!(tokio::fs::read(&new_part).await.expect("rebuilt part before any GET"), expected_part); + let repaired = disks[0] + .read_version("", bucket, &object, "", &ReadOptions::default()) + .await + .expect("read actual repaired generation"); + assert_eq!(repaired.data_dir, Some(new_dir)); + assert_eq!(repaired.mod_time, committed[0].mod_time); + for (disk, expected) in disks.iter().zip(&healthy_metadata).skip(1) { + assert_eq!( + disk.read_all(bucket, &format!("{object}/xl.meta")) + .await + .expect("healthy raw metadata after Heal"), + *expected + ); + } + println!( + "native_g01: geometry={disk_count}/{parity} read_repair={read_repair} locked_heal_completed new_generation_part_rebuilt_before_get" + ); + + barrier.release(); + let acknowledged = tokio::time::timeout(Duration::from_secs(30), writer) + .await + .expect("cleanup drains") + .expect("writer joins") + .expect("overwrite ACK remains successful"); + assert_eq!(acknowledged.size, i64::try_from(payload.len()).expect("fixture size fits i64")); + drop(barrier); + let mut reader = set + .get_object_reader(bucket, &object, None, HeaderMap::new(), &options) + .await + .expect("read final acknowledged generation"); + let mut body = Vec::new(); + reader + .stream + .read_to_end(&mut body) + .await + .expect("read complete final payload"); + assert_eq!(body, payload); + for disk in &disks { + let final_info = disk + .read_version("", bucket, &object, "", &ReadOptions::default()) + .await + .expect("final physical generation"); + assert_eq!(final_info.data_dir, Some(new_dir), "old cleanup must not revert the healed generation"); + } + println!( + "native_g01: geometry={disk_count}/{parity} read_repair={read_repair} overwrite_ack_preserved final_physical_generation_and_full_body_verified" + ); + } + } + #[tokio::test] async fn committed_put_releases_namespace_lock_before_old_data_cleanup() { let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await; diff --git a/scripts/test_functional_chain.py b/scripts/test_functional_chain.py index 177094a7b..95e40da16 100644 --- a/scripts/test_functional_chain.py +++ b/scripts/test_functional_chain.py @@ -16,6 +16,100 @@ import functional_chain_evidence as evidence import resolve_functional_candidate as candidate +class RaceNativeToolsWorkflowTests(unittest.TestCase): + def shell_step(self, name): + from check_test_wiring import yaml_block + source = (candidate.ROOT / '.github/workflows/rustfs-validation-tools.yml').read_text() + job = yaml_block(source.splitlines(), 'native-heal-regressions', 2) + start = job.index(' - name: ' + name) + end = next((index for index in range(start + 1, len(job)) + if job[index].startswith(' - name: ')), len(job)) + step = job[start:end] + return '\n'.join(line[10:] for line in step[step.index(' run: |') + 1:] + if not line.strip() or line.startswith(' ')) + + def fixture(self, root, *, new_helper): + directory = root / 'auto-testing/scripts/validation_heal' + directory.mkdir(parents=True) + stub = '''import json, os, pathlib, sys +kind = pathlib.Path(__file__).stem +with pathlib.Path('calls.jsonl').open('a') as out: + out.write(json.dumps({'kind': kind, 'args': sys.argv[1:]}) + '\\n') +if sys.argv[1] == 'supports': sys.exit(int(os.environ['SUPPORT_STATUS'])) +sys.exit(int(os.environ.get(kind.upper() + '_FAIL', '0'))) +''' + (directory / 'native_regressions.py').write_text(stub) + if new_helper: (directory / 'race_native.py').write_text(stub) + bin_dir = root / 'bin' + bin_dir.mkdir() + git = bin_dir / 'git' + git.write_text('#!/bin/sh\nprintf "%s\\n" "$FAKE_HEAD"\n') + git.chmod(0o755) + env = dict(os.environ, AUTO_TESTING_SHA='b' * 40, CANDIDATE_SHA='a' * 40, + FAKE_HEAD='b' * 40, SUPPORT_STATUS='2', PATH=str(bin_dir) + os.pathsep + os.environ['PATH']) + env.pop('BASH_ENV', None) + return env + + def test_actual_native_build_routes_old_new_unknown_and_failed_collectors(self): + shell = self.shell_step('Build and enumerate the unmodified candidate test executable') + for mode, helper, support, failure, head, expected_code, expected_calls in [ + ('old-harness', False, 2, None, 'b', 0, [('native_regressions', 'build')]), + ('legacy', True, 2, None, 'b', 0, [('race_native', 'supports'), ('native_regressions', 'build')]), + ('race', True, 0, None, 'b', 0, [('race_native', 'supports'), ('race_native', 'build')]), + ('support-error', True, 42, None, 'b', 42, [('race_native', 'supports')]), + ('invalid-support', True, 1, None, 'b', 1, [('race_native', 'supports')]), + ('race-build-error', True, 0, 'RACE_NATIVE_FAIL', 'b', 42, [('race_native', 'supports'), ('race_native', 'build')]), + ('legacy-build-error', True, 2, 'NATIVE_REGRESSIONS_FAIL', 'b', 42, [('race_native', 'supports'), ('native_regressions', 'build')]), + ('wrong-helper', True, 0, None, 'c', 1, []), + ]: + with tempfile.TemporaryDirectory() as temp, self.subTest(mode=mode): + root = Path(temp) + env = self.fixture(root, new_helper=helper) + env.update(SUPPORT_STATUS=str(support), FAKE_HEAD=head * 40) + if failure: env[failure] = '42' + result = subprocess.run(['bash', '-e', '-o', 'pipefail', '-c', shell], + cwd=root, env=env, capture_output=True, text=True) + self.assertEqual(result.returncode, expected_code, result.stderr) + calls = [json.loads(line) for line in (root / 'calls.jsonl').read_text().splitlines()] if (root / 'calls.jsonl').exists() else [] + self.assertEqual([(row['kind'], row['args'][0]) for row in calls], expected_calls) + for row in calls: + if row['args'][0] == 'supports': + self.assertEqual(row['args'], ['supports', '--commit', 'a' * 40]) + else: + expected = ['build', '--source', 'candidate', '--commit', 'a' * 40] + if row['kind'] == 'race_native': expected += ['--testing-commit', 'b' * 40] + self.assertEqual(row['args'], expected + ['--out', 'native-tools']) + + def test_actual_native_run_rejects_ambiguous_formats_and_propagates_failures(self): + shell = self.shell_step('Run exact native regressions separately from fleet validation') + for mode, helper, manifests, failure, expected_code, expected_kind in [ + ('legacy', False, ['native-regressions.json'], None, 0, 'native_regressions'), + ('race', True, ['race-native.json'], None, 0, 'race_native'), + ('ambiguous', True, ['native-regressions.json', 'race-native.json'], None, 1, None), + ('legacy-error', True, ['native-regressions.json'], 'NATIVE_REGRESSIONS_FAIL', 42, 'native_regressions'), + ('race-error', True, ['race-native.json'], 'RACE_NATIVE_FAIL', 42, 'race_native'), + ('missing-race-helper', False, ['race-native.json'], None, 2, None), + ]: + with tempfile.TemporaryDirectory() as temp, self.subTest(mode=mode): + root = Path(temp) + env = self.fixture(root, new_helper=helper) + if failure: env[failure] = '42' + (root / 'native-tools').mkdir() + for manifest in manifests: (root / 'native-tools' / manifest).write_text('{}') + result = subprocess.run(['bash', '-e', '-o', 'pipefail', '-c', shell], + cwd=root, env=env, capture_output=True, text=True) + self.assertEqual(result.returncode, expected_code, result.stderr) + calls = [json.loads(line) for line in (root / 'calls.jsonl').read_text().splitlines()] if (root / 'calls.jsonl').exists() else [] + self.assertEqual(len(calls), int(expected_kind is not None)) + if expected_kind: + self.assertEqual(calls, [{'kind': expected_kind, 'args': ['run', '--directory', 'native-tools', '--out', 'native-evidence']}]) + + def test_native_upload_retains_both_fixed_executable_kinds(self): + source = (candidate.ROOT / '.github/workflows/rustfs-validation-tools.yml').read_text() + self.assertIn(' native-tools/ecstore-native-tests\n', source) + self.assertIn(' native-tools/ecstore-race-native-tests\n', source) + + class PersistenceToolsWorkflowTests(unittest.TestCase): def test_supplement_preserves_old_pins_and_propagates_actual_collector_failure(self): from check_test_wiring import yaml_block