test(scanner): cover reset cleanup process crash boundaries

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();

crates/scanner/src/scanner/tests.rs

@@ -27,6 +27,7 @@ use crate::{
 use serial_test::serial;
 use std::collections::{HashMap, HashSet};
 use std::io::Cursor;
+use std::path::Path;
 use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
 use std::task::Poll;
 use temp_env::{with_var, with_var_unset};
@@ -55,11 +56,17 @@ async fn setup_scanner_cycle_store_with_pool_count(
 ) -> (tempfile::TempDir, Arc<ECStore>) {
     init_ecstore_config_for_scanner_tests();
     let temp_dir = tempfile::tempdir().expect("scanner cycle test directory should be created");
+    let store = setup_scanner_cycle_store_at_path(temp_dir.path(), seed_usage_baseline, pool_count).await;
+    (temp_dir, store)
+}
+
+async fn setup_scanner_cycle_store_at_path(root: &Path, seed_usage_baseline: bool, pool_count: usize) -> Arc<ECStore> {
+    init_ecstore_config_for_scanner_tests();
     let mut pools = Vec::with_capacity(pool_count);
     for pool_index in 0..pool_count {
         let mut endpoints = Vec::new();
         for disk_index in 0..4 {
-            let disk_path = temp_dir.path().join(format!("pool{pool_index}/disk{disk_index}"));
+            let disk_path = root.join(format!("pool{pool_index}/disk{disk_index}"));
             tokio::fs::create_dir_all(&disk_path)
                 .await
                 .expect("scanner cycle test disk should be created");
@@ -109,7 +116,7 @@ async fn setup_scanner_cycle_store_with_pool_count(
         .expect("scanner cycle usage baseline should persist");
     }
 
-    (temp_dir, store)
+    store
 }
 
 async fn restart_scanner_cycle_store_from(store: &Arc<ECStore>) -> Arc<ECStore> {

crates/scanner/src/scanner/tests/recovery_control.rs

@@ -117,6 +117,92 @@ async fn assert_reset_fences(store: &Arc<ECStore>) {
     ));
 }
 
+async fn assert_rebuilt_reset_fences(store: &Arc<ECStore>, expected_epoch: u64) {
+    let data = read_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str())
+        .await
+        .expect("rebuilt cycle remains durable");
+    let (cycle, epoch) = decode_scanner_cycle_state(&data).expect("valid rebuilt cycle");
+    assert_eq!((cycle.current, cycle.next, epoch), (0, 42, expected_epoch));
+    let usage: DataUsageInfo = serde_json::from_slice(
+        &read_config(store.clone(), DATA_USAGE_OBJ_NAME_PATH.as_str())
+            .await
+            .expect("durable rebuilt usage fence"),
+    )
+    .expect("valid usage");
+    assert_eq!(usage.scanner_epoch, Some(expected_epoch));
+    assert_eq!(usage.scanner_cycle, Some(41));
+    assert!(matches!(
+        read_config(store.clone(), DATA_USAGE_BLOOM_RECOVERY_PATH.as_str()).await,
+        Err(EcstoreError::ConfigNotFound)
+    ));
+}
+
+#[test]
+fn scanner_reset_crash_child_process_fixture() {
+    let Ok(root) = std::env::var("RUSTFS_SCANNER_RESET_CRASH_ROOT") else {
+        return;
+    };
+    let stage = match std::env::var("RUSTFS_SCANNER_RESET_CRASH_STAGE").as_deref() {
+        Ok("primary-read") => cleanup_io_fault::Stage::PrimaryRead,
+        Ok("primary-write") => cleanup_io_fault::Stage::PrimaryWrite,
+        Ok("usage-fence") => cleanup_io_fault::Stage::UsageFence,
+        other => panic!("unexpected reset crash stage: {other:?}"),
+    };
+    let runtime = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("child runtime should build");
+    runtime.block_on(async {
+        let store = setup_scanner_cycle_store_at_path(std::path::Path::new(&root), false, 1).await;
+        if stage == cleanup_io_fault::Stage::UsageFence {
+            save_config(store.clone(), DATA_USAGE_BLOOM_NAME_PATH.as_str(), b"corrupt-cycle".to_vec())
+                .await
+                .expect("force child through reconstruction branch");
+        }
+        let injection = cleanup_io_fault::install(&store, stage, false);
+        let error = resume_scanner_cycle_cleanup(CancellationToken::new(), store)
+            .await
+            .expect_err("child should stop at the injected owned I/O boundary");
+        assert!(injection.fired_while_owned());
+        let expected = match stage {
+            cleanup_io_fault::Stage::PrimaryRead => "injected primary read failure",
+            cleanup_io_fault::Stage::PrimaryWrite => "injected primary write failure",
+            cleanup_io_fault::Stage::UsageFence => "injected usage fence failure",
+        };
+        assert!(error.to_string().contains(expected), "{error}");
+    });
+    std::process::exit(77);
+}
+
+#[tokio::test]
+#[serial]
+async fn disabled_cleanup_recovers_after_child_process_crash_boundaries() {
+    for (name, expected_rebuilt_epoch) in [("primary-read", None), ("primary-write", None), ("usage-fence", Some(9))] {
+        let temp_dir = tempfile::tempdir().expect("crash fixture directory");
+        let store = setup_scanner_cycle_store_at_path(temp_dir.path(), false, 1).await;
+        seed_cleanup(&store, "cleanup-pending").await;
+        drop(store);
+
+        let status = std::process::Command::new(std::env::current_exe().expect("test binary path"))
+            .arg("scanner::tests::recovery_control::scanner_reset_crash_child_process_fixture")
+            .arg("--exact")
+            .arg("--nocapture")
+            .env("RUSTFS_SCANNER_RESET_CRASH_ROOT", temp_dir.path())
+            .env("RUSTFS_SCANNER_RESET_CRASH_STAGE", name)
+            .status()
+            .expect("child crash fixture should start");
+        assert_eq!(status.code(), Some(77), "{name} child did not reach the owned crash boundary");
+
+        let restarted = setup_scanner_cycle_store_at_path(temp_dir.path(), false, 1).await;
+        run_disabled_startup(CancellationToken::new(), restarted.clone()).await;
+        if let Some(epoch) = expected_rebuilt_epoch {
+            assert_rebuilt_reset_fences(&restarted, epoch).await;
+        } else {
+            assert_reset_fences(&restarted).await;
+        }
+    }
+}
+
 #[tokio::test]
 #[serial]
 async fn disabled_cleanup_reopens_persisted_intent_without_starting_scanner() {

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