test(scanner): collect heal evidence oracles from registry

Co-Authored-By: heihutu <heihutu@gmail.com>

Co-Authored-By: zhi22915 <qiuzgang@gmail.com>

crates/e2e_test/src/replication_extension_test.rs

@@ -3837,77 +3837,6 @@ async fn test_bucket_replication_converges_delete_marker_and_version_purge() ->
     Ok(())
 }
 
-/// Regression for rustfs/backlog#2340 (not Wasabi specific): a directory
-/// marker (`prefix/` with a body) in a versioned bucket is stored as the null
-/// version, like MinIO (`putOpts`: "for directory objects skip creating new
-/// versions"), and must still replicate to completion instead of staying
-/// `PENDING`.
-#[tokio::test]
-async fn test_bucket_replication_replicates_directory_marker_in_versioned_bucket() -> TestResult {
-    init_logging();
-
-    let mut source_env = RustFSTestEnvironment::new().await?;
-    let mut source_env_vars = replication_fast_env();
-    source_env_vars.extend_from_slice(LOOPBACK_REPLICATION_TARGET_ENV);
-    source_env.start_rustfs_server_with_env(vec![], &source_env_vars).await?;
-
-    let mut target_env = RustFSTestEnvironment::new().await?;
-    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
-
-    let source_bucket = "replication-dir-marker-src";
-    let target_bucket = "replication-dir-marker-dst";
-    let source_client = source_env.create_s3_client();
-    let target_client = target_env.create_s3_client();
-
-    source_client.create_bucket().bucket(source_bucket).send().await?;
-    target_client.create_bucket().bucket(target_bucket).send().await?;
-    enable_bucket_versioning(&source_env, source_bucket).await?;
-    enable_bucket_versioning(&target_env, target_bucket).await?;
-    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
-    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
-
-    let marker_key = "dir/trailing/";
-    let body = b"directory marker body";
-    let put = source_client
-        .put_object()
-        .bucket(source_bucket)
-        .key(marker_key)
-        .body(ByteStream::from_static(body))
-        .send()
-        .await?;
-    assert!(
-        put.version_id()
-            .is_none_or(|id| id == "null" || id == uuid::Uuid::nil().to_string()),
-        "a directory marker is the null version even in a versioned bucket: {:?}",
-        put.version_id()
-    );
-
-    wait_for_source_replication_status(&source_client, source_bucket, marker_key, "COMPLETED", false).await?;
-
-    let replica = target_client
-        .get_object()
-        .bucket(target_bucket)
-        .key(marker_key)
-        .send()
-        .await?;
-    assert_eq!(replica.body.collect().await?.into_bytes().as_ref(), body);
-    let listed = target_client
-        .list_object_versions()
-        .bucket(target_bucket)
-        .prefix(marker_key)
-        .send()
-        .await?;
-    let marker_versions: Vec<_> = listed.versions().iter().filter(|v| v.key() == Some(marker_key)).collect();
-    assert_eq!(marker_versions.len(), 1, "the marker must land exactly once: {marker_versions:?}");
-    assert_eq!(
-        marker_versions[0].version_id(),
-        Some("null"),
-        "the replica keeps the null version identity"
-    );
-
-    Ok(())
-}
-
 #[tokio::test]
 async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
     init_logging();

docs/architecture/s3-compatibility-matrix.md

@@ -62,7 +62,6 @@ Object keys are stored as file-system paths under each drive (`{drive}/{bucket}/
 | Behavior | RustFS | AWS S3 | Why |
 |---|---|---|---|
 | Object key with a `.` or `..` path segment, or an empty segment (`//`), such as `a//b/./c/../d` | `400 InvalidArgument` (`check_object_args` in `crates/ecstore/src/bucket/utils.rs`, mirroring MinIO `IsValidObjectPrefix`) | Accepted as an opaque key | A `..` segment would resolve to a parent directory and `.`/`//` segments would alias other keys on disk; encoding them would change the MinIO-compatible on-disk format. |
-| Directory marker (key ending in `/`, with or without a body) in a versioned bucket | Stored as the null version: `PutObject`/`HeadObject` report version id `00000000-0000-0000-0000-000000000000`, `ListObjectVersions` reports `null`, and a later PUT of the same key overwrites in place (`put_opts` in `rustfs/src/storage/options.rs`, mirroring MinIO `putOpts`: "for directory objects skip creating new versions") | A real version id per PUT, with a version history | The marker only exists to make an empty prefix listable; keeping a history for it would leave hidden versions behind every prefix delete. Replication still copies the marker as its null version (`test_bucket_replication_replicates_directory_marker_in_versioned_bucket` in `crates/e2e_test/src/replication_extension_test.rs`). |
 
 ## Update Rule
 

scripts/check_test_wiring.py

@@ -882,6 +882,22 @@ def evidence_integer(value: object, name: str, minimum: int, maximum: int) -> in
     return value
 
 
+def scanner_heal_oracle_names(root: Path) -> tuple[str, ...]:
+    registry = read_json(root / ".config/scanner-heal-required-tests.json")
+    evidence_integer(registry.get("schema"), "registry schema", 1, 1)
+    cases = registry.get("cases")
+    require(isinstance(cases, dict) and cases, "invalid scanner/heal registry")
+    names = set()
+    for case_id, requirement in cases.items():
+        require(isinstance(case_id, str) and case_id, "invalid scanner/heal case identity")
+        oracle = requirement.get("oracle")
+        require(isinstance(oracle, str) and oracle.endswith(".json"), f"invalid oracle for {case_id}")
+        path = Path(oracle)
+        require(not path.is_absolute() and ".." not in path.parts, f"oracle path escapes run directory for {case_id}")
+        names.add(oracle)
+    return tuple(sorted(names))
+
+
 def begin_scanner_heal_receipt(root: Path, directory: Path, binary: Path, test_binary: Path) -> None:
     """Record an existing build; this command never builds or runs a test."""
     require(not directory.exists(), "scanner/heal run directory must be new")
@@ -914,18 +930,28 @@ def begin_scanner_heal_receipt(root: Path, directory: Path, binary: Path, test_b
                                        "started_at": datetime.now(timezone.utc).timestamp(), **builds})
 
 
-def finish_scanner_heal_receipt(directory: Path, exit_code: int) -> None:
+def finish_scanner_heal_receipt(directory: Path, exit_code: int, root: Path = ROOT) -> None:
     require(type(exit_code) is int and 0 <= exit_code <= 255, "invalid test exit code")
     require(not (directory / "execution.json").exists(), "execution receipt already exists")
     run = read_json(directory / "run.json")
     artifacts = {}
-    for name in ("listing.json", "junit.xml", "background-target-restart.json"):
+    oracle_names = scanner_heal_oracle_names(root)
+    for name in ("listing.json", "junit.xml"):
         path = directory / name
         if exit_code != 0 and not path.exists():
             continue
         require(path.is_file() and 0 < path.stat().st_size <= MAX_JSON_BYTES, f"missing/oversized {name}")
         require(path.stat().st_mtime >= run["started_at"], f"stale {name}")
         artifacts[name] = digest(path)
+    for name in oracle_names:
+        path = directory / name
+        if not path.exists():
+            continue
+        require(path.is_file() and 0 < path.stat().st_size <= MAX_JSON_BYTES, f"missing/oversized {name}")
+        require(path.stat().st_mtime >= run["started_at"], f"stale {name}")
+        artifacts[name] = digest(path)
+    if exit_code == 0:
+        require(any(name in artifacts for name in oracle_names), "missing scanner/heal case oracle")
     write_json(directory / "execution.json", {"run_id": run["run_id"], "exit_code": exit_code,
                                              "finished_at": datetime.now(timezone.utc).timestamp(),
                                              "artifacts": artifacts})
@@ -1251,7 +1277,7 @@ class SelfTests(unittest.TestCase):
             "topology": {"nodes": 4, "drives_per_node": 1}, "pid_before": 10, "pid_after": 11,
             "objects": objects, "node_listings": [[item["key"] for item in objects]] * 4,
         })
-        finish_scanner_heal_receipt(run_dir, 0)
+        finish_scanner_heal_receipt(run_dir, 0, root)
         return root, run_dir
 
     def test_scanner_heal_case_does_not_approve_pending_release(self) -> None:
@@ -1264,6 +1290,25 @@ class SelfTests(unittest.TestCase):
             self.assertTrue(any(error.startswith("pending R-D:") for error in errors))
             self.assertTrue(any(error.startswith("pending R-L:") for error in errors))
 
+    def test_scanner_heal_finish_collects_oracles_from_registry(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root, run_dir = self.scanner_heal_fixture(Path(tmp))
+            registry = read_json(root / ".config/scanner-heal-required-tests.json")
+            alternate = dict(registry["cases"]["background-target-restart"])
+            alternate["oracle"] = "alternate-target-restart.json"
+            registry["cases"]["alternate-target-restart"] = alternate
+            write_json(root / ".config/scanner-heal-required-tests.json", registry)
+            oracle = read_json(run_dir / "background-target-restart.json")
+            oracle["case"] = "alternate-target-restart"
+            write_json(run_dir / "alternate-target-restart.json", oracle)
+            (run_dir / "background-target-restart.json").unlink()
+            (run_dir / "execution.json").unlink()
+
+            finish_scanner_heal_receipt(run_dir, 0, root)
+
+            self.assertIn("alternate-target-restart.json", read_json(run_dir / "execution.json")["artifacts"])
+            self.assertEqual(check_scanner_heal_evidence(root, run_dir, "alternate-target-restart"), [])
+
     def test_scanner_heal_rejects_broken_execution_and_artifacts(self) -> None:
         for fault in ("exit", "missing", "zero", "skipped", "failed", "retry", "filtered", "ignored", "stale",
                       "hash", "binary", "synthetic", "wrong-run", "same-pid", "body", "parts", "listing", "topology"):