Merge branch 'main' into overtrue/fix/heal-topology-lock

Keep the canonical commit module after concurrent storage changes merged.
The control-write and rollback changes are already present there.

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

crates/ecstore/src/set_disk/ops/heal.rs

@@ -2490,9 +2490,9 @@ impl crate::storage_api_contracts::heal::HealOperations for SetDisks {
             return Ok((result, err.map(|e| e.into())));
         }
 
-        let disks = self.disks.read().await;
-
-        let disks = disks.clone();
+        // The inner heal and missing-object report read the registry again;
+        // release this snapshot guard before a topology writer can queue between reads.
+        let disks = self.get_disks_internal().await;
         let (_, errs) = Self::read_all_fileinfo(&disks, "", bucket, object, version_id, false, false, false)
             .await
             .map_err(|e| to_object_err(e.into(), vec![bucket, object]))?;
@@ -3419,6 +3419,366 @@ mod heal_result_report_tests {
         assert_eq!(unformatted, DiskError::UnformattedDisk);
     }
 
+    #[derive(Clone, Copy)]
+    enum InventoryWriterHealCase {
+        Existing,
+        Missing,
+        MissingVersion,
+    }
+
+    async fn assert_heal_object_inventory_writer(case: InventoryWriterHealCase) {
+        use crate::set_disk::core::io_primitives::disk_call_counters;
+        use std::time::Duration;
+        use tokio::io::AsyncReadExt;
+
+        let (_temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "heal-inventory-writer-bucket";
+        let object = match case {
+            InventoryWriterHealCase::Existing => "heal-inventory-writer-existing",
+            InventoryWriterHealCase::Missing => "heal-inventory-writer-missing",
+            InventoryWriterHealCase::MissingVersion => "heal-inventory-writer-missing-version",
+        };
+        set.make_bucket(
+            bucket,
+            &MakeBucketOptions {
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("heal fixture bucket should be created");
+        let body = vec![0x67; 64 * 1024];
+        let stored_version = Uuid::new_v4();
+        let stored_version_string = stored_version.to_string();
+        let published = if matches!(case, InventoryWriterHealCase::Missing) {
+            None
+        } else {
+            let mut reader = PutObjReader::from_vec(body.clone());
+            let info = set
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        no_lock: true,
+                        versioned: true,
+                        version_id: Some(stored_version_string.clone()),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("full-fanout PUT should seed the heal fixture");
+            for disk in &disks {
+                let metadata = disk
+                    .read_version("", bucket, object, &stored_version_string, &ReadOptions::default())
+                    .await
+                    .expect("the seeded version must be present on every disk");
+                assert_eq!(metadata.version_id, Some(stored_version));
+                assert_eq!(metadata.size, i64::try_from(body.len()).expect("fixture size should fit i64"));
+            }
+            Some(info)
+        };
+        let requested_version = match case {
+            InventoryWriterHealCase::Existing => stored_version_string.clone(),
+            InventoryWriterHealCase::Missing => String::new(),
+            InventoryWriterHealCase::MissingVersion => Uuid::new_v4().to_string(),
+        };
+        let opts = HealOpts {
+            no_lock: true,
+            ..Default::default()
+        };
+        let calls = disk_call_counters::observe(object);
+        let read_gate = set.disks.read().await;
+        // UFCS selects the trait's outer precheck, not the same-named inherent heal.
+        let heal = <SetDisks as crate::storage_api_contracts::heal::HealOperations>::heal_object(
+            set.as_ref(),
+            bucket,
+            object,
+            &requested_version,
+            &opts,
+        );
+        tokio::pin!(heal);
+        assert!(matches!(
+            futures::poll!(tokio::task::unconstrained(heal.as_mut())),
+            std::task::Poll::Pending
+        ));
+        // These tests use the current-thread runtime: full-wait metadata tasks
+        // have been spawned, but cannot run during the single unconstrained poll.
+        assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 0);
+        let writer = set.disks.write();
+        tokio::pin!(writer);
+        assert!(matches!(
+            futures::poll!(tokio::task::unconstrained(writer.as_mut())),
+            std::task::Poll::Pending
+        ));
+        assert!(set.disks.try_read().is_err(), "the writer must already block new inventory readers");
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while calls.total(disk_call_counters::KIND_READ_VERSION) < 4 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the suspended trait heal must have started the real metadata fanout");
+        for disk_index in 0..4 {
+            assert_eq!(calls.for_disk(disk_call_counters::KIND_READ_VERSION, disk_index), 1);
+        }
+        drop(read_gate);
+
+        let (_, outcome) =
+            tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(async { drop(writer.await) }, heal) })
+                .await
+                .expect("trait heal must not deadlock its nested inventory read with the queued writer");
+        let (result, error) = outcome.expect("heal should report the object's outcome");
+        match case {
+            InventoryWriterHealCase::Existing => assert!(error.is_none(), "existing object heal failed: {error:?}"),
+            InventoryWriterHealCase::Missing => assert!(matches!(error, Some(Error::FileNotFound))),
+            InventoryWriterHealCase::MissingVersion => assert!(matches!(error, Some(Error::FileVersionNotFound))),
+        }
+        assert_eq!(result.bucket, bucket);
+        assert_eq!(result.object, object);
+        assert_eq!(result.version_id, requested_version);
+        assert_eq!(result.disk_count, 4);
+        assert_eq!(result.before.drives.len(), 4);
+        assert_eq!(result.after.drives.len(), 4);
+        for disk_index in 0..4 {
+            let endpoint = set.set_endpoints[disk_index].to_string();
+            assert_eq!(result.before.drives[disk_index].endpoint, endpoint);
+            assert_eq!(result.after.drives[disk_index].endpoint, endpoint);
+        }
+        if let Some(published) = published {
+            tokio::time::timeout(Duration::from_secs(10), async {
+                let mut reader = set
+                    .get_object_reader(
+                        bucket,
+                        object,
+                        None,
+                        Default::default(),
+                        &ObjectOptions {
+                            versioned: true,
+                            version_id: Some(stored_version_string),
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("the stored version must remain readable after heal");
+                assert_eq!(reader.object_info.etag, published.etag);
+                assert_eq!(reader.object_info.version_id, Some(stored_version));
+                let mut observed_body = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut observed_body)
+                    .await
+                    .expect("stored body should stream");
+                assert_eq!(observed_body, body);
+            })
+            .await
+            .expect("GET must finish after the inventory writer and heal");
+        }
+    }
+
+    #[tokio::test]
+    async fn heal_object_inventory_writer_existing() {
+        assert_heal_object_inventory_writer(InventoryWriterHealCase::Existing).await;
+    }
+
+    #[tokio::test]
+    async fn heal_object_inventory_writer_missing() {
+        assert_heal_object_inventory_writer(InventoryWriterHealCase::Missing).await;
+    }
+
+    #[tokio::test]
+    async fn heal_object_inventory_writer_missing_version() {
+        assert_heal_object_inventory_writer(InventoryWriterHealCase::MissingVersion).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn heal_object_with_queued_disk_renewal() {
+        use crate::layout::endpoints::SetupType;
+        use crate::runtime::instance::InstanceContext;
+        use crate::set_disk::core::io_primitives::disk_call_counters;
+        use std::collections::HashMap;
+        use std::future::Future;
+        use std::task::Poll;
+        use std::time::Duration;
+        use tokio::io::AsyncReadExt;
+
+        // renew_disk still registers local disks on the ambient context. Match
+        // the default serial group used by its other setup/registry fixtures,
+        // and restore only this temporary endpoint, including on a failed join.
+        struct RenewDiskTestState {
+            ctx: Arc<InstanceContext>,
+            was_dist_erasure: bool,
+            map: Arc<RwLock<HashMap<String, Option<DiskStore>>>>,
+            endpoint: String,
+            previous_disk: Option<Option<DiskStore>>,
+        }
+
+        impl Drop for RenewDiskTestState {
+            fn drop(&mut self) {
+                let ctx = self.ctx.clone();
+                let was_dist_erasure = self.was_dist_erasure;
+                let map = self.map.clone();
+                let endpoint = self.endpoint.clone();
+                let previous_disk = self.previous_disk.take();
+                let handle = tokio::runtime::Handle::current();
+                std::thread::spawn(move || {
+                    handle.block_on(async move {
+                        let mut map = map.write().await;
+                        match previous_disk {
+                            Some(disk) => {
+                                map.insert(endpoint, disk);
+                            }
+                            None => {
+                                map.remove(&endpoint);
+                            }
+                        }
+                        drop(map);
+                        if was_dist_erasure {
+                            ctx.update_erasure_type(SetupType::DistErasure).await;
+                        }
+                    });
+                })
+                .join()
+                .expect("renew fixture state restoration should finish");
+            }
+        }
+
+        let (_temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let endpoint = set.set_endpoints[0].clone();
+        let ctx = crate::runtime::global::current_ctx();
+        let map = ctx.local_disk_map();
+        let restore = RenewDiskTestState {
+            ctx: ctx.clone(),
+            was_dist_erasure: ctx.is_dist_erasure().await,
+            map: map.clone(),
+            endpoint: endpoint.to_string(),
+            previous_disk: map.read().await.get(&endpoint.to_string()).cloned(),
+        };
+        // Only distributed erasure needs an override to avoid the ambient slot array.
+        if restore.was_dist_erasure {
+            ctx.update_erasure_type(SetupType::Erasure).await;
+        }
+
+        let bucket = "heal-disk-renewal-bucket";
+        let object = "heal-disk-renewal-object";
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("renew fixture bucket should be created");
+        let body = vec![0x73; 64 * 1024];
+        let mut reader = PutObjReader::from_vec(body.clone());
+        let published = set
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("full-fanout PUT should seed the renewal fixture");
+        for disk in &disks {
+            let metadata = disk
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("the seeded object must be present on every disk");
+            assert_eq!(metadata.size, i64::try_from(body.len()).expect("fixture size should fit i64"));
+        }
+
+        let opts = HealOpts {
+            no_lock: true,
+            ..Default::default()
+        };
+        let calls = disk_call_counters::observe(object);
+        let read_gate = set.disks.read().await;
+        let heal = <SetDisks as crate::storage_api_contracts::heal::HealOperations>::heal_object(
+            set.as_ref(),
+            bucket,
+            object,
+            "",
+            &opts,
+        );
+        tokio::pin!(heal);
+        assert!(matches!(futures::poll!(tokio::task::unconstrained(heal.as_mut())), Poll::Pending));
+        assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 0);
+
+        let renew = set.renew_disk(&endpoint);
+        tokio::pin!(renew);
+        tokio::time::timeout(
+            Duration::from_secs(5),
+            futures::future::poll_fn(|cx| {
+                assert!(
+                    std::pin::pin!(tokio::task::unconstrained(renew.as_mut()))
+                        .poll(cx)
+                        .is_pending(),
+                    "renewal must reach its inventory write before returning"
+                );
+                if set.disks.try_read().is_err() {
+                    Poll::Ready(())
+                } else {
+                    Poll::Pending
+                }
+            }),
+        )
+        .await
+        .expect("real renewal must queue its topology writer behind the read gate");
+        let registered = map
+            .read()
+            .await
+            .get(&endpoint.to_string())
+            .cloned()
+            .flatten()
+            .expect("renewal must register the connected disk before its inventory write");
+        assert!(!Arc::ptr_eq(&registered, &disks[0]), "renewal must construct a new disk handle");
+        tokio::time::timeout(Duration::from_secs(5), async {
+            while calls.total(disk_call_counters::KIND_READ_VERSION) < 4 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("the suspended trait heal must have started the real metadata fanout");
+        for disk_index in 0..4 {
+            assert_eq!(calls.for_disk(disk_call_counters::KIND_READ_VERSION, disk_index), 1);
+        }
+        drop(read_gate);
+
+        let (_, outcome) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(renew, heal) })
+            .await
+            .expect("trait heal and real disk renewal must finish without a nested inventory read deadlock");
+        let (report, error) = outcome.expect("heal should report the existing object");
+        assert!(error.is_none(), "existing object heal failed after renewal: {error:?}");
+        assert_eq!(report.bucket, bucket);
+        assert_eq!(report.object, object);
+        assert_eq!(report.disk_count, 4);
+        let renewed = set.get_disks_internal().await[0]
+            .clone()
+            .expect("the renewed slot must remain online");
+        assert!(Arc::ptr_eq(&renewed, &registered), "the set must publish the newly connected handle");
+        assert_eq!(renewed.endpoint(), endpoint);
+        let format = load_format_erasure(&renewed, false)
+            .await
+            .expect("renewed disk format should remain readable");
+        assert_eq!(format.erasure.this, set.format.erasure.sets[0][0]);
+        tokio::time::timeout(Duration::from_secs(10), async {
+            let mut reader = set
+                .get_object_reader(bucket, object, None, Default::default(), &ObjectOptions::default())
+                .await
+                .expect("the object must remain readable after renewal and heal");
+            assert_eq!(reader.object_info.etag, published.etag);
+            let mut observed_body = Vec::new();
+            reader
+                .stream
+                .read_to_end(&mut observed_body)
+                .await
+                .expect("stored body should stream");
+            assert_eq!(observed_body, body);
+        })
+        .await
+        .expect("GET must finish after renewal and heal");
+    }
+
     // Regression for #955: an offline disk must contribute exactly one drive
     // record. Before the fix the offline branch fell through and pushed a second
     // (Corrupt) record for the same disk, so `before/after.drives` grew to

crates/lifecycle/src/core.rs

@@ -43,10 +43,6 @@ const ERR_LIFECYCLE_BUCKET_LOCKED: &str =
     "ExpiredObjectAllVersions element and DelMarkerExpiration action cannot be used on an object locked bucket";
 const ERR_LIFECYCLE_TOO_MANY_RULES: &str = "Lifecycle configuration should have at most 1000 rules";
 const ERR_LIFECYCLE_INVALID_EXPIRATION_DAYS: &str = "'Days' for Expiration action must be a positive integer";
-const ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT: &str = "Expiration cannot specify both Days and Date";
-const ERR_LIFECYCLE_MULTIPLE_TRANSITIONS: &str = "Only one Transition action per lifecycle rule is supported";
-const ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS: &str =
-    "Only one NoncurrentVersionTransition action per lifecycle rule is supported";
 const ERR_LIFECYCLE_INVALID_NONCURRENT_EXPIRATION_DAYS: &str =
     "'NoncurrentDays' for NoncurrentVersionExpiration action must be a positive integer";
 const ERR_LIFECYCLE_INVALID_ABORT_INCOMPLETE_MPU_DAYS: &str =
@@ -365,12 +361,6 @@ impl Lifecycle for BucketLifecycleConfiguration {
                 {
                     return Err(std::io::Error::other(ERR_LIFECYCLE_INVALID_EXPIRED_OBJECT_ALL_VERSIONS));
                 }
-                if expiration.days.is_some() && expiration.date.is_some() {
-                    return Err(std::io::Error::new(
-                        std::io::ErrorKind::InvalidInput,
-                        ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT,
-                    ));
-                }
                 if let Some(expiration_date) = &expiration.date {
                     let date = OffsetDateTime::from(expiration_date.clone());
                     if date.hour() != 0 || date.minute() != 0 || date.second() != 0 || date.nanosecond() != 0 {
@@ -404,20 +394,11 @@ impl Lifecycle for BucketLifecycleConfiguration {
                 }
             }
             if let Some(transitions) = &r.transitions {
-                if transitions.len() > 1 {
-                    return Err(std::io::Error::new(std::io::ErrorKind::InvalidInput, ERR_LIFECYCLE_MULTIPLE_TRANSITIONS));
-                }
                 for transition in transitions {
                     TransitionOps::validate(transition)?;
                 }
             }
             if let Some(noncurrent_transitions) = &r.noncurrent_version_transitions {
-                if noncurrent_transitions.len() > 1 {
-                    return Err(std::io::Error::new(
-                        std::io::ErrorKind::InvalidInput,
-                        ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS,
-                    ));
-                }
                 for transition in noncurrent_transitions {
                     NoncurrentVersionTransitionOps::validate(transition)?;
                 }
@@ -492,8 +473,6 @@ impl Lifecycle for BucketLifecycleConfiguration {
     }
 
     async fn eval(&self, obj: &ObjectOpts) -> Event {
-        // A single-object lookup cannot prove how many newer historical versions
-        // survive. Count-dependent actions wait for the complete-group evaluator.
         self.eval_inner(obj, OffsetDateTime::now_utc(), 0).await
     }
 
@@ -557,8 +536,23 @@ impl Lifecycle for BucketLifecycleConfiguration {
             return Event::default();
         };
 
-        if let Some(event) = obj.restored_copy_expiry(now) {
-            events.push(event);
+        if let Some(restore_expires) = obj.restore_expires
+            && restore_expires.unix_timestamp() != 0
+            && now.unix_timestamp() > restore_expires.unix_timestamp()
+        {
+            let mut action = IlmAction::DeleteRestoredAction;
+            if !obj.is_latest {
+                action = IlmAction::DeleteRestoredVersionAction;
+            }
+
+            events.push(Event {
+                action,
+                due: Some(now),
+                rule_id: "".into(),
+                noncurrent_days: 0,
+                newer_noncurrent_versions: 0,
+                storage_class: "".into(),
+            });
         }
 
         if let Some(ref lc_rules) = self.filter_rules(obj).await {
@@ -617,12 +611,17 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     continue;
                 }
 
+                if !obj.is_latest
+                    && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
+                    && let Some(retain_newer_noncurrent_versions) = noncurrent_version_expiration.newer_noncurrent_versions
+                    && newer_noncurrent_versions < usize::try_from(retain_newer_noncurrent_versions).unwrap_or(usize::MAX)
+                {
+                    continue;
+                }
+
                 if !obj.is_latest
                     && let Some(ref noncurrent_version_expiration) = rule.noncurrent_version_expiration
                     && let Some(noncurrent_days) = noncurrent_version_expiration.noncurrent_days
-                    && noncurrent_version_expiration
-                        .newer_noncurrent_versions
-                        .is_none_or(|retain| usize::try_from(retain).is_ok_and(|retain| newer_noncurrent_versions >= retain))
                 {
                     if let Some(successor_mod_time) = obj.successor_mod_time {
                         let expected_expiry = expected_expiry_time(successor_mod_time, noncurrent_days);
@@ -652,11 +651,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     && let Some(noncurrent_version_transition) = rule
                         .noncurrent_version_transitions
                         .as_ref()
-                        .filter(|transitions| transitions.len() == 1)
                         .and_then(|transitions| transitions.first())
-                    && noncurrent_version_transition
-                        .newer_noncurrent_versions
-                        .is_none_or(|retain| usize::try_from(retain).is_ok_and(|retain| newer_noncurrent_versions >= retain))
                     && let Some(storage_class) = noncurrent_version_transition.storage_class.as_ref()
                     && !storage_class.as_str().is_empty()
                     && !obj.delete_marker
@@ -740,11 +735,7 @@ impl Lifecycle for BucketLifecycleConfiguration {
                     }
 
                     if obj.transition_status != TRANSITION_COMPLETE
-                        && let Some(transition) = rule
-                            .transitions
-                            .as_ref()
-                            .filter(|transitions| transitions.len() == 1)
-                            .and_then(|transitions| transitions.first())
+                        && let Some(transition) = rule.transitions.as_ref().and_then(|transitions| transitions.first())
                         && let Some(storage_class) = transition.storage_class.as_ref()
                         && !storage_class.as_str().is_empty()
                     {
@@ -767,15 +758,18 @@ impl Lifecycle for BucketLifecycleConfiguration {
         }
 
         if !events.is_empty() {
-            // Eligible expiration takes precedence over transition, even when a
-            // failed transition has an earlier deadline. Within each action class,
-            // prefer the earliest deadline using a deterministic total order.
+            // Select the winning event using a strict total order (MinIO semantics):
+            // the earliest `due` wins, and ties break toward delete-type actions. A
+            // missing `due` is treated as UNIX_EPOCH. This replaces a hand-written
+            // `sort_by` comparator that was not a strict weak ordering (it could return
+            // `Ordering::Less` for both `(a, b)` and `(b, a)`), which panics on the
+            // repository toolchain and did not deterministically pick the earliest event.
             let event = events
                 .iter()
                 .min_by_key(|event| {
                     (
-                        ilm_action_priority_rank(&event.action),
                         event.due.unwrap_or(OffsetDateTime::UNIX_EPOCH).unix_timestamp(),
+                        ilm_action_priority_rank(&event.action),
                     )
                 })
                 .cloned()
@@ -1048,27 +1042,6 @@ impl ObjectOpts {
     pub fn expired_object_deletemarker(&self) -> bool {
         self.delete_marker && self.is_latest && self.num_versions == 1
     }
-
-    pub(crate) fn restored_copy_expiry(&self, now: OffsetDateTime) -> Option<Event> {
-        let restore_expires = self.restore_expires?;
-        // Restore metadata alone does not prove that a durable remote copy exists.
-        if self.transition_status != TRANSITION_COMPLETE
-            || restore_expires.unix_timestamp() == 0
-            || now.unix_timestamp() <= restore_expires.unix_timestamp()
-        {
-            return None;
-        }
-        let action = if self.is_latest {
-            IlmAction::DeleteRestoredAction
-        } else {
-            IlmAction::DeleteRestoredVersionAction
-        };
-        expiration_action_has_valid_target(action, self.version_id, self.is_latest, self.delete_marker).then(|| Event {
-            action,
-            due: Some(now),
-            ..Default::default()
-        })
-    }
 }
 
 /// Returns whether an expiry action has enough identity to target the object
@@ -1091,8 +1064,11 @@ pub fn expiration_action_has_valid_target(
     }
 }
 
-/// Eligible logical expiration takes precedence over transition and restore-copy
-/// cleanup. Deadlines break ties within an action class.
+/// Total-order rank for lifecycle actions used to break `due` ties.
+///
+/// Delete-type actions rank before every other action so that, when two events
+/// share the same `due`, a delete wins (MinIO semantics). The concrete numeric
+/// values only matter relative to each other.
 fn ilm_action_priority_rank(action: &IlmAction) -> u8 {
     match action {
         IlmAction::DeleteAllVersionsAction
@@ -4183,392 +4159,6 @@ mod tests {
         assert_eq!(event.action, IlmAction::NoneAction);
     }
 
-    mod adversarial_regressions {
-        use super::*;
-        use s3s::dto::NoncurrentVersionExpiration;
-
-        fn run(test: impl std::future::Future<Output = ()>) {
-            with_default_ilm_process_time(|| {
-                tokio::runtime::Builder::new_current_thread()
-                    .build()
-                    .expect("lifecycle regression runtime should build")
-                    .block_on(test);
-            });
-        }
-
-        fn noncurrent_object() -> ObjectOpts {
-            ObjectOpts {
-                name: "logs/object".to_string(),
-                mod_time: Some(datetime!(2020-01-01 00:00:00 UTC)),
-                successor_mod_time: Some(datetime!(2020-01-02 00:00:00 UTC)),
-                version_id: Some(Uuid::from_u128(1)),
-                size: 1024 * 1024,
-                ..Default::default()
-            }
-        }
-
-        #[test]
-        #[serial]
-        fn noncurrent_transition_retains_the_requested_newer_versions() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("retain-two-hot-versions"));
-                rule.filter = Some(LifecycleRuleFilter::default());
-                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
-                    noncurrent_days: Some(1),
-                    newer_noncurrent_versions: Some(2),
-                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                }]);
-                let lc = Arc::new(BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                });
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("valid noncurrent transition policy");
-                let objects = (0..4)
-                    .map(|index| ObjectOpts {
-                        mod_time: Some(datetime!(2020-01-05 00:00:00 UTC) - Duration::days(index)),
-                        successor_mod_time: (index > 0).then_some(datetime!(2020-01-06 00:00:00 UTC) - Duration::days(index)),
-                        version_id: Some(Uuid::from_u128(u128::try_from(index + 1).expect("small version index"))),
-                        is_latest: index == 0,
-                        num_versions: 4,
-                        ..noncurrent_object()
-                    })
-                    .collect::<Vec<_>>();
-                let actions = crate::Evaluator::new(lc)
-                    .eval(&objects)
-                    .await
-                    .expect("complete version chain should evaluate")
-                    .into_iter()
-                    .map(|event| event.action)
-                    .collect::<Vec<_>>();
-                assert_eq!(
-                    actions,
-                    [
-                        IlmAction::NoneAction,
-                        IlmAction::NoneAction,
-                        IlmAction::NoneAction,
-                        IlmAction::TransitionVersionAction
-                    ],
-                    "the two newest noncurrent versions must remain in their current storage class"
-                );
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn noncurrent_transition_checks_count_age_and_single_object_context() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("retain-two"));
-                rule.filter = Some(LifecycleRuleFilter::default());
-                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
-                    noncurrent_days: Some(3),
-                    newer_noncurrent_versions: Some(2),
-                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                }]);
-                let mut lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("valid counted transition");
-                let object = noncurrent_object();
-                let now = datetime!(2020-01-10 00:00:00 UTC);
-                for (newer, expected) in [
-                    (0, IlmAction::NoneAction),
-                    (1, IlmAction::NoneAction),
-                    (2, IlmAction::TransitionVersionAction),
-                    (3, IlmAction::TransitionVersionAction),
-                ] {
-                    assert_eq!(lc.eval_inner(&object, now, newer).await.action, expected, "newer count: {newer}");
-                }
-                assert_eq!(
-                    lc.eval_inner(&object, datetime!(2020-01-04 00:00:00 UTC), 2).await.action,
-                    IlmAction::NoneAction,
-                    "the retention count does not replace the age condition"
-                );
-                assert_eq!(
-                    lc.eval(&object).await.action,
-                    IlmAction::NoneAction,
-                    "a single-object lookup must not assume a complete version history"
-                );
-                for retain in [None, Some(0), Some(-1), Some(i32::MAX)] {
-                    lc.rules[0]
-                        .noncurrent_version_transitions
-                        .as_mut()
-                        .expect("transition exists")[0]
-                        .newer_noncurrent_versions = retain;
-                    let expected = if matches!(retain, None | Some(0)) {
-                        IlmAction::TransitionVersionAction
-                    } else {
-                        IlmAction::NoneAction
-                    };
-                    assert_eq!(lc.eval_inner(&object, now, 2).await.action, expected, "retention: {retain:?}");
-                }
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn noncurrent_expiration_and_transition_have_independent_retention_counts() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("independent-counts"));
-                rule.filter = Some(LifecycleRuleFilter::default());
-                rule.noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
-                    noncurrent_days: Some(90),
-                    newer_noncurrent_versions: Some(4),
-                });
-                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
-                    noncurrent_days: Some(30),
-                    newer_noncurrent_versions: Some(2),
-                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                }]);
-                let lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("valid independent retention limits");
-                let object = noncurrent_object();
-                let now = datetime!(2020-05-01 00:00:00 UTC);
-                for (newer, expected) in [
-                    (1, IlmAction::NoneAction),
-                    (2, IlmAction::TransitionVersionAction),
-                    (3, IlmAction::TransitionVersionAction),
-                    (4, IlmAction::DeleteVersionAction),
-                ] {
-                    assert_eq!(lc.eval_inner(&object, now, newer).await.action, expected, "newer count: {newer}");
-                }
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn expiration_retention_does_not_skip_an_independent_transition() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("transition-then-expire"));
-                rule.filter = Some(LifecycleRuleFilter::default());
-                rule.noncurrent_version_transitions = Some(vec![NoncurrentVersionTransition {
-                    noncurrent_days: Some(1),
-                    newer_noncurrent_versions: None,
-                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                }]);
-                let mut lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                let object = noncurrent_object();
-                let now = datetime!(2020-01-10 00:00:00 UTC);
-                let transition_only = lc.eval_inner(&object, now, 0).await;
-                assert_eq!(transition_only.action, IlmAction::TransitionVersionAction);
-
-                lc.rules[0].noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
-                    noncurrent_days: Some(90),
-                    newer_noncurrent_versions: Some(2),
-                });
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("valid combined policy");
-                let combined = lc.eval_inner(&object, now, 0).await;
-                assert_eq!(combined.action, transition_only.action, "retention limits expiration, not transition");
-                assert_eq!(combined.storage_class, transition_only.storage_class);
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn current_transition_rejects_multiple_stages_in_any_order() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("two-current-transitions"));
-                rule.transitions = Some(vec![
-                    Transition {
-                        date: Some(datetime!(2020-03-01 00:00:00 UTC).into()),
-                        days: None,
-                        storage_class: Some(TransitionStorageClass::from_static("COLD")),
-                    },
-                    Transition {
-                        date: Some(datetime!(2020-01-03 00:00:00 UTC).into()),
-                        days: None,
-                        storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                    },
-                ]);
-                let mut lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                let object = ObjectOpts {
-                    is_latest: true,
-                    ..noncurrent_object()
-                };
-                let now = datetime!(2020-01-10 00:00:00 UTC);
-                for status in [ExpirationStatus::ENABLED, ExpirationStatus::DISABLED] {
-                    lc.rules[0].status = ExpirationStatus::from_static(status);
-                    for _ in 0..2 {
-                        let err = lc
-                            .validate(&ObjectLockConfiguration::default())
-                            .await
-                            .expect_err("multiple transition stages must be rejected");
-                        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
-                        assert_eq!(err.to_string(), ERR_LIFECYCLE_MULTIPLE_TRANSITIONS);
-                        assert_eq!(
-                            lc.eval_inner(&object, now, 0).await.action,
-                            IlmAction::NoneAction,
-                            "legacy multi-stage configurations must not silently execute their first stage"
-                        );
-                        lc.rules[0]
-                            .transitions
-                            .as_mut()
-                            .expect("transition array is present")
-                            .reverse();
-                    }
-                }
-                lc.rules[0]
-                    .transitions
-                    .as_mut()
-                    .expect("transition array is present")
-                    .remove(0);
-                lc.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::ENABLED);
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("one stage is supported");
-                let event = lc.eval_inner(&object, now, 0).await;
-                assert_eq!(event.action, IlmAction::TransitionAction);
-                assert_eq!(event.storage_class, "WARM");
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn noncurrent_transition_rejects_multiple_stages_in_any_order() {
-            run(async {
-                let mut rule = enabled_rule(None, None, Some("two-noncurrent-transitions"));
-                rule.noncurrent_version_transitions = Some(vec![
-                    NoncurrentVersionTransition {
-                        noncurrent_days: Some(30),
-                        newer_noncurrent_versions: None,
-                        storage_class: Some(TransitionStorageClass::from_static("COLD")),
-                    },
-                    NoncurrentVersionTransition {
-                        noncurrent_days: Some(1),
-                        newer_noncurrent_versions: None,
-                        storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                    },
-                ]);
-                let mut lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                let object = noncurrent_object();
-                let now = datetime!(2020-01-10 00:00:00 UTC);
-                for status in [ExpirationStatus::ENABLED, ExpirationStatus::DISABLED] {
-                    lc.rules[0].status = ExpirationStatus::from_static(status);
-                    for _ in 0..2 {
-                        let err = lc
-                            .validate(&ObjectLockConfiguration::default())
-                            .await
-                            .expect_err("multiple noncurrent transition stages must be rejected");
-                        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
-                        assert_eq!(err.to_string(), ERR_LIFECYCLE_MULTIPLE_NONCURRENT_TRANSITIONS);
-                        assert_eq!(
-                            lc.eval_inner(&object, now, 0).await.action,
-                            IlmAction::NoneAction,
-                            "legacy multi-stage configurations must not silently execute their first stage"
-                        );
-                        lc.rules[0]
-                            .noncurrent_version_transitions
-                            .as_mut()
-                            .expect("transition array is present")
-                            .reverse();
-                    }
-                }
-                lc.rules[0]
-                    .noncurrent_version_transitions
-                    .as_mut()
-                    .expect("transition array is present")
-                    .remove(0);
-                lc.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::ENABLED);
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("one stage is supported");
-                let event = lc.eval_inner(&object, now, 0).await;
-                assert_eq!(event.action, IlmAction::TransitionVersionAction);
-                assert_eq!(event.storage_class, "WARM");
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn expiration_rejects_simultaneous_days_and_date() {
-            run(async {
-                let mut lc = BucketLifecycleConfiguration {
-                    rules: vec![enabled_rule(
-                        Some(LifecycleExpiration {
-                            days: Some(1),
-                            ..Default::default()
-                        }),
-                        None,
-                        Some("ambiguous-expiry"),
-                    )],
-                    expiry_updated_at: None,
-                };
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("a single Days expiration is valid");
-                lc.rules[0].expiration.as_mut().expect("expiration is present").date =
-                    Some(datetime!(2099-01-01 00:00:00 UTC).into());
-                let err = lc
-                    .validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect_err("Days and Date are mutually exclusive; accepting both silently overrides Days");
-                assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
-                assert_eq!(err.to_string(), ERR_LIFECYCLE_EXPIRATION_DAYS_DATE_CONFLICT);
-            });
-        }
-
-        #[test]
-        #[serial]
-        fn overdue_transition_does_not_starve_permanent_expiration() {
-            run(async {
-                let mut rule = enabled_rule(
-                    Some(LifecycleExpiration {
-                        days: Some(90),
-                        ..Default::default()
-                    }),
-                    None,
-                    Some("archive-then-delete"),
-                );
-                rule.transitions = Some(vec![Transition {
-                    days: Some(30),
-                    date: None,
-                    storage_class: Some(TransitionStorageClass::from_static("WARM")),
-                }]);
-                let lc = BucketLifecycleConfiguration {
-                    rules: vec![rule],
-                    expiry_updated_at: None,
-                };
-                lc.validate(&ObjectLockConfiguration::default())
-                    .await
-                    .expect("valid transition and expiration policy");
-                let object = ObjectOpts {
-                    is_latest: true,
-                    version_id: None,
-                    transition_status: TRANSITION_PENDING.to_string(),
-                    ..noncurrent_object()
-                };
-                let before_expiration = lc.eval_inner(&object, datetime!(2020-02-15 00:00:00 UTC), 0).await;
-                assert_eq!(before_expiration.action, IlmAction::TransitionAction);
-                let overdue = lc.eval_inner(&object, datetime!(2020-05-01 00:00:00 UTC), 0).await;
-                assert_eq!(
-                    overdue.action,
-                    IlmAction::DeleteAction,
-                    "an unavailable tier must not prevent permanent expiration indefinitely"
-                );
-            });
-        }
-    }
-
     /// Property-based tests for the rule evaluator (backlog#1148 ilm-14,
     /// follow-up to backlog#1030 / rustfs#4455).
     ///
@@ -4579,7 +4169,7 @@ mod tests {
     ///
     /// * `eval_inner` never panics and is deterministic for a fixed input;
     /// * the winning event matches an independently recomputed candidate set:
-    ///   eligible expiration wins over transition, then earliest `due` wins (the
+    ///   earliest `due` wins, ties break toward delete-class actions (the
     ///   `min_by_key` selection that replaced the rustfs#4455 comparator);
     /// * `expected_expiry_time` is monotonically non-decreasing in `days` and
     ///   always lands on the processing boundary, both at production defaults
@@ -4868,8 +4458,8 @@ mod tests {
         /// consider for a live current version under `selection`-shaped rules
         /// (expiration and first-transition only, no filters): expiration
         /// fires when `now >= due`, transition when `now > due` and the object
-        /// has not already transitioned. Eligible expiration wins over transition;
-        /// the earliest deadline wins within the selected action class.
+        /// has not already transitioned. Selection semantics under test:
+        /// earliest due wins, ties prefer delete-class.
         fn oracle_candidates(lc: &BucketLifecycleConfiguration, obj: &ObjectOpts, now: OffsetDateTime) -> Vec<Candidate> {
             let mod_time = obj.mod_time.expect("selection strategy always sets mod_time");
             let mut candidates = Vec::new();
@@ -4958,8 +4548,8 @@ mod tests {
             /// Differential test of winner selection (the rustfs#4455 fix):
             /// for a live current version under randomized expiration and
             /// transition rules, `eval_inner`'s winner must carry the
-            /// earliest expiration from the independently recomputed candidate
-            /// set, or the earliest transition when no expiration is eligible,
+            /// minimum `(due, rank)` of the independently recomputed
+            /// candidate set — earliest due wins, ties prefer delete-class —
             /// and must be `NoneAction` exactly when that set is empty.
             #[test]
             #[serial]
@@ -4988,13 +4578,7 @@ mod tests {
 
                 // Oracle and evaluator must observe the same (pinned) time env.
                 let (event, expected) = with_production_time_env(|| {
-                    let candidates = oracle_candidates(&lc, &obj, now);
-                    let expected = candidates
-                        .iter()
-                        .filter(|(_, rank)| *rank == 0)
-                        .min()
-                        .copied()
-                        .or_else(|| candidates.into_iter().min());
+                    let expected = oracle_candidates(&lc, &obj, now).into_iter().min();
                     let rt = tokio::runtime::Builder::new_current_thread()
                         .enable_all()
                         .build()

crates/lifecycle/src/evaluator.rs

@@ -116,10 +116,13 @@ impl Evaluator {
                             break 'top_loop;
                         }
                     }
-                    // Restore expiry removes only the temporary local copy; the
-                    // retained logical version and its remote data remain intact.
-                    IlmAction::DeleteAction | IlmAction::DeleteVersionAction if self.is_object_locked(obj) => {
-                        event = obj.restored_copy_expiry(now).unwrap_or_default();
+                    IlmAction::DeleteAction
+                    | IlmAction::DeleteRestoredAction
+                    | IlmAction::DeleteVersionAction
+                    | IlmAction::DeleteRestoredVersionAction
+                        if self.is_object_locked(obj) =>
+                    {
+                        event = Event::default();
                     }
                     _ => {}
                 }
@@ -203,95 +206,6 @@ mod tests {
 
     use super::*;
     use rustfs_replication::{ReplicationStatusType, VersionPurgeStatusType};
-
-    #[tokio::test]
-    async fn adversarial_restore_expiry_survives_legal_hold() {
-        let mut policy = (*latest_expiration_lifecycle()).clone();
-        policy.rules[0].status = ExpirationStatus::from_static(ExpirationStatus::DISABLED);
-        let policy = Arc::new(policy);
-        policy
-            .validate(&lock_enabled_without_default_retention())
-            .await
-            .expect("valid disabled lifecycle rule");
-        let mut objects = [true, false].map(|is_latest| ObjectOpts {
-            is_latest,
-            num_versions: 2,
-            mod_time: Some(
-                OffsetDateTime::from_unix_timestamp(if is_latest { 1_200_000 } else { 1_000_000 })
-                    .expect("fixed version timestamp"),
-            ),
-            successor_mod_time: (!is_latest)
-                .then(|| OffsetDateTime::from_unix_timestamp(1_200_000).expect("fixed successor timestamp")),
-            transition_status: crate::TRANSITION_COMPLETE.to_string(),
-            restore_expires: Some(OffsetDateTime::from_unix_timestamp(2_000_000).expect("fixed expired restore timestamp")),
-            ..current_object_opts(ReplicationStatusType::Completed)
-        });
-        let evaluator = Evaluator::new(policy).with_lock_retention(Some(lock_enabled_without_default_retention()));
-        let expected = [IlmAction::DeleteRestoredAction, IlmAction::DeleteRestoredVersionAction];
-        let unlocked = evaluator
-            .eval(&objects)
-            .await
-            .expect("unlocked restored versions should evaluate");
-        assert_eq!(unlocked.iter().map(|event| event.action).collect::<Vec<_>>(), expected);
-
-        for object in &mut objects {
-            object
-                .user_defined
-                .insert(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), "ON".to_string());
-        }
-        let locked = evaluator
-            .eval(&objects)
-            .await
-            .expect("locked restored versions should evaluate");
-        assert_eq!(
-            locked.iter().map(|event| event.action).collect::<Vec<_>>(),
-            expected,
-            "expiring a restored local copy preserves the retained logical version and remote object"
-        );
-
-        let mut expiring_policy = (*latest_expiration_lifecycle()).clone();
-        expiring_policy.rules[0].noncurrent_version_expiration = Some(NoncurrentVersionExpiration {
-            noncurrent_days: Some(1),
-            newer_noncurrent_versions: None,
-        });
-        let expiring_evaluator =
-            Evaluator::new(Arc::new(expiring_policy)).with_lock_retention(Some(lock_enabled_without_default_retention()));
-        let locked = expiring_evaluator
-            .eval(&objects)
-            .await
-            .expect("locked expired versions should evaluate");
-        assert_eq!(
-            locked.iter().map(|event| event.action).collect::<Vec<_>>(),
-            expected,
-            "blocked logical expiration must still allow an eligible restore-copy cleanup"
-        );
-
-        for status in [ReplicationStatusType::Pending, ReplicationStatusType::Failed] {
-            for object in &mut objects {
-                object.replication_status = status.clone();
-            }
-            for evaluator in [&evaluator, &expiring_evaluator] {
-                let events = evaluator.eval(&objects).await.expect("pending replication should evaluate");
-                assert!(events.iter().all(|event| event.action == IlmAction::NoneAction));
-            }
-        }
-        for object in &mut objects {
-            object.replication_status = ReplicationStatusType::Completed;
-        }
-        for transition_status in ["", crate::TRANSITION_PENDING, "unknown"] {
-            for object in &mut objects {
-                object.transition_status = transition_status.to_string();
-            }
-            for evaluator in [&evaluator, &expiring_evaluator] {
-                let events = evaluator.eval(&objects).await.expect("incomplete transition should evaluate");
-                assert!(
-                    events.iter().all(|event| event.action == IlmAction::NoneAction),
-                    "restore metadata cannot authorize cleanup without a completed transition"
-                );
-            }
-        }
-    }
-
     fn expired_marker_lifecycle() -> Arc<BucketLifecycleConfiguration> {
         Arc::new(BucketLifecycleConfiguration {
             expiry_updated_at: None,

docs/operations/tier-ilm-debugging.md

@@ -22,16 +22,6 @@
 | `FileMeta` / `FileInfo` / version metadata | `crates/filemeta/src/` |
 | Dual-key internal metadata helpers (`insert_bytes` / `get_bytes`) | `crates/utils/src/http/metadata_compat.rs` |
 
-## Lifecycle rule limits and evaluation
-
-Each lifecycle rule supports at most one `Transition` and one `NoncurrentVersionTransition`. A version can make one initial transition; chaining additional tiers after it reaches `complete` is not supported. Splitting stages across overlapping rules does not enable a transition chain. `PutBucketLifecycleConfiguration` rejects multiple entries in either transition array with `InvalidArgument`, including in disabled rules. Existing stored multi-entry arrays are not executed; replace each with a single intended destination. Independent expiration actions in the rule remain eligible.
-
-`Expiration.Days` and `Expiration.Date` are mutually exclusive. A request containing both is rejected instead of silently selecting the date. When expiration and transition are both eligible, expiration takes precedence; a failed earlier transition does not keep an expired object indefinitely. Deadlines select the earliest action within the same action class.
-
-Noncurrent expiration and transition have independent `NewerNoncurrentVersions` limits. A transition with a positive limit waits for a complete version-group evaluation to establish that enough newer noncurrent versions remain. Single-object evaluation, including the current manual transition and immediate-enqueue paths, conservatively defers these counted transitions to the lifecycle scanner. An unmet expiration retention limit does not suppress a separately eligible transition.
-
-An expired restored local copy can be cleaned up under Object Lock because the retained logical version and remote data remain intact. Cleanup requires a completed transition and still waits for pending or failed replication. The storage layer revalidates the source identity and restore metadata before removing the local copy; restore headers alone do not authorize cleanup.
-
 ## Free-version recovery controls
 
 The dedicated free-version recovery loop is enabled by default and is independent of the data scanner and heal switches. Setting `RUSTFS_SCANNER_ENABLED=false` does not stop this repair loop. Set `RUSTFS_TIER_FREE_VERSION_RECOVERY_ENABLED=false` before process startup to disable only the dedicated persisted-marker walk. That setting does not disable lifecycle workers or prevent another scanner path from discovering a free version, and it can leave remote cleanup markers pending for longer, so use it as a break-glass pressure control rather than a cleanup mechanism.