Fix/correctly handle terraform s3 backend with versioned bucket (#1686)

Co-authored-by: loverustfs <hello@rustfs.com>

crates/e2e_test/src/version_id_regression_test.rs

@@ -395,4 +395,46 @@ mod tests {
 
         info!("✅ PASSED: Veeam backup workflow simulation completed successfully");
     }
+
+    #[tokio::test]
+    #[serial]
+    async fn test_terraform_put_after_delete() -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
+        init_logging();
+
+        let mut env = RustFSTestEnvironment::new().await.expect("Failed to create test environment");
+        env.start_rustfs_server(vec![]).await.expect("Failed to start RustFS");
+
+        // Use a versioned bucket for this test
+        let bucket = "terraform";
+
+        let client = env.create_s3_client();
+        env.create_test_bucket(bucket).await?;
+
+        let key = "terraform.tfstate";
+        let response = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from(b"v1".to_vec()))
+            .send()
+            .await;
+        assert!(response.is_ok());
+
+        client.delete_object().bucket(bucket).key(key).send().await?;
+
+        let response = client
+            .put_object()
+            .bucket(bucket)
+            .key(key)
+            .body(ByteStream::from(b"v1".to_vec()))
+            .send()
+            .await;
+
+        assert!(response.is_ok());
+
+        let get_response = client.get_object().bucket(bucket).key(key).send().await;
+        assert!(get_response.is_ok(), "Object should exist after PUT");
+
+        Ok(())
+    }
 }

crates/ecstore/src/set_disk.rs

@@ -3583,6 +3583,10 @@ impl SetDisks {
 
         match oi {
             Ok(oi) => {
+                // If top level is a delete marker proceed to upload.
+                if oi.delete_marker {
+                    return None;
+                }
                 if should_prevent_write(&oi, http_preconditions.if_none_match, http_preconditions.if_match) {
                     return Some(StorageError::PreconditionFailed);
                 }

rustfs/src/profiling/allocator.rs

@@ -28,6 +28,10 @@ use std::path::Path;
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
 use std::sync::{Arc, LazyLock, Weak};
 
+type AllocatorShardedMap = ShardedHashMap<usize, (usize, Arc<Vec<usize>>)>;
+type LazyAllocatorShardedMap = LazyLock<AllocatorShardedMap>;
+
+type AllocatorSampleHashMap = HashMap<*const Vec<usize>, (i64, i64, Arc<Vec<usize>>)>;
 /// A wrapper around a GlobalAlloc that samples allocations and records stack traces.
 pub struct TracingAllocator<A: GlobalAlloc> {
     inner: A,
@@ -45,7 +49,7 @@ static ENABLED: AtomicBool = AtomicBool::new(false);
 // Global storage for profile data
 // Map: Address (usize) -> (Size (usize), StackTrace (Arc<Vec<usize>>))
 // We store the Arc to keep the stack trace alive as long as the allocation is live.
-static LIVE_ALLOCATIONS: LazyLock<ShardedHashMap<usize, (usize, Arc<Vec<usize>>)>> = LazyLock::new(|| ShardedHashMap::new(64));
+static LIVE_ALLOCATIONS: LazyAllocatorShardedMap = LazyLock::new(|| ShardedHashMap::new(64));
 
 // Cache for deduplicating stack traces.
 // Map: StackHash (u64) -> Weak<Vec<usize>>
@@ -216,7 +220,7 @@ fn dump_profile_inner(path: &Path) -> Result<(), String> {
     // Collect samples
     // Aggregate by Stack Trace Pointer (deduplication via Arc pointer)
     // Map: Arc pointer -> (Count, Bytes, Arc<Vec<usize>>)
-    let mut aggregated_samples: HashMap<*const Vec<usize>, (i64, i64, Arc<Vec<usize>>)> = HashMap::new();
+    let mut aggregated_samples: AllocatorSampleHashMap = HashMap::new();
 
     // Step 1: Collect data from LIVE_ALLOCATIONS while holding the lock (implicitly via iter)
     // We do NOT perform symbol resolution here to avoid deadlocks.
@@ -233,8 +237,10 @@ fn dump_profile_inner(path: &Path) -> Result<(), String> {
 
     // Step 2: Process samples and resolve symbols (outside of LIVE_ALLOCATIONS lock)
     for (_key, (count, bytes, frames)) in aggregated_samples {
-        let mut sample = pb::Sample::default();
-        sample.value = vec![count, bytes];
+        let mut sample = pb::Sample {
+            value: vec![count, bytes],
+            ..Default::default()
+        };
 
         // Process frames
         for &addr in frames.iter() {