test(ecstore): exercise heal during committed overwrite cleanup

.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

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;

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