fix(tier): sweep transitioned copies from delete handlers (#2287)

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -45,7 +45,7 @@ use lazy_static::lazy_static;
 use rustfs_common::data_usage::TierStats;
 use rustfs_common::heal_channel::rep_has_active_rules;
 use rustfs_common::metrics::{IlmAction, Metrics};
-use rustfs_filemeta::{NULL_VERSION_ID, RestoreStatusOps, is_restored_object_on_disk};
+use rustfs_filemeta::{FileInfo, NULL_VERSION_ID, RestoreStatusOps, is_restored_object_on_disk};
 use rustfs_s3_common::EventName;
 use rustfs_utils::{get_env_i64, get_env_usize, path::encode_dir_object, string::strings_has_prefix_fold};
 use s3s::Body;
@@ -393,12 +393,81 @@ impl ExpiryState {
                         //delete_object_versions(api, &v.bucket, &v.versions, v.event).await;
                     }
                     else if v.as_any().is::<Jentry>() {
-                        //transitionLogIf(es.ctx, deleteObjectFromRemoteTier(es.ctx, v.ObjName, v.VersionID, v.TierName))
+                        let v = v.as_any().downcast_ref::<Jentry>().expect("err!");
+                        if let Err(err) = delete_object_from_remote_tier(&v.obj_name, &v.version_id, &v.tier_name).await {
+                            warn!(
+                                object = %v.obj_name,
+                                version_id = %v.version_id,
+                                tier = %v.tier_name,
+                                error = ?err,
+                                "failed to delete transitioned object from remote tier"
+                            );
+                        }
                     }
                     else if v.as_any().is::<FreeVersionTask>() {
                         let v = v.as_any().downcast_ref::<FreeVersionTask>().expect("err!");
-                        let _oi = v.0.clone();
+                        let oi = v.0.clone();
+                        if let Err(err) = delete_object_from_remote_tier(
+                            &oi.transitioned_object.name,
+                            &oi.transitioned_object.version_id,
+                            &oi.transitioned_object.tier,
+                        )
+                        .await
+                        {
+                            warn!(
+                                bucket = %oi.bucket,
+                                object = %oi.name,
+                                remote_object = %oi.transitioned_object.name,
+                                remote_version_id = %oi.transitioned_object.version_id,
+                                tier = %oi.transitioned_object.tier,
+                                error = ?err,
+                                "failed to sweep transitioned free version from remote tier"
+                            );
+                            continue;
+                        }
 
+                        let mut fi = FileInfo {
+                            name: oi.name.clone(),
+                            version_id: oi.version_id,
+                            deleted: true,
+                            ..Default::default()
+                        };
+                        fi.set_tier_free_version();
+
+                        let mut deleted_locally = false;
+                        for pool in api.pools.iter() {
+                            let set = pool.get_disks_by_key(&oi.name);
+                            match set.delete_object_version(&oi.bucket, &oi.name, &fi, false).await {
+                                Ok(()) => {
+                                    deleted_locally = true;
+                                    break;
+                                }
+                                Err(err) if is_err_version_not_found(&err) || is_err_object_not_found(&err) => continue,
+                                Err(err) => {
+                                    warn!(
+                                        bucket = %oi.bucket,
+                                        object = %oi.name,
+                                        remote_object = %oi.transitioned_object.name,
+                                        remote_version_id = %oi.transitioned_object.version_id,
+                                        tier = %oi.transitioned_object.tier,
+                                        error = ?err,
+                                        "failed to delete transitioned free version after remote tier sweep"
+                                    );
+                                    break;
+                                }
+                            }
+                        }
+
+                        if !deleted_locally {
+                            warn!(
+                                bucket = %oi.bucket,
+                                object = %oi.name,
+                                remote_object = %oi.transitioned_object.name,
+                                remote_version_id = %oi.transitioned_object.version_id,
+                                tier = %oi.transitioned_object.tier,
+                                "transitioned free version was not found during local cleanup"
+                            );
+                        }
                     }
                     else {
                         //info!("Invalid work type - {:?}", v);

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -120,9 +120,9 @@ impl ObjSweeper {
 #[derive(Debug, Clone)]
 #[allow(unused_assignments)]
 pub struct Jentry {
-    obj_name: String,
-    version_id: String,
-    tier_name: String,
+    pub(crate) obj_name: String,
+    pub(crate) version_id: String,
+    pub(crate) tier_name: String,
 }
 
 impl ExpiryOp for Jentry {
@@ -147,5 +147,37 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
     w.remove(obj_name, rv_id).await
 }
 
+pub fn transitioned_delete_journal_entry(
+    version_id: Option<Uuid>,
+    versioned: bool,
+    suspended: bool,
+    transitioned: &TransitionedObject,
+) -> Option<Jentry> {
+    let sweeper = ObjSweeper {
+        version_id,
+        versioned,
+        suspended,
+        transition_status: transitioned.status.clone(),
+        transition_tier: transitioned.tier.clone(),
+        transition_version_id: transitioned.version_id.clone(),
+        remote_object: transitioned.name.clone(),
+        ..Default::default()
+    };
+
+    sweeper.should_remove_remote_object()
+}
+
+pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject) -> Option<Jentry> {
+    if transitioned.status != lifecycle::TRANSITION_COMPLETE {
+        return None;
+    }
+
+    Some(Jentry {
+        obj_name: transitioned.name.clone(),
+        version_id: transitioned.version_id.clone(),
+        tier_name: transitioned.tier.clone(),
+    })
+}
+
 #[cfg(test)]
 mod test {}

crates/ecstore/src/set_disk.rs

@@ -1077,7 +1077,7 @@ impl ObjectOperations for SetDisks {
 
         let (mut metas, errs) = {
             if let Some(vid) = &src_opts.version_id {
-                Self::read_all_fileinfo(&disks, "", src_bucket, src_object, vid, true, false).await?
+                Self::read_all_fileinfo(&disks, "", src_bucket, src_object, vid, true, false, false).await?
             } else {
                 Self::read_all_xl(&disks, src_bucket, src_object, true, false).await
             }
@@ -1565,8 +1565,6 @@ impl ObjectOperations for SetDisks {
             return Ok(oi);
         }
 
-        let version_id = opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok());
-
         // Create a single object deletion request
         let mut dfi = FileInfo {
             name: object.to_string(),
@@ -1673,7 +1671,7 @@ impl ObjectOperations for SetDisks {
 
         let (metas, errs) = {
             if let Some(version_id) = &opts.version_id {
-                Self::read_all_fileinfo(&disks, "", bucket, object, version_id.to_string().as_str(), false, false).await?
+                Self::read_all_fileinfo(&disks, "", bucket, object, version_id.to_string().as_str(), false, false, false).await?
             } else {
                 Self::read_all_xl(&disks, bucket, object, false, false).await
             }
@@ -3279,7 +3277,7 @@ impl HealOperations for SetDisks {
         let disks = self.disks.read().await;
 
         let disks = disks.clone();
-        let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false).await?;
+        let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false).await?;
         if DiskError::is_all_not_found(&errs) {
             warn!(
                 "heal_object failed, all obj part not found, bucket: {}, obj: {}, version_id: {}",

crates/ecstore/src/set_disk/heal.rs

@@ -56,7 +56,8 @@ impl SetDisks {
             }
         };
 
-        let (mut parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true).await?;
+        let (mut parts_metadata, errs) =
+            Self::read_all_fileinfo(&disks, "", bucket, object, version_id, true, true, false).await?;
 
         info!(
             parts_count = parts_metadata.len(),

crates/ecstore/src/set_disk/multipart.rs

@@ -105,7 +105,8 @@ impl SetDisks {
         let disks = disks.clone();
 
         let (parts_metadata, errs) =
-            Self::read_all_fileinfo(&disks, bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, "", false, false).await?;
+            Self::read_all_fileinfo(&disks, bucket, RUSTFS_META_MULTIPART_BUCKET, &upload_id_path, "", false, false, false)
+                .await?;
 
         let map_err_notfound = |err: DiskError| {
             if err == DiskError::FileNotFound {

crates/ecstore/src/set_disk/read.rs

@@ -131,6 +131,7 @@ impl SetDisks {
         Ok(ret)
     }
 
+    #[allow(clippy::too_many_arguments)]
     #[tracing::instrument(level = "debug", skip(disks))]
     pub(super) async fn read_all_fileinfo(
         disks: &[Option<DiskStore>],
@@ -140,13 +141,14 @@ impl SetDisks {
         version_id: &str,
         read_data: bool,
         healing: bool,
+        incl_free_versions: bool,
     ) -> disk::error::Result<(Vec<FileInfo>, Vec<Option<DiskError>>)> {
         let mut ress = Vec::with_capacity(disks.len());
         let mut errors = Vec::with_capacity(disks.len());
         let opts = Arc::new(ReadOptions {
+            incl_free_versions,
             read_data,
             healing,
-            ..Default::default()
         });
         let org_bucket = Arc::new(org_bucket.to_string());
         let bucket = Arc::new(bucket.to_string());
@@ -474,7 +476,8 @@ impl SetDisks {
         let vid = opts.version_id.clone().unwrap_or_default();
 
         // TODO: optimize concurrency and break once enough slots are available
-        let (parts_metadata, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false).await?;
+        let (parts_metadata, errs) =
+            Self::read_all_fileinfo(&disks, "", bucket, object, vid.as_str(), read_data, false, opts.incl_free_versions).await?;
         // warn!("get_object_fileinfo parts_metadata {:?}", &parts_metadata);
         // warn!("get_object_fileinfo {}/{} errs {:?}", bucket, object, &errs);
 
@@ -541,6 +544,9 @@ impl SetDisks {
         }
 
         if fi.deleted {
+            if opts.incl_free_versions && fi.tier_free_version() && opts.version_id.is_some() {
+                return (oi, write_quorum, None);
+            }
             return if opts.version_id.is_none() || opts.delete_marker {
                 (oi, write_quorum, Some(to_object_err(StorageError::FileNotFound, vec![bucket, object])))
             } else {

crates/ecstore/src/store/object.rs

@@ -286,7 +286,9 @@ impl ECStore {
         }
 
         if !errs.is_empty() && !opts.versioned && !opts.version_suspended {
-            return self.delete_object_from_all_pools(bucket, object, &opts, errs).await;
+            let mut obj = self.delete_object_from_all_pools(bucket, object, &opts, errs).await?;
+            obj.name = decode_dir_object(object);
+            return Ok(obj);
         }
 
         for pool in self.pools.iter() {

crates/ecstore/src/store_api/types.rs

@@ -47,6 +47,7 @@ pub struct ObjectOptions {
 
     pub versioned: bool,
     pub version_suspended: bool,
+    pub incl_free_versions: bool,
 
     pub skip_decommissioned: bool,
     pub skip_rebalancing: bool,

crates/scanner/src/scanner_io.rs

@@ -23,6 +23,7 @@ use rand::seq::SliceRandom as _;
 use rustfs_common::heal_channel::HealScanMode;
 use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete};
 use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
+use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState;
 use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle;
 use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_object_lock_config, get_replication_config};
 use rustfs_ecstore::bucket::replication::{ReplicationConfig, ReplicationConfigurationExt};
@@ -530,6 +531,11 @@ impl ScannerIODisk for Disk {
             .iter()
             .map(|v| ObjectInfo::from_file_info(v, item.bucket.as_str(), item.object_path().as_str(), versioned))
             .collect::<Vec<ObjectInfo>>();
+        let free_version_infos = fivs
+            .free_versions
+            .iter()
+            .map(|v| ObjectInfo::from_file_info(v, item.bucket.as_str(), item.object_path().as_str(), versioned))
+            .collect::<Vec<ObjectInfo>>();
 
         let mut size_summary = SizeSummary::default();
 
@@ -563,9 +569,14 @@ impl ScannerIODisk for Disk {
         item.apply_actions(ecstore, object_infos, lock_config, &mut size_summary)
             .await;
 
-        done_object();
+        if !free_version_infos.is_empty() {
+            let mut expiry_state = GLOBAL_ExpiryState.write().await;
+            for oi in free_version_infos {
+                expiry_state.enqueue_free_version(oi).await;
+            }
+        }
 
-        // TODO: enqueueFreeVersion
+        done_object();
 
         Ok(size_summary)
     }

crates/scanner/tests/lifecycle_integration_test.rs

@@ -18,8 +18,10 @@ use rustfs_ecstore::{
     bucket::{lifecycle::bucket_lifecycle_ops::enqueue_transition_for_existing_objects, metadata_sys},
     client::transition_api::{ReadCloser, ReaderImpl},
     disk::endpoint::Endpoint,
+    disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, new_disk},
     endpoints::{EndpointServerPools, Endpoints, PoolEndpoints},
     global::GLOBAL_TierConfigMgr,
+    pools::path2_bucket_object_with_base_path,
     store::ECStore,
     store_api::{
         BucketOperations, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader,
@@ -29,13 +31,17 @@ use rustfs_ecstore::{
         warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options},
     },
 };
+use rustfs_filemeta::FileMeta;
 use rustfs_scanner::scanner::init_data_scanner;
+use rustfs_scanner::scanner_folder::ScannerItem;
+use rustfs_scanner::scanner_io::ScannerIODisk;
+use rustfs_utils::path::path_join_buf;
 use s3s::dto::RestoreRequest;
 use serial_test::serial;
 use std::{
     collections::HashMap,
     io::Cursor,
-    path::PathBuf,
+    path::{Path, PathBuf},
     sync::{Arc, Once, OnceLock},
     time::Duration,
 };
@@ -137,6 +143,70 @@ async fn setup_test_env() -> (Vec<PathBuf>, Arc<ECStore>) {
     (disk_paths, ecstore)
 }
 
+async fn setup_isolated_test_env(init_expiry: bool) -> (Vec<PathBuf>, Arc<ECStore>) {
+    init_tracing();
+
+    let test_base_dir = format!("/tmp/rustfs_scanner_lifecycle_test_{}", uuid::Uuid::new_v4());
+    let temp_dir = std::path::PathBuf::from(&test_base_dir);
+    if temp_dir.exists() {
+        fs::remove_dir_all(&temp_dir).await.ok();
+    }
+    fs::create_dir_all(&temp_dir).await.unwrap();
+
+    let disk_paths = vec![
+        temp_dir.join("disk1"),
+        temp_dir.join("disk2"),
+        temp_dir.join("disk3"),
+        temp_dir.join("disk4"),
+    ];
+
+    for disk_path in &disk_paths {
+        fs::create_dir_all(disk_path).await.unwrap();
+    }
+
+    let mut endpoints = Vec::new();
+    for (i, disk_path) in disk_paths.iter().enumerate() {
+        let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(i);
+        endpoints.push(endpoint);
+    }
+
+    let pool_endpoints = PoolEndpoints {
+        legacy: false,
+        set_count: 1,
+        drives_per_set: 4,
+        endpoints: Endpoints::from(endpoints),
+        cmd_line: "test".to_string(),
+        platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH),
+    };
+
+    let endpoint_pools = EndpointServerPools(vec![pool_endpoints]);
+    rustfs_ecstore::store::init_local_disks(endpoint_pools.clone()).await.unwrap();
+
+    let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap();
+    let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new())
+        .await
+        .unwrap();
+
+    let buckets_list = ecstore
+        .list_bucket(&rustfs_ecstore::store_api::BucketOptions {
+            no_metadata: true,
+            ..Default::default()
+        })
+        .await
+        .unwrap();
+    let buckets = buckets_list.into_iter().map(|v| v.name).collect();
+    rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await;
+
+    if init_expiry {
+        rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::init_background_expiry(ecstore.clone()).await;
+    }
+
+    (disk_paths, ecstore)
+}
+
 /// Test helper: Create a test bucket
 #[allow(dead_code)]
 async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
@@ -374,6 +444,85 @@ async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &
     }
 }
 
+async fn wait_for_remote_absence(backend: &MockWarmBackend, object: &str, timeout: Duration) -> bool {
+    let deadline = tokio::time::Instant::now() + timeout;
+
+    loop {
+        if !backend.objects.lock().await.contains_key(object) {
+            return true;
+        }
+
+        if tokio::time::Instant::now() >= deadline {
+            return false;
+        }
+
+        tokio::time::sleep(Duration::from_millis(50)).await;
+    }
+}
+
+async fn free_version_count(disk_path: &Path, bucket: &str, object: &str) -> usize {
+    let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
+    endpoint.set_pool_index(0);
+    endpoint.set_set_index(0);
+    endpoint.set_disk_index(0);
+    let disk = new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .expect("failed to open local disk");
+    let data = disk
+        .read_metadata(bucket, &path_join_buf(&[object, STORAGE_FORMAT_FILE]))
+        .await;
+    let Ok(data) = data else {
+        return 0;
+    };
+    let meta = FileMeta::load(&data).expect("failed to load file metadata");
+    meta.get_file_info_versions(bucket, object, false)
+        .expect("failed to decode file info versions")
+        .free_versions
+        .len()
+}
+
+async fn scan_object_metadata(disk_path: &Path, bucket: &str, object: &str) {
+    let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
+    endpoint.set_pool_index(0);
+    endpoint.set_set_index(0);
+    endpoint.set_disk_index(0);
+    let disk = new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .expect("failed to open local disk");
+    let metadata_path = disk_path.join(bucket).join(object).join(STORAGE_FORMAT_FILE);
+    let relative_path = metadata_path.to_string_lossy().to_string();
+    let (_, scanner_path) = path2_bucket_object_with_base_path(disk_path.to_string_lossy().as_ref(), relative_path.as_str());
+    let file_type = fs::metadata(&metadata_path)
+        .await
+        .expect("failed to stat object metadata")
+        .file_type();
+    let item = ScannerItem {
+        path: scanner_path.clone(),
+        bucket: bucket.to_string(),
+        prefix: object.to_string(),
+        object_name: STORAGE_FORMAT_FILE.to_string(),
+        file_type,
+        lifecycle: None,
+        replication: None,
+        heal_enabled: false,
+        heal_bitrot: false,
+        debug: false,
+    };
+    disk.get_size(item).await.expect("scanner get_size should succeed");
+}
+
 #[derive(Clone, Default)]
 struct MockStoredObject {
     bytes: Vec<u8>,
@@ -480,6 +629,15 @@ async fn register_mock_tier(tier_name: &str) -> MockWarmBackend {
             version: "v1".to_string(),
             tier_type: TierType::MinIO,
             name: tier_name.to_string(),
+            minio: Some(TierMinIO {
+                access_key: "minioadmin".to_string(),
+                secret_key: "minioadmin".to_string(),
+                bucket: "mock-tier".to_string(),
+                endpoint: "http://127.0.0.1:0".to_string(),
+                prefix: format!("mock/{}/", Uuid::new_v4()),
+                region: String::new(),
+                ..Default::default()
+            }),
             ..Default::default()
         },
     );
@@ -913,4 +1071,64 @@ mod serial_tests {
             .expect("Failed to consume restored object stream");
         assert_eq!(data, expected);
     }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+    #[serial]
+    #[ignore = "requires isolated global object layer state"]
+    async fn test_scanner_enqueues_free_version_cleanup_for_stale_transitioned_object() {
+        let (disk_paths, ecstore) = setup_isolated_test_env(false).await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&tier_name).await;
+
+        let bucket_name = format!("test-scanner-free-version-{}", &Uuid::new_v4().simple().to_string()[..8]);
+        let object_name = "test/object.txt";
+        let initial_payload = b"scanner should clean stale transitioned null version";
+        create_test_bucket(&ecstore, bucket_name.as_str()).await;
+        set_bucket_lifecycle_transition_with_tier(bucket_name.as_str(), &tier_name)
+            .await
+            .expect("Failed to set lifecycle configuration");
+
+        upload_test_object(&ecstore, bucket_name.as_str(), object_name, initial_payload).await;
+        enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
+            .await
+            .expect("Failed to enqueue transitioned object");
+
+        let transitioned = wait_for_transition(&ecstore, bucket_name.as_str(), object_name, TRANSITION_WAIT_TIMEOUT)
+            .await
+            .expect("object should transition before overwrite");
+        let stale_remote_object = transitioned.transitioned_object.name.clone();
+        assert!(backend.objects.lock().await.contains_key(&stale_remote_object));
+
+        ecstore
+            .delete_object(bucket_name.as_str(), object_name, ObjectOptions::default())
+            .await
+            .expect("Failed to delete transitioned object without expiry workers");
+
+        assert!(
+            free_version_count(&disk_paths[0], bucket_name.as_str(), object_name).await > 0,
+            "deleting a transitioned null version should leave a free version for async cleanup"
+        );
+        assert!(
+            backend.objects.lock().await.contains_key(&stale_remote_object),
+            "stale transitioned remote object should still exist before scanner fallback runs"
+        );
+
+        rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::init_background_expiry(ecstore.clone()).await;
+        scan_object_metadata(&disk_paths[0], bucket_name.as_str(), object_name).await;
+
+        assert!(
+            wait_for_remote_absence(&backend, &stale_remote_object, TRANSITION_WAIT_TIMEOUT).await,
+            "scanner should enqueue stale free-version cleanup for the transitioned remote object"
+        );
+        assert_eq!(
+            free_version_count(&disk_paths[0], bucket_name.as_str(), object_name).await,
+            0,
+            "free-version metadata should be removed after scanner-triggered cleanup"
+        );
+        assert!(
+            wait_for_object_absence(&ecstore, bucket_name.as_str(), object_name, Duration::from_secs(1)).await,
+            "deleted object should remain absent after scanner cleanup"
+        );
+    }
 }