fix(ecstore): invalidate metadata cache after ILM transition persists (#4951)

A duplicate transition task admitted after the winner released its
in-flight claim (#4839) re-reads the version before uploading, but on
unversioned buckets that read could hit a stale pre-transition entry in
the 2s-TTL GET metadata cache: transition_object never invalidated the
cache after delete_object_version persisted transition_status=complete
and freed the local data. The stale hit defeated the TRANSITION_COMPLETE
early-return, so the duplicate streamed the already-deleted local data
to the remote tier (NotFound reader errors + rejected duplicate tier
PUT with UnexpectedContent).

Invalidate the cache right after the transitioned metadata is persisted,
matching the other metadata-mutating paths, and add a regression test
that runs a duplicate transition against an already-transitioned version
and asserts no second tier upload and unchanged remote object metadata.

Fixes #4827

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

@@ -2354,6 +2354,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
         }
 
+        // delete_object_version persisted transition_status=complete and freed the
+        // local data, but does not touch the GET metadata cache. Drop any cached
+        // pre-transition entry so a late duplicate transition task (or a plain GET)
+        // re-reads the fresh state; a stale hit here defeats the TRANSITION_COMPLETE
+        // early-return above and streams the already-deleted local data to the
+        // remote tier again (rustfs/rustfs#4827).
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+
         for disk in disks.iter() {
             if let Some(disk) = disk {
                 continue;

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -25,7 +25,7 @@ use super::storage_api::test::contract::{
 };
 use super::storage_api::test::ecfs::FS;
 use super::storage_api::test::object_utils::to_s3s_etag;
-use super::storage_api::test::runtime::{MockWarmBackend, register_mock_tier as register_mock_tier_util};
+use super::storage_api::test::runtime::{MockWarmBackend, MockWarmOp, register_mock_tier as register_mock_tier_util};
 use super::storage_api::test::{
     ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, StorageObjectInfo as ObjectInfo,
     StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader,
@@ -903,6 +903,100 @@ async fn get_transitioned_object_uses_remote_codec_fallback_path() {
     .await;
 }
 
+/// Regression test for rustfs/rustfs#4827: a duplicate transition task that runs
+/// after the winning transition completed must return early without a second
+/// tier upload and without clobbering the winner's remote object / metadata.
+///
+/// The duplicate models the residual race left after the in-flight enqueue dedup
+/// (#4839): a stale-snapshot task admitted after the winner released its claim.
+/// On an unversioned bucket the duplicate's metadata re-read is eligible for the
+/// GET metadata cache; before the fix the cache still held the pre-transition
+/// entry (transition_object never invalidated it after persisting), so the
+/// TRANSITION_COMPLETE early-return never fired and the duplicate streamed the
+/// already-deleted local data to the tier again (NotFound reader errors +
+/// rejected duplicate tier PUT).
+#[tokio::test(flavor = "multi_thread", worker_threads = 1)]
+#[serial]
+#[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]
+async fn duplicate_transition_task_skips_already_transitioned_version() {
+    let (_disk_paths, ecstore) = setup_test_env().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 = format!("test-api-dup-transition-{}", &Uuid::new_v4().simple().to_string()[..8]);
+    let object = "test/duplicate-transition.txt";
+    let payload = b"duplicate transition tasks must not re-upload";
+
+    // Unversioned bucket: the duplicate's re-read goes through the GET metadata
+    // cache (versioned reads bypass it), which is the stale-read window under test.
+    (*ecstore)
+        .make_bucket(bucket.as_str(), &MakeBucketOptions::default())
+        .await
+        .expect("Failed to create unversioned test bucket");
+
+    let uploaded = upload_test_object(&ecstore, bucket.as_str(), object, payload).await;
+    let transition_opts = ObjectOptions {
+        transition: lifecycle::lifecycle_contract::TransitionOptions {
+            status: lifecycle::lifecycle_contract::TRANSITION_PENDING.to_string(),
+            tier: tier_name.clone(),
+            etag: uploaded.etag.clone().unwrap_or_default(),
+            ..Default::default()
+        },
+        version_id: uploaded.version_id.map(|version| version.to_string()),
+        mod_time: uploaded.mod_time,
+        ..Default::default()
+    };
+
+    // Winner: transitions the object, persists transition_status=complete, and
+    // deletes the local data.
+    ecstore
+        .transition_object(bucket.as_str(), object, &transition_opts)
+        .await
+        .expect("winning transition should succeed");
+
+    let puts_after_winner: Vec<String> = backend
+        .op_log()
+        .await
+        .into_iter()
+        .filter_map(|op| match op {
+            MockWarmOp::Put { object } => Some(object),
+            _ => None,
+        })
+        .collect();
+    assert_eq!(puts_after_winner.len(), 1, "winner must upload exactly one tier object");
+    let winner_remote_object = puts_after_winner[0].clone();
+
+    // Duplicate task with the same stale snapshot (same mod_time / etag), run
+    // immediately so the pre-fix stale metadata-cache entry is still live. It
+    // must observe transition_status=complete and return Ok without uploading.
+    ecstore
+        .transition_object(bucket.as_str(), object, &transition_opts)
+        .await
+        .expect("duplicate transition must return early without error");
+
+    let puts_after_duplicate = backend
+        .op_log()
+        .await
+        .into_iter()
+        .filter(|op| matches!(op, MockWarmOp::Put { .. }))
+        .count();
+    assert_eq!(puts_after_duplicate, 1, "duplicate transition must not attempt a second tier upload");
+
+    // The winner's remote object and local metadata must be intact.
+    assert!(backend.contains(&winner_remote_object).await);
+    let info = wait_for_transition(&ecstore, bucket.as_str(), object, TRANSITION_WAIT_TIMEOUT)
+        .await
+        .expect("object must remain transitioned after duplicate task");
+    assert_eq!(info.transitioned_object.status, "complete");
+    assert_eq!(info.transitioned_object.tier, tier_name);
+    assert_eq!(
+        info.transitioned_object.name, winner_remote_object,
+        "duplicate transition must not repoint metadata at a new remote object"
+    );
+    assert_eq!(read_object_bytes(&ecstore, bucket.as_str(), object).await, payload);
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
 #[serial]
 #[ignore = "global-state ILM integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"]

rustfs/src/app/storage_api.rs

@@ -138,7 +138,7 @@ pub(crate) mod runtime {
     // The mock (and the tier types it used to need) now live in ecstore behind
     // the `test-util` feature.
     #[cfg(test)]
-    pub(crate) use crate::storage::storage_api::ecstore_tier::test_util::{MockWarmBackend, register_mock_tier};
+    pub(crate) use crate::storage::storage_api::ecstore_tier::test_util::{MockWarmBackend, MockWarmOp, register_mock_tier};
 
     pub(crate) fn set_global_storage_class(cfg: StorageClassConfig) {
         crate::storage::storage_api::ecstore_config::set_global_storage_class(cfg);