ci(validation): bind recovery to retained runner evidence

.config/make/tests.mak

@@ -63,6 +63,8 @@ script-tests: ## Run shell script tests
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_nightly_candidate.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_functional_chain.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_functional_chain_health.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_inspect_validation_runner.py
+	$(RUSTFS_PYTHON_BIN) ./scripts/test_retain_validation_evidence.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/test_ci_timing_report.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_report_compat.py
 	$(RUSTFS_PYTHON_BIN) ./scripts/s3-tests/test_runner_tools.py

.github/workflows/rustfs-functional-chain.yml

@@ -90,6 +90,11 @@ on:
         type: string
         required: false
         default: ''
+      validation_cleanup_retained_evidence:
+        description: Recover from verified persistent runner-local evidence when the failed run uploaded no artifact
+        type: boolean
+        required: false
+        default: false
       validation_tools_run_id:
         description: Successful tool build run; required by Heal physical metadata checks
         type: string
@@ -187,7 +192,7 @@ jobs:
               raise ValueError("package build requires an exact source SHA")
           exclusive = ("validation_retain_evidence_run_id", "validation_runner_inspect_only", "build_run_id", "build_run_attempt", "validation_case", "auto_testing_sha",
                        "validation_tools_only", "validation_native_tools", "validation_shared_tools",
-                       "validation_manager_tools", "validation_cleanup_run_id", "validation_tools_run_id")
+                       "validation_manager_tools", "validation_cleanup_run_id", "validation_cleanup_retained_evidence", "validation_tools_run_id")
           if any(value.get(name) for name in exclusive):
               raise ValueError("package-only build cannot select existing candidate, tools, fleet or recovery execution")
           PY
@@ -205,7 +210,7 @@ jobs:
       R2_BUCKET: ${{ secrets.R2_BUCKET }}
 
   prepare:
-    if: ${{ (github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && inputs.validation_case == '' && !inputs.validation_tools_only && !inputs.validation_native_tools && !inputs.validation_shared_tools && !inputs.validation_manager_tools && inputs.validation_cleanup_run_id == '') || github.event.workflow_run.event == 'schedule' }}
+    if: ${{ (github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && inputs.validation_case == '' && !inputs.validation_tools_only && !inputs.validation_native_tools && !inputs.validation_shared_tools && !inputs.validation_manager_tools && !inputs.validation_cleanup_retained_evidence && inputs.validation_cleanup_run_id == '') || github.event.workflow_run.event == 'schedule' }}
     runs-on: smoke-testing
     timeout-minutes: 10
     outputs:
@@ -254,7 +259,7 @@ jobs:
           fi
 
   validation:
-    if: ${{ github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && (inputs.validation_case != '' || inputs.validation_tools_only || inputs.validation_native_tools || inputs.validation_shared_tools || inputs.validation_manager_tools || inputs.validation_cleanup_run_id != '') }}
+    if: ${{ github.event_name == 'workflow_dispatch' && !inputs.validation_runner_inspect_only && inputs.validation_retain_evidence_run_id == '' && inputs.validation_package_sha == '' && (inputs.validation_case != '' || inputs.validation_tools_only || inputs.validation_native_tools || inputs.validation_shared_tools || inputs.validation_manager_tools || inputs.validation_cleanup_run_id != '' || inputs.validation_cleanup_retained_evidence) }}
     runs-on: ${{ inputs.validation_tools_only && 'ubuntu-latest' || 'smoke-testing' }}
     timeout-minutes: 90
     outputs:
@@ -290,6 +295,7 @@ jobs:
           BUILD_RUN_ATTEMPT: ${{ inputs.build_run_attempt }}
           AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
           CLEANUP_RUN_ID: ${{ inputs.validation_cleanup_run_id }}
+          CLEANUP_RETAINED_EVIDENCE: ${{ inputs.validation_cleanup_retained_evidence }}
           SELECTED_CASE: ${{ inputs.validation_case }}
           TOOLS_ONLY: ${{ inputs.validation_tools_only }}
           NATIVE_TOOLS: ${{ inputs.validation_native_tools }}
@@ -307,6 +313,8 @@ jobs:
 
           revision = os.environ["AUTO_TESTING_SHA"]
           cleanup = os.environ["CLEANUP_RUN_ID"]
+          if os.environ["CLEANUP_RETAINED_EVIDENCE"] == "true" and not cleanup:
+              raise ValueError("persistent recovery input requires one exact cleanup run")
           if os.environ["TOOLS_ONLY"] == "true" and (cleanup or os.environ["SELECTED_CASE"]):
               raise ValueError("tools-only execution cannot select a fleet case or cleanup")
           if os.environ["NATIVE_TOOLS"] == "true" and (os.environ["TOOLS_ONLY"] != "true" or cleanup or os.environ["SELECTED_CASE"]):
@@ -586,6 +594,7 @@ jobs:
           GH_TOKEN: ${{ github.token }}
           FAILED_RUN_ID: ${{ inputs.validation_cleanup_run_id }}
           AUTO_TESTING_SHA: ${{ inputs.auto_testing_sha }}
+          CLEANUP_RETAINED_EVIDENCE: ${{ inputs.validation_cleanup_retained_evidence }}
         run: |
           set -euo pipefail
           test "$(git -C auto-testing rev-parse HEAD)" = "${AUTO_TESTING_SHA}"
@@ -600,8 +609,13 @@ jobs:
               or run["workflowName"] != "RustFS Functional Chain" or run["headBranch"] != os.environ["GITHUB_REF_NAME"]):
               raise ValueError("cleanup source must be a completed failed or cancelled Functional Chain run")
           PY
-          gh run download "${FAILED_RUN_ID}" --repo "${GITHUB_REPOSITORY}" \
-            --pattern "rustfs-validation-${FAILED_RUN_ID}-*" --dir "${VALIDATION_ROOT}/failed-artifact"
+          if [ "${CLEANUP_RETAINED_EVIDENCE}" = true ]; then
+            python3 -B scripts/prepare_retained_validation_recovery.py \
+              --report "${VALIDATION_ROOT}/retained-recovery-input.json"
+          else
+            gh run download "${FAILED_RUN_ID}" --repo "${GITHUB_REPOSITORY}" \
+              --pattern "rustfs-validation-${FAILED_RUN_ID}-*" --dir "${VALIDATION_ROOT}/failed-artifact"
+          fi
           python3 -B auto-testing/scripts/validation_heal/recover_environment.py \
             --failed-run-id "${FAILED_RUN_ID}" --artifact "${VALIDATION_ROOT}/failed-artifact" \
             --candidate "${VALIDATION_ROOT}/candidate.json" --evidence "${VALIDATION_ROOT}/recovery"

scripts/prepare_retained_validation_recovery.py

@@ -0,0 +1,59 @@
+#!/usr/bin/env python3
+"""Copy verified local evidence into a recovery input without claiming an old artifact."""
+
+import argparse
+import json
+import os
+from pathlib import Path
+import re
+import sys
+
+from resolve_functional_candidate import REPOSITORY, api, require
+from retain_validation_evidence import inventory, prepare
+
+
+def stage(env, report):
+    source = env["FAILED_RUN_ID"]
+    require(re.fullmatch(r"[1-9][0-9]{0,19}", source), "exact failed run required")
+    run = api(f"repos/{REPOSITORY}/actions/runs/{source}")
+    require(run.get("head_branch") == env["GITHUB_REF_NAME"]
+            and type(run.get("run_attempt")) is int and run["run_attempt"] > 0,
+            "local recovery source branch or attempt differs")
+    root = prepare({**env, "EVIDENCE_RETAIN_INPUTS": json.dumps({
+        "validation_retain_evidence_run_id": source,
+        "validation_retain_evidence_run_attempt": str(run["run_attempt"])})}, report)
+    report.update(complete=False, stage="stable-recovery-copy")
+    destination = Path(env["VALIDATION_ROOT"]) / "failed-artifact"
+    require(destination.parent.is_absolute() and destination.parent.resolve(strict=True) == destination.parent,
+            "recovery evidence root is not canonical")
+    destination.mkdir(mode=0o700, exist_ok=False)
+    copied = destination / "retained-source-evidence"
+    files, size = inventory(root, copy_to=copied)
+    require(files == report["files"] and size == report["bytes"]
+            and inventory(copied) == (files, size), "retained evidence changed during recovery copy")
+    report.update(complete=True, stage="complete", recoveryInput=str(copied))
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--report", type=Path, required=True)
+    args = parser.parse_args()
+    report = {"schemaVersion": 1, "collectionRunId": os.environ["GITHUB_RUN_ID"],
+              "collectionAttempt": os.environ["GITHUB_RUN_ATTEMPT"], "complete": False,
+              "stage": "source-workflow",
+              "collectionKind": "runner-local-recovery-input-copy", "sourceArtifactClaimed": False,
+              "fleetAccessed": False, "originalEvidenceModified": False,
+              "sourceVerdictRewritten": False, "canonicalCasePassClaimed": False}
+    try:
+        stage(os.environ, report)
+    except Exception as error:
+        report.update(complete=False, errorType=type(error).__name__)
+    with args.report.open("x") as stream:
+        json.dump(report, stream, indent=2)
+        stream.write("\n")
+    print(json.dumps({key: report[key] for key in ("complete", "stage")}))
+    return 0 if report["complete"] else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/retain_validation_evidence.py

@@ -23,9 +23,13 @@ def stable(info):
             info.st_nlink, info.st_size, info.st_mtime_ns, info.st_ctime_ns)
 
 
-def inventory(root):
+def inventory(root, *, copy_to=None):
     files, total, entries = [], 0, 0
     device = root.lstat().st_dev
+    if copy_to is not None:
+        require(copy_to.is_absolute() and copy_to.parent.resolve(strict=True) == copy_to.parent,
+                "evidence copy parent is not canonical")
+        copy_to.mkdir(mode=0o700, exist_ok=False)
 
     def visit(fd, prefix):
         nonlocal total, entries
@@ -41,6 +45,8 @@ def inventory(root):
             path = prefix + name
             require(info.st_uid == os.getuid() and info.st_dev == device, "evidence entry has a different owner or device")
             if stat.S_ISDIR(info.st_mode):
+                if copy_to is not None:
+                    (copy_to / path).mkdir(mode=0o700, exist_ok=False)
                 child = os.open(name, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, dir_fd=fd)
                 try:
                     require(stable(os.fstat(child)) == stable(info), "evidence directory changed before inspection")
@@ -52,17 +58,28 @@ def inventory(root):
                         and len(files) < MAX_FILES and total + info.st_size <= MAX_BYTES,
                         "evidence contains links, special files or exceeds its byte/file budget")
                 child = os.open(name, os.O_RDONLY | os.O_NOFOLLOW | os.O_NONBLOCK, dir_fd=fd)
+                target = None
                 try:
                     require(stable(os.fstat(child)) == stable(info), "evidence file changed before inspection")
+                    if copy_to is not None:
+                        target = os.open(copy_to / path, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
                     digest, size = hashlib.sha256(), 0
                     while chunk := os.read(child, 1024**2):
                         size += len(chunk)
                         require(size <= info.st_size, "evidence file grew during inspection")
                         digest.update(chunk)
+                        if target is not None:
+                            remaining = memoryview(chunk)
+                            while remaining:
+                                written = os.write(target, remaining)
+                                require(written > 0, "evidence copy made no progress")
+                                remaining = remaining[written:]
                     require(size == info.st_size and stable(os.fstat(child)) == stable(info),
                             "evidence file changed during inspection")
                 finally:
                     os.close(child)
+                    if target is not None:
+                        os.close(target)
                 total += size
                 files.append({"path": path, "size": size, "sha256": digest.hexdigest()})
             require(stable(os.stat(name, dir_fd=fd, follow_symlinks=False)) == stable(info),

scripts/test_retain_validation_evidence.py

@@ -9,6 +9,7 @@ import unittest
 from unittest.mock import patch
 
 import retain_validation_evidence as retain
+import prepare_retained_validation_recovery as recovery_input
 
 
 class RetainEvidenceTests(unittest.TestCase):
@@ -131,6 +132,58 @@ class RetainEvidenceTests(unittest.TestCase):
             self.assertEqual(value["errorType"], "FileNotFoundError")
             self.assertFalse(Path(env["GITHUB_ENV"]).exists())
 
+    def test_recovery_copy_keeps_original_bytes_and_separate_collection_identity(self):
+        with tempfile.TemporaryDirectory() as directory:
+            env, root, run, jobs = self.fixture(directory)
+            destination = Path(directory).resolve() / 'new-recovery'; destination.mkdir(mode=0o700)
+            env.update(FAILED_RUN_ID='77', GITHUB_REF_NAME='topic', VALIDATION_ROOT=str(destination))
+            run['head_branch'] = 'topic'
+            nested = root / 'nested'; nested.mkdir(); (nested / 'receipt').write_bytes(b'ACK')
+            before = retain.inventory(root)
+            report = {'collectionRunId': '88', 'sourceArtifactClaimed': False}
+            with patch.object(recovery_input, 'api', return_value=run), patch.object(retain, 'api', side_effect=[run, jobs]):
+                recovery_input.stage(env, report)
+            self.assertTrue(report['complete'])
+            self.assertEqual(report['collectionRunId'], '88')
+            self.assertEqual(report['sourceRun']['id'], 77)
+            self.assertIs(report['sourceArtifactClaimed'], False)
+            copied = Path(report['recoveryInput'])
+            self.assertEqual(retain.inventory(copied), before)
+            self.assertEqual(retain.inventory(root), before)
+            self.assertNotEqual((copied / 'original.json').stat().st_ino, (root / 'original.json').stat().st_ino)
+
+    def test_recovery_copy_refuses_changed_source_and_foreign_branch(self):
+        for mutation in ('drift', 'branch'):
+            with self.subTest(mutation=mutation), tempfile.TemporaryDirectory() as directory:
+                env, root, run, jobs = self.fixture(directory)
+                destination = Path(directory).resolve() / 'recovery'; destination.mkdir(mode=0o700)
+                env.update(FAILED_RUN_ID='77', GITHUB_REF_NAME='topic', VALIDATION_ROOT=str(destination))
+                run['head_branch'] = 'foreign' if mutation == 'branch' else 'topic'
+                original = retain.prepare
+                def changed(environment, report):
+                    source = original(environment, report)
+                    (source / 'original.json').write_bytes(b'new bytes after admission')
+                    return source
+                with patch.object(recovery_input, 'api', return_value=run), \
+                     patch.object(retain, 'api', side_effect=[run, jobs]), \
+                     patch.object(recovery_input, 'prepare', side_effect=changed), self.assertRaises(ValueError):
+                    recovery_input.stage(env, {})
+
+    def test_copy_does_not_follow_a_symlink_swapped_after_directory_listing(self):
+        with tempfile.TemporaryDirectory() as directory:
+            env, root, run, jobs = self.fixture(directory)
+            secret = Path(directory) / 'private'; secret.write_bytes(b'outside-root-secret')
+            destination = Path(directory).resolve() / 'copy'
+            original_stat = os.stat
+            def swapped(path, *args, **kwargs):
+                info = original_stat(path, *args, **kwargs)
+                if path == 'original.json' and kwargs.get('follow_symlinks') is False:
+                    (root / path).unlink(); (root / path).symlink_to(secret)
+                return info
+            with patch.object(retain.os, 'stat', side_effect=swapped), self.assertRaises(OSError):
+                retain.inventory(root, copy_to=destination)
+            self.assertFalse((destination / 'original.json').exists())
+
 
 if __name__ == "__main__":
     unittest.main()