fix(object-lock): read cleared empty lock metadata as unlocked

A local PutObjectRetention / PutObjectLegalHold clear persists the lock
keys as empty strings (MinIO on-disk shape). The commit-time WORM gate and
the delete gate parsed the empty mode / legal-hold value as corrupt
metadata, so every later explicit-version PUT (the inbound replication
transport) and version delete of that version failed. Treat empty values
as absent; non-empty malformed values still fail closed.

Refs rustfs/backlog#1953

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -239,7 +239,12 @@ pub(crate) fn check_object_lock_for_deletion_with_config(
         return Ok(None);
     }
 
-    if let Some(status) = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) {
+    // A cleared retention / legal hold is persisted as empty strings (the
+    // MinIO on-disk shape, `parse_object_lock_retention`); read it as "no lock"
+    // rather than as corrupt metadata.
+    let persisted = |key: &str| obj_info.user_defined.get(key).filter(|value| !value.is_empty());
+
+    if let Some(status) = persisted(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) {
         if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) {
             return Ok(Some(ObjectLockBlockReason::LegalHold));
         }
@@ -248,8 +253,8 @@ pub(crate) fn check_object_lock_for_deletion_with_config(
         }
     }
 
-    let mode = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_MODE.as_str());
-    let retain_until = obj_info.user_defined.get(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
+    let mode = persisted(X_AMZ_OBJECT_LOCK_MODE.as_str());
+    let retain_until = persisted(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
     let explicit_ret = match (mode, retain_until) {
         (None, None) => None,
         (Some(mode), Some(retain_until)) => {
@@ -486,6 +491,45 @@ mod tests {
         }
     }
 
+    /// A local PutObjectRetention / PutObjectLegalHold "clear" persists the
+    /// lock keys as empty strings (the MinIO on-disk shape, see
+    /// `parse_object_lock_retention`); that is "no lock", not corruption, and
+    /// must not wedge later explicit-version PUTs or deletes
+    /// (rustfs/backlog#1953).
+    #[test]
+    fn deletion_treats_cleared_empty_lock_metadata_as_unlocked() {
+        use rustfs_utils::http::headers::{
+            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+        };
+
+        let cases: [(&str, &[&str]); 3] = [
+            (
+                "cleared retention",
+                &[AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER],
+            ),
+            ("cleared legal hold", &[AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER]),
+            (
+                "all cleared",
+                &[
+                    AMZ_OBJECT_LOCK_MODE_LOWER,
+                    AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+                    AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER,
+                ],
+            ),
+        ];
+
+        for (case, keys) in cases {
+            let user_defined = keys.iter().map(|key| (key.to_string(), String::new())).collect();
+            let obj_info = ObjectInfo {
+                user_defined: Arc::new(user_defined),
+                ..Default::default()
+            };
+
+            let result = check_object_lock_for_deletion_with_config(None, &obj_info, false);
+            assert!(matches!(result, Ok(None)), "{case}: empty lock keys must read as unlocked: {result:?}");
+        }
+    }
+
     #[test]
     fn deletion_rejects_invalid_persisted_legal_hold_metadata() {
         let mut user_defined = std::collections::HashMap::new();

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

@@ -14344,6 +14344,72 @@ mod put_object_tmp_cleanup_tests {
         assert_eq!(body, original_body);
     }
 
+    /// A local PutObjectRetention / PutObjectLegalHold clear persists empty
+    /// lock keys (`parse_object_lock_retention`). The commit-time WORM gate
+    /// must read that as unlocked: an explicit-version PUT (the inbound
+    /// replication transport) and a version delete both have to succeed
+    /// (rustfs/backlog#1953).
+    #[tokio::test]
+    async fn explicit_version_overwrite_and_delete_succeed_after_local_lock_clear() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "put-explicit-version-cleared-lock";
+        let object = "object";
+        for disk in &disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+
+        let mut initial_reader = PutObjReader::from_vec(b"original".to_vec());
+        let initial = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut initial_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("initial version should be written");
+        let version_id = initial
+            .version_id
+            .expect("versioned PUT should return a version ID")
+            .to_string();
+        let version_opts = ObjectOptions {
+            versioned: true,
+            version_id: Some(version_id.clone()),
+            delete_replication_config_snapshot: Some(Arc::new(DeleteReplicationConfigSnapshot::default())),
+            object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(ObjectLockConfigState::ConfirmedAbsent))),
+            ..Default::default()
+        };
+        set_disks
+            .put_object_metadata(
+                bucket,
+                object,
+                &ObjectOptions {
+                    eval_metadata: Some(HashMap::from([
+                        (X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(), String::new()),
+                        (X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(), String::new()),
+                        (X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), String::new()),
+                    ])),
+                    ..version_opts.clone()
+                },
+            )
+            .await
+            .expect("cleared lock metadata should be written");
+
+        let mut replacement = PutObjReader::from_vec(b"replacement".to_vec());
+        set_disks
+            .put_object(bucket, object, &mut replacement, &version_opts)
+            .await
+            .expect("explicit-version PUT must not be wedged by cleared lock metadata");
+
+        set_disks
+            .delete_object(bucket, object, version_opts)
+            .await
+            .expect("version delete must not be wedged by cleared lock metadata");
+    }
+
     #[tokio::test]
     async fn version_only_copy_checks_the_destination_version_object_lock() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;