fix(filemeta): keep data dir of a version awaiting purge replication

crates/e2e_test/src/replication_extension_test.rs

@@ -3837,6 +3837,157 @@ 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(())
+}
+
+/// Regression for rustfs/backlog#2340 (not Wasabi specific): permanently
+/// deleting a version whose payload lives in a data dir must leave the source
+/// clean once the purge replicates. Managed-SSE objects are never inlined and a
+/// plain object above the inline threshold takes the same layout. The version
+/// retained with a pending purge used to lose its data dir, so the purge state
+/// could never be applied (`VersionNotFound` on every retry) and the bucket
+/// stayed `BucketNotEmpty` while `ListObjectVersions` was already empty.
+#[tokio::test]
+async fn test_bucket_replication_version_purge_of_non_inline_object_releases_source_bucket() -> TestResult {
+    init_logging();
+
+    let (source_env, target_env, source_bucket, target_bucket) = build_sse_replication_pair("purge-datadir", true, true).await?;
+    let target_arn = wait_for_remote_target_arn(&source_env, &source_bucket).await?;
+    put_bucket_replication_with_delete_statuses(&source_env, &source_bucket, &target_arn, "Enabled", Some("Enabled")).await?;
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    let sse_key = "sse-object.bin";
+    let large_key = "large-object.bin";
+    let sse_put = source_client
+        .put_object()
+        .bucket(&source_bucket)
+        .key(sse_key)
+        .body(ByteStream::from_static(b"encrypted source payload"))
+        .server_side_encryption(ServerSideEncryption::Aes256)
+        .send()
+        .await?;
+    let large_put = source_client
+        .put_object()
+        .bucket(&source_bucket)
+        .key(large_key)
+        .body(ByteStream::from(vec![0x5a; 2 * 1024 * 1024]))
+        .send()
+        .await?;
+    let purged = [
+        (sse_key, sse_put.version_id().ok_or("SSE PUT omitted version ID")?.to_string()),
+        (large_key, large_put.version_id().ok_or("large PUT omitted version ID")?.to_string()),
+    ];
+    assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
+
+    for (key, version_id) in &purged {
+        source_client
+            .delete_object()
+            .bucket(&source_bucket)
+            .key(*key)
+            .version_id(version_id)
+            .send()
+            .await?;
+    }
+    assert_replication_converged(&source_client, &source_bucket, &target_client, &target_bucket).await?;
+    let target_state = list_replication_state(&target_client, &target_bucket).await?;
+    assert!(target_state.is_empty(), "target retained an explicitly purged version: {target_state:?}");
+
+    // The purge state is applied on the source asynchronously after the target
+    // acknowledges the delete; only then does the retained version go away and
+    // the bucket become deletable. A listing that is empty while DeleteBucket
+    // keeps answering BucketNotEmpty is exactly the regression.
+    let deadline = tokio::time::Instant::now() + Duration::from_secs(60);
+    loop {
+        let listing = source_client.list_object_versions().bucket(&source_bucket).send().await?;
+        let listed = listing.versions().len() + listing.delete_markers().len();
+        match source_client.delete_bucket().bucket(&source_bucket).send().await {
+            Ok(_) => break,
+            Err(err) if err.code() == Some("BucketNotEmpty") => {
+                if tokio::time::Instant::now() >= deadline {
+                    return Err(format!(
+                        "source bucket stayed BucketNotEmpty after the version purge replicated; \
+                         ListObjectVersions shows {listed} entries"
+                    )
+                    .into());
+                }
+                sleep(Duration::from_millis(500)).await;
+            }
+            Err(err) => return Err(err.into()),
+        }
+    }
+
+    Ok(())
+}
+
 #[tokio::test]
 async fn test_bucket_replication_disabled_delete_marker_does_not_propagate() -> TestResult {
     init_logging();

crates/filemeta/src/filemeta.rs

@@ -692,10 +692,15 @@ impl FileMeta {
                             }
                         }
 
-                        let old_dir = v.object.as_ref().map(|v| v.data_dir).unwrap_or_default();
+                        // The version stays on disk while the purge replicates
+                        // (status PENDING/FAILED); its data dir must stay with
+                        // it. Returning the dir here made the disk layer delete
+                        // it, which turned every non-inline retained version
+                        // into an unreadable zombie: the purge state could never
+                        // be applied and the bucket could never be deleted.
                         self.set_idx(i, v)?;
 
-                        return Ok(old_dir);
+                        return Ok(None);
                     }
                     found_index = Some(i);
                 }
@@ -2702,6 +2707,58 @@ mod test {
         );
     }
 
+    /// Regression for rustfs/backlog#2340: a version purge that still awaits
+    /// the replication target keeps the object version on disk with a pending
+    /// purge status. Its data dir must be retained with it; handing the dir
+    /// back here made the disk layer delete it, leaving every non-inline
+    /// retained version unreadable. The dir is released only once the purge
+    /// completes and the version itself goes away.
+    #[test]
+    fn delete_version_pending_version_purge_retains_object_data_dir() {
+        let version_id = Uuid::new_v4();
+        let data_dir = Uuid::new_v4();
+        let mut fm = FileMeta::new();
+        let mut fi = FileInfo::new("object", 2, 2);
+        fi.version_id = Some(version_id);
+        fi.data_dir = Some(data_dir);
+        fi.mod_time = Some(OffsetDateTime::now_utc());
+        fm.add_version(fi).unwrap();
+
+        let pending_purge = FileInfo {
+            name: "object".to_string(),
+            version_id: Some(version_id),
+            mark_deleted: true,
+            replication_state_internal: Some(ReplicationState {
+                version_purge_status_internal: Some("target=PENDING;".to_string()),
+                purge_targets: version_purge_statuses_map("target=PENDING;"),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let freed = fm.delete_version(&pending_purge).unwrap();
+        assert_eq!(freed, None, "a pending purge must not release the retained version's data dir");
+        assert_eq!(fm.versions.len(), 1, "the version must stay until the purge replicates");
+        let retained = fm
+            .into_fileinfo("vol", "object", &version_id.to_string(), false, false, true)
+            .unwrap();
+        assert_eq!(retained.data_dir, Some(data_dir));
+        assert_eq!(retained.version_purge_status(), VersionPurgeStatusType::Pending);
+
+        let completed_purge = FileInfo {
+            name: "object".to_string(),
+            version_id: Some(version_id),
+            replication_state_internal: Some(ReplicationState {
+                version_purge_status_internal: Some("target=COMPLETE;".to_string()),
+                purge_targets: version_purge_statuses_map("target=COMPLETE;"),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        let freed = fm.delete_version(&completed_purge).unwrap();
+        assert_eq!(freed, Some(data_dir), "a completed purge removes the version and releases its data dir");
+        assert!(fm.versions.is_empty());
+    }
+
     #[test]
     fn delete_version_accepts_delete_only_marker_and_free_version_paths() {
         let marker_version_id = Uuid::new_v4();

crates/heal/src/heal/mrf_queue/snapshot.rs

@@ -123,13 +123,6 @@ pub struct CommittedSnapshot {
     payload: Vec<u8>,
 }
 
-#[derive(Default)]
-struct SnapshotReadStats {
-    file_reads: usize,
-    bytes_read: usize,
-    peak_file_bytes: usize,
-}
-
 impl CommittedSnapshot {
     /// Persistent single-writer sequence, not a process UUID ordering.
     pub fn sequence(&self) -> u64 {
@@ -169,15 +162,6 @@ pub enum RecoverySnapshot {
 }
 
 async fn read_bounded(disk: &EcstoreDiskStore, path: &str, limit: usize) -> Result<Option<Vec<u8>>, SnapshotError> {
-    read_bounded_with_stats(disk, path, limit, None).await
-}
-
-async fn read_bounded_with_stats(
-    disk: &EcstoreDiskStore,
-    path: &str,
-    limit: usize,
-    mut stats: Option<&mut SnapshotReadStats>,
-) -> Result<Option<Vec<u8>>, SnapshotError> {
     let reader = match EcstoreDiskAPI::read_file(disk.as_ref(), RUSTFS_META_BUCKET, path).await {
         Ok(reader) => reader,
         Err(EcstoreDiskError::FileNotFound | EcstoreDiskError::VolumeNotFound) => return Ok(None),
@@ -194,11 +178,6 @@ async fn read_bounded_with_stats(
     if bytes.len() > limit {
         return Err(SnapshotError::TooLarge);
     }
-    if let Some(stats) = stats.as_mut() {
-        stats.file_reads += 1;
-        stats.bytes_read = stats.bytes_read.checked_add(bytes.len()).ok_or(SnapshotError::TooLarge)?;
-        stats.peak_file_bytes = stats.peak_file_bytes.max(bytes.len());
-    }
     Ok(Some(bytes))
 }
 
@@ -218,26 +197,17 @@ fn select_snapshot(selected: &mut Option<CommittedSnapshot>, candidate: Committe
 }
 
 async fn read_committed(disks: &[EcstoreDiskStore], limit: usize) -> Result<Option<CommittedSnapshot>, SnapshotError> {
-    read_committed_with_stats(disks, limit, None).await
-}
-
-async fn read_committed_with_stats(
-    disks: &[EcstoreDiskStore],
-    limit: usize,
-    mut stats: Option<&mut SnapshotReadStats>,
-) -> Result<Option<CommittedSnapshot>, SnapshotError> {
     let mut selected = None;
     let mut damaged = None;
     let mut identities = HashMap::new();
     for disk in disks {
         for (manifest_path, payload_path) in MANIFEST_PATHS.into_iter().zip(PAYLOAD_PATHS) {
             let candidate = async {
-                let Some(manifest) = read_bounded_with_stats(disk, manifest_path, MANIFEST_LEN, stats.as_deref_mut()).await?
-                else {
+                let Some(manifest) = read_bounded(disk, manifest_path, MANIFEST_LEN).await? else {
                     return Ok(None);
                 };
                 let header = Manifest::decode(&manifest, limit)?;
-                let payload = read_bounded_with_stats(disk, payload_path, header.payload_len, stats.as_deref_mut())
+                let payload = read_bounded(disk, payload_path, header.payload_len)
                     .await?
                     .ok_or(SnapshotError::Corrupt)?;
                 CommittedSnapshot::decode(&manifest, payload, limit).map(Some)
@@ -522,69 +492,6 @@ mod tests {
         assert_eq!(recovered.manifest.sequence, 1);
     }
 
-    #[tokio::test]
-    async fn committed_reader_reopens_previous_anchor_across_publication_boundaries() {
-        let owner = Uuid::new_v4();
-        let old = payload("old");
-        let next = payload("next");
-        let boundaries = [
-            ("payload-only", next.clone(), None),
-            ("torn-manifest", next.clone(), Some(manifest(owner, 2, &next)[..20].to_vec())),
-            ("stale-payload", old.clone(), Some(manifest(owner, 2, &next))),
-        ];
-        for (case, successor_payload, successor_manifest) in boundaries {
-            let root = TempDir::new().expect("test directory");
-            let store = disk(&root, "disk").await;
-            commit(&store, 0, owner, 1, &old).await;
-            install(&store, PAYLOAD_PATHS[1], &successor_payload).await;
-            if let Some(manifest) = &successor_manifest {
-                install(&store, MANIFEST_PATHS[1], manifest).await;
-            }
-
-            let reopened = disk(&root, "disk").await;
-            let recovered = read_committed(std::slice::from_ref(&reopened), 4096)
-                .await
-                .unwrap_or_else(|error| panic!("{case}: old anchor must remain readable after reopen: {error:?}"))
-                .unwrap_or_else(|| panic!("{case}: previous committed anchor missing after reopen"));
-            assert_eq!(recovered.manifest.sequence, 1, "{case}: successor must not become authoritative");
-            assert_eq!(recovered.payload, old, "{case}: previous payload must survive");
-            assert_eq!(
-                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[0])
-                    .await
-                    .expect("old payload retained")
-                    .as_ref(),
-                old.as_slice(),
-                "{case}: previous payload bytes changed"
-            );
-            assert_eq!(
-                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[0])
-                    .await
-                    .expect("old manifest retained")
-                    .as_ref(),
-                manifest(owner, 1, &old).as_slice(),
-                "{case}: previous manifest bytes changed"
-            );
-            assert_eq!(
-                EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, PAYLOAD_PATHS[1])
-                    .await
-                    .expect("successor payload retained")
-                    .as_ref(),
-                successor_payload.as_slice(),
-                "{case}: successor evidence changed"
-            );
-            if let Some(manifest) = &successor_manifest {
-                assert_eq!(
-                    EcstoreDiskAPI::read_all(reopened.as_ref(), RUSTFS_META_BUCKET, MANIFEST_PATHS[1])
-                        .await
-                        .expect("successor manifest retained")
-                        .as_ref(),
-                    manifest.as_slice(),
-                    "{case}: successor manifest evidence changed"
-                );
-            }
-        }
-    }
-
     #[tokio::test]
     async fn stale_manifest_cas_cannot_replace_committed_anchor() {
         let root = TempDir::new().expect("test directory");
@@ -664,35 +571,6 @@ mod tests {
         );
     }
 
-    #[tokio::test]
-    async fn committed_reader_resource_bounds_are_measured() {
-        let root = TempDir::new().expect("test directory");
-        let first = disk(&root, "first").await;
-        let second = disk(&root, "second").await;
-        let owner = Uuid::new_v4();
-        let old = [payload("old-0"), payload("old-1")].concat();
-        let new = [payload("new-0"), payload("new-1"), payload("new-2")].concat();
-        commit(&first, 0, owner, 1, &old).await;
-        commit(&second, 1, owner, 2, &new).await;
-
-        let mut stats = SnapshotReadStats::default();
-        let recovered = read_committed_with_stats(&[first, second], 4096, Some(&mut stats))
-            .await
-            .expect("read committed replicas")
-            .expect("committed snapshot");
-
-        assert_eq!(recovered.sequence(), 2);
-        assert_eq!(recovered.payload(), new.as_slice());
-        assert_eq!(recovered.manifest.payload_len, new.len());
-        assert_eq!(stats.file_reads, 4, "only committed manifests and their payloads are materialized");
-        assert_eq!(stats.bytes_read, (MANIFEST_LEN * 2) + old.len() + new.len());
-        assert_eq!(
-            stats.peak_file_bytes,
-            new.len().max(MANIFEST_LEN),
-            "reader peak allocation remains bounded by one manifest or payload file"
-        );
-    }
-
     #[tokio::test]
     async fn legacy_inspection_rejects_complete_subsets_and_scope_ambiguity() {
         let scoped = |set_index| {

docs/architecture/s3-compatibility-matrix.md

@@ -62,6 +62,7 @@ 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