refactor(ecstore): move object_lock WORM evaluation onto storage-level types (#6666)

The object_lock module evaluated WORM state through s3s wire DTOs (ObjectLockRetention, ObjectLockLegalHold, DefaultRetention, Date) and s3s header constants, keeping the storage engine coupled to the serving protocol (rustfs/backlog#1842, ARCHITECTURE.md invariant 4). This PR gives the module its own storage-level vocabulary and pushes the DTO conversions to the boundaries that already speak s3s.

New crates/ecstore/src/bucket/object_lock/types.rs defines RetentionMode, LegalHoldStatus, ObjectRetention, ObjectLegalHold, and DefaultRetention with no s3s dependency. objectlock.rs parses persisted metadata into these types using the rustfs-utils lowercase header constants (the same literal keys as before, pinned by the existing g-key-002 test). objectlock_sys.rs evaluates retention/legal-hold/default-retention from them; the fail-closed error messages and decision logic are unchanged line for line where possible.

Boundary conversions:
- bucket/metadata_sys.rs gains default_retention_from_object_lock_config, converting the persisted s3s configuration into the storage-level DefaultRetention; a rule without a usable GOVERNANCE/COMPLIANCE mode converts to None exactly like the evaluation code always ignored it, and days/years pass through so an invalid period still fails closed at evaluation time.
- check_object_lock_for_deletion_with_config becomes check_object_lock_for_deletion_with_default_retention (it only ever read the default retention); the lifecycle object_lock_boundary keeps the old s3s-typed signature and converts.
- The ObjectLockApi / ObjectLockStatusExt trait impls for the s3s DTOs move next to the persisted configuration owner in bucket/metadata.rs; the traits stay in object_lock/mod.rs.
- check_retention_for_modification now takes Option<RetentionMode>. The serving-layer wrappers (rustfs storage_api, set_disk options path) convert the request string with the new RetentionMode::parse_exact, which accepts only the canonical spelling — preserving the historical literal comparison where a non-canonical requested mode reads as a mode change and stays blocked.
- rustfs app-layer wrappers return the storage types; the replication-overwrite gate in object_usecase.rs uses the typed API (legal_hold.is_on(), RetentionMode::Compliance).

Ratchet: the ecstore-scoped s3s counter drops 42 -> 39 and the repo-wide file counter 211 -> 208 in scripts/check_s3s_footprint.sh.

Verification: cargo check -p rustfs-ecstore --all-targets and -p rustfs (lib+bins); cargo clippy -p rustfs-ecstore --all-targets and -p rustfs --lib --bins (clean); cargo nextest run -p rustfs-ecstore --no-fail-fast (4534/4542; the 8 failures are the same store::rebalance / store::heal machine-baseline set that fails identically on pristine origin/main, plus one fencing flake that passes in isolation); all object_lock/retention/legal-hold tests pass; guard scripts (layer deps, migration rules, s3s footprint, logging, error-format ratchet, doc paths) pass.

crates/ecstore/src/api/mod.rs

@@ -154,6 +154,12 @@ pub mod bucket {
     pub mod object_lock {
         pub use crate::bucket::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 
+        pub mod types {
+            pub use crate::bucket::object_lock::types::{
+                DefaultRetention, LegalHoldStatus, ObjectLegalHold, ObjectRetention, RetentionMode,
+            };
+        }
+
         pub mod objectlock {
             pub use crate::bucket::object_lock::objectlock::{get_object_legalhold_meta, get_object_retention_meta};
         }

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

@@ -26,7 +26,8 @@ pub(crate) fn check_object_lock_for_deletion_with_config(
     obj_info: &ObjectInfo,
     bypass_governance: bool,
 ) -> crate::error::Result<Option<ObjectLockBlockReason>> {
-    objectlock_sys::check_object_lock_for_deletion_with_config(config, obj_info, bypass_governance)
+    let default_retention = config.and_then(crate::bucket::metadata_sys::default_retention_from_object_lock_config);
+    objectlock_sys::check_object_lock_for_deletion_with_default_retention(default_retention.as_ref(), obj_info, bypass_governance)
 }
 
 #[cfg(test)]

crates/ecstore/src/bucket/metadata.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::msgp_decode::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time};
-use super::object_lock::ObjectLockApi;
+use super::object_lock::{ObjectLockApi, ObjectLockStatusExt};
 use super::versioning::VersioningApi;
 use super::{quota::BucketQuota, target::BucketTargets};
 use crate::bucket::replication::invalid_replication_config_status_field;
@@ -39,6 +39,26 @@ use time::{Date, OffsetDateTime, PrimitiveDateTime, Time as CivilTime, UtcOffset
 use tracing::error;
 use uuid::Uuid;
 
+// The serving-layer DTO impls for the storage-level Object Lock traits live
+// here because this module owns the persisted `ObjectLockConfiguration`
+// during the s3s ratchet migration (rustfs/backlog#1842).
+impl ObjectLockApi for ObjectLockConfiguration {
+    fn enabled(&self) -> bool {
+        self.object_lock_enabled
+            .as_ref()
+            .is_some_and(|v| v.as_str() == s3s::dto::ObjectLockEnabled::ENABLED)
+    }
+}
+
+impl ObjectLockStatusExt for s3s::dto::ObjectLockLegalHoldStatus {
+    fn valid(&self) -> bool {
+        matches!(
+            self.as_str(),
+            s3s::dto::ObjectLockLegalHoldStatus::ON | s3s::dto::ObjectLockLegalHoldStatus::OFF
+        )
+    }
+}
+
 fn read_msgp_str<R: Read>(rd: &mut R) -> Result<String> {
     let len = rmp::decode::read_str_len(rd)? as usize;
     let mut buf = vec![0u8; len];

crates/ecstore/src/bucket/metadata_sys.rs

@@ -167,6 +167,53 @@ pub(crate) fn object_lock_config_state_from_authoritative_metadata(bm: &BucketMe
     Ok(ObjectLockConfigState::ConfirmedAbsent)
 }
 
+/// Convert the persisted serving-layer configuration into the storage-level
+/// [`DefaultRetention`](crate::bucket::object_lock::types::DefaultRetention)
+/// the WORM evaluation code consumes (rustfs/backlog#1842). A rule without a
+/// usable GOVERNANCE/COMPLIANCE mode converts to `None`, exactly like the
+/// evaluation code has always ignored such rules; days/years are passed
+/// through untouched so an invalid period still fails closed at evaluation.
+pub(crate) fn default_retention_from_object_lock_config(
+    config: &ObjectLockConfiguration,
+) -> Option<crate::bucket::object_lock::types::DefaultRetention> {
+    let default_retention = config.rule.as_ref()?.default_retention.as_ref()?;
+    let mode = crate::bucket::object_lock::types::RetentionMode::parse(default_retention.mode.as_ref()?.as_str())?;
+    Some(crate::bucket::object_lock::types::DefaultRetention {
+        mode,
+        days: default_retention.days,
+        years: default_retention.years,
+    })
+}
+
+/// Test-only builder for a `Configured` Object Lock state carrying a default
+/// retention, so storage-side tests do not have to name serving-layer DTOs.
+#[cfg(test)]
+pub(crate) fn configured_object_lock_state_for_tests(
+    mode: crate::bucket::object_lock::types::RetentionMode,
+    days: i32,
+) -> ObjectLockConfigState {
+    ObjectLockConfigState::Configured {
+        config: ObjectLockConfiguration {
+            object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
+            rule: Some(s3s::dto::ObjectLockRule {
+                default_retention: Some(s3s::dto::DefaultRetention {
+                    mode: Some(s3s::dto::ObjectLockRetentionMode::from_static(match mode {
+                        crate::bucket::object_lock::types::RetentionMode::Governance => {
+                            s3s::dto::ObjectLockRetentionMode::GOVERNANCE
+                        }
+                        crate::bucket::object_lock::types::RetentionMode::Compliance => {
+                            s3s::dto::ObjectLockRetentionMode::COMPLIANCE
+                        }
+                    })),
+                    days: Some(days),
+                    years: None,
+                }),
+            }),
+        },
+        updated_at: OffsetDateTime::now_utc(),
+    }
+}
+
 fn validate_authoritative_object_lock_config(config: &ObjectLockConfiguration) -> Result<()> {
     if config.object_lock_enabled.as_ref().map(ObjectLockEnabled::as_str) != Some(ObjectLockEnabled::ENABLED) {
         return Err(Error::other("persisted bucket Object Lock enabled state is invalid"));

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

@@ -14,27 +14,18 @@
 
 pub mod objectlock;
 pub mod objectlock_sys;
+pub mod types;
 
-use s3s::dto::{ObjectLockConfiguration, ObjectLockEnabled, ObjectLockLegalHoldStatus};
-
+/// Whether a bucket Object Lock configuration has locking enabled. The
+/// serving-layer `ObjectLockConfiguration` DTO implements this in
+/// the bucket-metadata module, which owns the persisted configuration type
+/// during the s3s ratchet migration (rustfs/backlog#1842).
 pub trait ObjectLockApi {
     fn enabled(&self) -> bool;
 }
 
-impl ObjectLockApi for ObjectLockConfiguration {
-    fn enabled(&self) -> bool {
-        self.object_lock_enabled
-            .as_ref()
-            .is_some_and(|v| v.as_str() == ObjectLockEnabled::ENABLED)
-    }
-}
-
+/// Whether a legal-hold status value is one of the two valid wire values.
+/// Implemented for the serving-layer DTO in the bucket-metadata module.
 pub trait ObjectLockStatusExt {
     fn valid(&self) -> bool;
 }
-
-impl ObjectLockStatusExt for ObjectLockLegalHoldStatus {
-    fn valid(&self) -> bool {
-        matches!(self.as_str(), ObjectLockLegalHoldStatus::ON | ObjectLockLegalHoldStatus::OFF)
-    }
-}

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

@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use s3s::dto::{Date, ObjectLockLegalHold, ObjectLockLegalHoldStatus, ObjectLockRetention, ObjectLockRetentionMode};
-use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
+use super::types::{LegalHoldStatus, ObjectLegalHold, ObjectRetention, RetentionMode};
+use rustfs_utils::http::headers::{
+    AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+};
 use std::collections::HashMap;
 use time::{OffsetDateTime, format_description};
 
@@ -31,65 +33,48 @@ pub fn utc_now_ntp() -> OffsetDateTime {
     OffsetDateTime::now_utc()
 }
 
-pub fn get_object_retention_meta(meta: &HashMap<String, String>) -> ObjectLockRetention {
-    // Note: X_AMZ_OBJECT_LOCK_MODE.as_str() is already lowercase ("x-amz-object-lock-mode")
-    let mode_str = meta.get(X_AMZ_OBJECT_LOCK_MODE.as_str());
+pub fn get_object_retention_meta(meta: &HashMap<String, String>) -> ObjectRetention {
+    // The persisted metadata keys are the lowercase wire header names.
+    let mode_str = meta.get(AMZ_OBJECT_LOCK_MODE_LOWER);
 
     let Some(mode_str) = mode_str else {
-        return ObjectLockRetention {
-            mode: None,
-            retain_until_date: None,
-        };
+        return ObjectRetention::default();
     };
 
     // If mode is invalid, return empty retention (don't panic)
     let Some(mode) = parse_ret_mode(mode_str.as_str()) else {
-        return ObjectLockRetention {
-            mode: None,
-            retain_until_date: None,
-        };
+        return ObjectRetention::default();
     };
 
-    let till_str = meta.get(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
+    let till_str = meta.get(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER);
 
-    let retain_until_date = till_str
-        .and_then(|s| OffsetDateTime::parse(s, &format_description::well_known::Iso8601::DEFAULT).ok())
-        .map(Date::from);
+    let retain_until_date =
+        till_str.and_then(|s| OffsetDateTime::parse(s, &format_description::well_known::Iso8601::DEFAULT).ok());
 
-    ObjectLockRetention {
+    ObjectRetention {
         mode: Some(mode),
         retain_until_date,
     }
 }
 
-pub fn get_object_legalhold_meta(meta: &HashMap<String, String>) -> ObjectLockLegalHold {
-    // Note: X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str() is already lowercase
-    let hold_str = meta.get(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str());
+pub fn get_object_legalhold_meta(meta: &HashMap<String, String>) -> ObjectLegalHold {
+    let hold_str = meta.get(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER);
 
-    match hold_str.and_then(|s| parse_legalhold_status(s)) {
-        Some(status) => ObjectLockLegalHold { status: Some(status) },
-        None => ObjectLockLegalHold { status: None },
+    ObjectLegalHold {
+        status: hold_str.and_then(|s| parse_legalhold_status(s)),
     }
 }
 
-/// Parse retention mode string into ObjectLockRetentionMode.
+/// Parse retention mode string into [`RetentionMode`].
 /// Returns None for invalid/unknown mode strings instead of panicking.
-pub fn parse_ret_mode(mode_str: &str) -> Option<ObjectLockRetentionMode> {
-    match mode_str.to_uppercase().as_str() {
-        "GOVERNANCE" => Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::GOVERNANCE)),
-        "COMPLIANCE" => Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)),
-        _ => None,
-    }
+pub fn parse_ret_mode(mode_str: &str) -> Option<RetentionMode> {
+    RetentionMode::parse(mode_str)
 }
 
-/// Parse legal hold status string into ObjectLockLegalHoldStatus.
+/// Parse legal hold status string into [`LegalHoldStatus`].
 /// Returns None for invalid/unknown status strings instead of panicking.
-pub fn parse_legalhold_status(hold_str: &str) -> Option<ObjectLockLegalHoldStatus> {
-    match hold_str.to_uppercase().as_str() {
-        "ON" => Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)),
-        "OFF" => Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::OFF)),
-        _ => None,
-    }
+pub fn parse_legalhold_status(hold_str: &str) -> Option<LegalHoldStatus> {
+    LegalHoldStatus::parse(hold_str)
 }
 
 #[cfg(test)]
@@ -101,25 +86,25 @@ mod tests {
         // Test uppercase
         let mode = parse_ret_mode("GOVERNANCE");
         assert!(mode.is_some());
-        assert_eq!(mode.unwrap().as_str(), ObjectLockRetentionMode::GOVERNANCE);
+        assert_eq!(mode.unwrap().as_str(), RetentionMode::GOVERNANCE);
 
         let mode = parse_ret_mode("COMPLIANCE");
         assert!(mode.is_some());
-        assert_eq!(mode.unwrap().as_str(), ObjectLockRetentionMode::COMPLIANCE);
+        assert_eq!(mode.unwrap().as_str(), RetentionMode::COMPLIANCE);
 
         // Test lowercase
         let mode = parse_ret_mode("governance");
         assert!(mode.is_some());
-        assert_eq!(mode.unwrap().as_str(), ObjectLockRetentionMode::GOVERNANCE);
+        assert_eq!(mode.unwrap().as_str(), RetentionMode::GOVERNANCE);
 
         let mode = parse_ret_mode("compliance");
         assert!(mode.is_some());
-        assert_eq!(mode.unwrap().as_str(), ObjectLockRetentionMode::COMPLIANCE);
+        assert_eq!(mode.unwrap().as_str(), RetentionMode::COMPLIANCE);
 
         // Test mixed case
         let mode = parse_ret_mode("Governance");
         assert!(mode.is_some());
-        assert_eq!(mode.unwrap().as_str(), ObjectLockRetentionMode::GOVERNANCE);
+        assert_eq!(mode.unwrap().as_str(), RetentionMode::GOVERNANCE);
     }
 
     #[test]
@@ -136,20 +121,20 @@ mod tests {
         // Test uppercase
         let status = parse_legalhold_status("ON");
         assert!(status.is_some());
-        assert_eq!(status.unwrap().as_str(), ObjectLockLegalHoldStatus::ON);
+        assert_eq!(status.unwrap().as_str(), LegalHoldStatus::ON);
 
         let status = parse_legalhold_status("OFF");
         assert!(status.is_some());
-        assert_eq!(status.unwrap().as_str(), ObjectLockLegalHoldStatus::OFF);
+        assert_eq!(status.unwrap().as_str(), LegalHoldStatus::OFF);
 
         // Test lowercase
         let status = parse_legalhold_status("on");
         assert!(status.is_some());
-        assert_eq!(status.unwrap().as_str(), ObjectLockLegalHoldStatus::ON);
+        assert_eq!(status.unwrap().as_str(), LegalHoldStatus::ON);
 
         let status = parse_legalhold_status("off");
         assert!(status.is_some());
-        assert_eq!(status.unwrap().as_str(), ObjectLockLegalHoldStatus::OFF);
+        assert_eq!(status.unwrap().as_str(), LegalHoldStatus::OFF);
     }
 
     #[test]
@@ -175,7 +160,7 @@ mod tests {
         meta.insert("x-amz-object-lock-mode".to_string(), "GOVERNANCE".to_string());
         let retention = get_object_retention_meta(&meta);
         assert!(retention.mode.is_some());
-        assert_eq!(retention.mode.unwrap().as_str(), ObjectLockRetentionMode::GOVERNANCE);
+        assert_eq!(retention.mode.unwrap().as_str(), RetentionMode::GOVERNANCE);
         assert!(retention.retain_until_date.is_none());
     }
 
@@ -196,7 +181,7 @@ mod tests {
         meta.insert("x-amz-object-lock-retain-until-date".to_string(), "2030-01-01T00:00:00Z".to_string());
         let retention = get_object_retention_meta(&meta);
         assert!(retention.mode.is_some());
-        assert_eq!(retention.mode.unwrap().as_str(), ObjectLockRetentionMode::COMPLIANCE);
+        assert_eq!(retention.mode.unwrap().as_str(), RetentionMode::COMPLIANCE);
         assert!(retention.retain_until_date.is_some());
     }
 
@@ -210,17 +195,11 @@ mod tests {
         meta.insert("x-amz-object-lock-legal-hold".to_string(), "ON".to_string());
 
         let retention = get_object_retention_meta(&meta);
-        assert_eq!(
-            retention.mode.as_ref().map(|mode| mode.as_str()),
-            Some(ObjectLockRetentionMode::COMPLIANCE)
-        );
+        assert_eq!(retention.mode.as_ref().map(|mode| mode.as_str()), Some(RetentionMode::COMPLIANCE));
         assert!(retention.retain_until_date.is_some(), "persisted retention date must remain readable");
 
         let legal_hold = get_object_legalhold_meta(&meta);
-        assert_eq!(
-            legal_hold.status.as_ref().map(|status| status.as_str()),
-            Some(ObjectLockLegalHoldStatus::ON)
-        );
+        assert_eq!(legal_hold.status.as_ref().map(|status| status.as_str()), Some(LegalHoldStatus::ON));
     }
 
     #[test]
@@ -236,7 +215,7 @@ mod tests {
         meta.insert("x-amz-object-lock-legal-hold".to_string(), "ON".to_string());
         let legalhold = get_object_legalhold_meta(&meta);
         assert!(legalhold.status.is_some());
-        assert_eq!(legalhold.status.unwrap().as_str(), ObjectLockLegalHoldStatus::ON);
+        assert_eq!(legalhold.status.unwrap().as_str(), LegalHoldStatus::ON);
     }
 
     #[test]
@@ -245,7 +224,7 @@ mod tests {
         meta.insert("x-amz-object-lock-legal-hold".to_string(), "OFF".to_string());
         let legalhold = get_object_legalhold_meta(&meta);
         assert!(legalhold.status.is_some());
-        assert_eq!(legalhold.status.unwrap().as_str(), ObjectLockLegalHoldStatus::OFF);
+        assert_eq!(legalhold.status.unwrap().as_str(), LegalHoldStatus::OFF);
     }
 
     #[test]

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

@@ -12,12 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::bucket::metadata_sys::{ObjectLockConfigState, get_object_lock_config, get_object_lock_config_state};
+use crate::bucket::metadata_sys::{
+    ObjectLockConfigState, default_retention_from_object_lock_config, get_object_lock_config, get_object_lock_config_state,
+};
 use crate::bucket::object_lock::objectlock;
+use crate::bucket::object_lock::types::{DefaultRetention, LegalHoldStatus, RetentionMode};
 use crate::error::{Error, Result, StorageError};
 use crate::object_api::{ObjectInfo, ObjectOptions};
-use s3s::dto::{Date, DefaultRetention, ObjectLockConfiguration, ObjectLockLegalHoldStatus, ObjectLockRetentionMode};
-use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
+use rustfs_utils::http::headers::{
+    AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+};
 use std::sync::Arc;
 use time::OffsetDateTime;
 
@@ -29,11 +33,12 @@ impl BucketObjectLockSys {
         Arc::new(Self {})
     }
 
+    /// The bucket's active default retention, if the bucket has an
+    /// authoritative Object Lock configuration with a usable
+    /// GOVERNANCE/COMPLIANCE default retention rule.
     pub async fn get(bucket: &str) -> Option<DefaultRetention> {
-        if let Ok(object_lock_config) = get_object_lock_config(bucket).await
-            && let Some(object_lock_rule) = object_lock_config.0.rule
-        {
-            return object_lock_rule.default_retention;
+        if let Ok(object_lock_config) = get_object_lock_config(bucket).await {
+            return default_retention_from_object_lock_config(&object_lock_config.0);
         }
         None
     }
@@ -54,13 +59,10 @@ pub(crate) fn ensure_recursive_force_delete_allowed_for_state(bucket: &str, stat
 }
 
 /// Check if a retention period is still active based on mode and retain_until_date
-pub fn is_retention_active(mode: &str, retain_until_date: Option<&s3s::dto::Date>) -> bool {
-    if mode != ObjectLockRetentionMode::COMPLIANCE && mode != ObjectLockRetentionMode::GOVERNANCE {
-        return false;
-    }
+pub fn is_retention_active(_mode: RetentionMode, retain_until_date: Option<OffsetDateTime>) -> bool {
     if let Some(retain_until) = retain_until_date {
         let now = objectlock::utc_now_ntp();
-        return OffsetDateTime::from(retain_until.clone()).unix_timestamp() > now.unix_timestamp();
+        return retain_until.unix_timestamp() > now.unix_timestamp();
     }
     false
 }
@@ -68,23 +70,20 @@ pub fn is_retention_active(mode: &str, retain_until_date: Option<&s3s::dto::Date
 /// Check if retention modification is blocked for the given object.
 pub fn check_retention_for_modification(
     user_defined: &std::collections::HashMap<String, String>,
-    new_mode: Option<&str>,
+    new_mode: Option<RetentionMode>,
     new_retain_until: Option<OffsetDateTime>,
     bypass_governance: bool,
 ) -> Option<ObjectLockBlockReason> {
     let retention = objectlock::get_object_retention_meta(user_defined);
 
-    let Some(mode) = &retention.mode else {
-        return None;
-    };
+    let mode = retention.mode?;
 
-    let mode_str = mode.as_str();
-    if !is_retention_active(mode_str, retention.retain_until_date.as_ref()) {
+    if !is_retention_active(mode, retention.retain_until_date) {
         return None;
     }
 
-    let existing_retain_until = retention.retain_until_date.as_ref().map(|d| OffsetDateTime::from(d.clone()));
-    let mode_changed = new_mode != Some(mode_str);
+    let existing_retain_until = retention.retain_until_date;
+    let mode_changed = new_mode != Some(mode);
 
     // Check if new retention period is shorter than existing
     let is_shortening = match (&existing_retain_until, &new_retain_until) {
@@ -93,35 +92,34 @@ pub fn check_retention_for_modification(
         _ => false,
     };
 
-    // COMPLIANCE mode: cannot shorten retention at all (even with bypass)
-    // Can only extend the retention period
-    if mode_str == ObjectLockRetentionMode::COMPLIANCE {
-        if mode_changed || is_shortening {
-            return Some(ObjectLockBlockReason::Retention {
-                mode: mode_str.to_string(),
-                retain_until: existing_retain_until,
-            });
+    match mode {
+        // COMPLIANCE mode: cannot shorten retention at all (even with bypass)
+        // Can only extend the retention period
+        RetentionMode::Compliance => {
+            if mode_changed || is_shortening {
+                return Some(ObjectLockBlockReason::Retention {
+                    mode,
+                    retain_until: existing_retain_until,
+                });
+            }
+            // Extending retention in COMPLIANCE mode is allowed
+            None
         }
-        // Extending retention in COMPLIANCE mode is allowed
-        return None;
-    }
-
-    // GOVERNANCE mode: extending is always allowed, shortening requires bypass
-    // This matches AWS S3 behavior where:
-    // - Extending retention: allowed without bypass permission
-    // - Shortening/removing retention: requires bypass permission
-    if mode_str == ObjectLockRetentionMode::GOVERNANCE {
-        if (mode_changed || is_shortening) && !bypass_governance {
-            return Some(ObjectLockBlockReason::Retention {
-                mode: mode_str.to_string(),
-                retain_until: existing_retain_until,
-            });
+        // GOVERNANCE mode: extending is always allowed, shortening requires bypass
+        // This matches AWS S3 behavior where:
+        // - Extending retention: allowed without bypass permission
+        // - Shortening/removing retention: requires bypass permission
+        RetentionMode::Governance => {
+            if (mode_changed || is_shortening) && !bypass_governance {
+                return Some(ObjectLockBlockReason::Retention {
+                    mode,
+                    retain_until: existing_retain_until,
+                });
+            }
+            // Extending retention or shortening with bypass is allowed
+            None
         }
-        // Extending retention or shortening with bypass is allowed
-        return None;
     }
-
-    None
 }
 
 pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
@@ -137,8 +135,7 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
-    let lhold = objectlock::get_object_legalhold_meta(user_defined);
-    matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
+    objectlock::get_object_legalhold_meta(user_defined).is_on()
 }
 
 /// Whether an authorized replication write (`ObjectOptions::replication_request`)
@@ -172,11 +169,11 @@ pub fn replication_write_may_pass_worm_gate(
         // Delete markers are never locked (same as the WORM gate).
         return Ok(true);
     }
-    let config = object_lock_config_from_state(state)?;
+    let default_retention = default_retention_from_state(state)?;
     if legal_hold_locks(obj_info)? && opts.replication_legalhold_timestamp.is_none() {
         return Ok(false);
     }
-    let retention_locked = active_retention(config, obj_info)?.is_some();
+    let retention_locked = active_retention(default_retention.as_ref(), obj_info)?.is_some();
     Ok(!(retention_locked && opts.replication_retention_timestamp.is_none()))
 }
 
@@ -204,8 +201,8 @@ pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<Str
 
     // Check retention - reuse is_retention_active to avoid code duplication
     let ret = objectlock::get_object_retention_meta(user_defined);
-    if let Some(mode) = &ret.mode
-        && is_retention_active(mode.as_str(), ret.retain_until_date.as_ref())
+    if let Some(mode) = ret.mode
+        && is_retention_active(mode, ret.retain_until_date)
     {
         return true;
     }
@@ -220,7 +217,7 @@ pub enum ObjectLockBlockReason {
     LegalHold,
     /// Object is under retention until the specified date
     Retention {
-        mode: String,
+        mode: RetentionMode,
         retain_until: Option<OffsetDateTime>,
     },
 }
@@ -246,30 +243,28 @@ impl ObjectLockBlockReason {
 /// Check if retention blocks deletion based on mode and bypass permission.
 /// Returns Some(ObjectLockBlockReason) if blocked, None if allowed.
 fn check_retention_blocks_deletion(
-    mode_str: &str,
+    mode: RetentionMode,
     retain_until: Option<OffsetDateTime>,
     bypass_governance: bool,
 ) -> Option<ObjectLockBlockReason> {
     // COMPLIANCE mode cannot be bypassed; GOVERNANCE can only be bypassed with permission
-    let can_bypass = mode_str == ObjectLockRetentionMode::GOVERNANCE && bypass_governance;
+    let can_bypass = mode == RetentionMode::Governance && bypass_governance;
     if !can_bypass {
-        return Some(ObjectLockBlockReason::Retention {
-            mode: mode_str.to_string(),
-            retain_until,
-        });
+        return Some(ObjectLockBlockReason::Retention { mode, retain_until });
     }
     None
 }
 
-/// Check an object's lock metadata using an already resolved bucket Object
-/// Lock configuration. `None` means the configuration is confirmed absent.
+/// Check an object's lock metadata using an already resolved bucket default
+/// retention. `None` means the bucket configuration is confirmed absent or
+/// carries no usable default retention rule.
 ///
 /// # S3 Standard Behavior
 /// - COMPLIANCE mode: Cannot be deleted even with bypass header
 /// - GOVERNANCE mode: Can be deleted if bypass_governance is true (caller must verify s3:BypassGovernanceRetention permission)
 /// - Legal Hold: Cannot be bypassed regardless of mode
-pub(crate) fn check_object_lock_for_deletion_with_config(
-    config: Option<&ObjectLockConfiguration>,
+pub(crate) fn check_object_lock_for_deletion_with_default_retention(
+    default_retention: Option<&DefaultRetention>,
     obj_info: &ObjectInfo,
     bypass_governance: bool,
 ) -> Result<Option<ObjectLockBlockReason>> {
@@ -281,8 +276,8 @@ pub(crate) fn check_object_lock_for_deletion_with_config(
         return Ok(Some(ObjectLockBlockReason::LegalHold));
     }
 
-    if let Some((mode_str, retain_until)) = active_retention(config, obj_info)?
-        && let Some(reason) = check_retention_blocks_deletion(mode_str, Some(retain_until), bypass_governance)
+    if let Some((mode, retain_until)) = active_retention(default_retention, obj_info)?
+        && let Some(reason) = check_retention_blocks_deletion(mode, Some(retain_until), bypass_governance)
     {
         return Ok(Some(reason));
     }
@@ -300,56 +295,41 @@ fn persisted_lock_value<'a>(obj_info: &'a ObjectInfo, key: &str) -> Option<&'a S
 /// Whether the version's persisted legal hold is ON. Any other non-empty
 /// value than ON/OFF is malformed metadata and fails closed.
 fn legal_hold_locks(obj_info: &ObjectInfo) -> Result<bool> {
-    let Some(status) = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str()) else {
+    let Some(status) = persisted_lock_value(obj_info, AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER) else {
         return Ok(false);
     };
-    if status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::ON) {
-        return Ok(true);
+    match LegalHoldStatus::parse(status) {
+        Some(LegalHoldStatus::On) => Ok(true),
+        Some(LegalHoldStatus::Off) => Ok(false),
+        None => Err(Error::other("persisted object legal-hold metadata is invalid")),
     }
-    if !status.eq_ignore_ascii_case(ObjectLockLegalHoldStatus::OFF) {
-        return Err(Error::other("persisted object legal-hold metadata is invalid"));
-    }
-    Ok(false)
 }
 
 /// The retention that currently locks the version, if any: the explicit
 /// persisted retention when the keys are present, otherwise the bucket
 /// default retention computed from the version's modification time. Returns
 /// `(mode, retain_until)` only while the retention is still active.
-fn active_retention<'a>(
-    config: Option<&'a ObjectLockConfiguration>,
+fn active_retention(
+    default_retention: Option<&DefaultRetention>,
     obj_info: &ObjectInfo,
-) -> Result<Option<(&'a str, OffsetDateTime)>> {
-    let mode = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_MODE.as_str());
-    let retain_until = persisted_lock_value(obj_info, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str());
+) -> Result<Option<(RetentionMode, OffsetDateTime)>> {
+    let mode = persisted_lock_value(obj_info, AMZ_OBJECT_LOCK_MODE_LOWER);
+    let retain_until = persisted_lock_value(obj_info, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER);
     match (mode, retain_until) {
         (None, None) => {}
         (Some(mode), Some(retain_until)) => {
             let mode =
                 objectlock::parse_ret_mode(mode).ok_or_else(|| Error::other("persisted object retention mode is invalid"))?;
             let retain_until = OffsetDateTime::parse(retain_until, &time::format_description::well_known::Iso8601::DEFAULT)
-                .map(Date::from)
                 .map_err(|_| Error::other("persisted object retention date is invalid"))?;
-            let mode_str = match mode.as_str() {
-                ObjectLockRetentionMode::COMPLIANCE => ObjectLockRetentionMode::COMPLIANCE,
-                ObjectLockRetentionMode::GOVERNANCE => ObjectLockRetentionMode::GOVERNANCE,
-                _ => return Err(Error::other("persisted object retention mode is invalid")),
-            };
-            return Ok(is_retention_active(mode_str, Some(&retain_until)).then(|| (mode_str, OffsetDateTime::from(retain_until))));
+            return Ok(is_retention_active(mode, Some(retain_until)).then_some((mode, retain_until)));
         }
         _ => return Err(Error::other("persisted object retention metadata is incomplete")),
     }
 
-    let Some(default_retention) = config.and_then(|config| config.rule.as_ref()?.default_retention.as_ref()) else {
+    let Some(default_retention) = default_retention else {
         return Ok(None);
     };
-    let Some(mode) = &default_retention.mode else {
-        return Ok(None);
-    };
-    let mode_str = mode.as_str();
-    if mode_str != ObjectLockRetentionMode::COMPLIANCE && mode_str != ObjectLockRetentionMode::GOVERNANCE {
-        return Ok(None);
-    }
     // Calculate retention expiration date from object modification time
     let mod_time = obj_info
         .mod_time
@@ -363,12 +343,16 @@ fn active_retention<'a>(
             .ok_or_else(|| Error::other("persisted bucket Object Lock retention period is invalid"))?;
         add_years(mod_time, years)
     };
-    Ok((retain_until.unix_timestamp() > now.unix_timestamp()).then_some((mode_str, retain_until)))
+    Ok((retain_until.unix_timestamp() > now.unix_timestamp()).then_some((default_retention.mode, retain_until)))
 }
 
-fn object_lock_config_from_state(state: &ObjectLockConfigState) -> Result<Option<&ObjectLockConfiguration>> {
+/// The bucket default retention carried by an authoritative Object Lock
+/// state. `ConfirmedAbsent` and a configuration without a usable default
+/// retention rule are both `None`; a fabricated state is an error, never a
+/// pass.
+fn default_retention_from_state(state: &ObjectLockConfigState) -> Result<Option<DefaultRetention>> {
     match state {
-        ObjectLockConfigState::Configured { config, .. } => Ok(Some(config)),
+        ObjectLockConfigState::Configured { config, .. } => Ok(default_retention_from_object_lock_config(config)),
         ObjectLockConfigState::ConfirmedAbsent => Ok(None),
         ObjectLockConfigState::Fabricated => Err(Error::other("bucket Object Lock metadata is not authoritative")),
     }
@@ -379,7 +363,11 @@ pub(crate) fn check_object_lock_for_deletion_with_state(
     obj_info: &ObjectInfo,
     bypass_governance: bool,
 ) -> Result<Option<ObjectLockBlockReason>> {
-    check_object_lock_for_deletion_with_config(object_lock_config_from_state(state)?, obj_info, bypass_governance)
+    check_object_lock_for_deletion_with_default_retention(
+        default_retention_from_state(state)?.as_ref(),
+        obj_info,
+        bypass_governance,
+    )
 }
 
 /// Compatibility wrapper for callers that predate fallible metadata lookup.
@@ -402,7 +390,10 @@ pub async fn check_object_lock_for_deletion(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use s3s::dto::{ObjectLockEnabled, ObjectLockRule};
+    use crate::bucket::metadata_sys::configured_object_lock_state_for_tests;
+    use rustfs_utils::http::headers::{
+        AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
+    };
     use time::{Date, Month, PrimitiveDateTime, Time};
 
     fn make_datetime(year: i32, month: u8, day: u8) -> OffsetDateTime {
@@ -411,51 +402,46 @@ mod tests {
         PrimitiveDateTime::new(date, time).assume_utc()
     }
 
-    fn default_retention_config(mode: &'static str) -> ObjectLockConfiguration {
-        ObjectLockConfiguration {
-            object_lock_enabled: Some(ObjectLockEnabled::from_static(ObjectLockEnabled::ENABLED)),
-            rule: Some(ObjectLockRule {
-                default_retention: Some(DefaultRetention {
-                    mode: Some(ObjectLockRetentionMode::from_static(mode)),
-                    days: Some(30),
-                    years: None,
-                }),
-            }),
+    fn default_retention(mode: RetentionMode) -> DefaultRetention {
+        DefaultRetention {
+            mode,
+            days: Some(30),
+            years: None,
         }
     }
 
     #[test]
     fn deletion_with_config_blocks_active_default_compliance_even_with_bypass() {
-        let config = default_retention_config(ObjectLockRetentionMode::COMPLIANCE);
+        let retention = default_retention(RetentionMode::Compliance);
         let obj_info = ObjectInfo {
             mod_time: Some(OffsetDateTime::now_utc()),
             ..Default::default()
         };
 
-        let result = check_object_lock_for_deletion_with_config(Some(&config), &obj_info, true);
+        let result = check_object_lock_for_deletion_with_default_retention(Some(&retention), &obj_info, true);
 
         assert!(matches!(result, Ok(Some(ObjectLockBlockReason::Retention { .. }))));
     }
 
     #[test]
     fn deletion_with_config_allows_active_default_governance_with_bypass() {
-        let config = default_retention_config(ObjectLockRetentionMode::GOVERNANCE);
+        let retention = default_retention(RetentionMode::Governance);
         let obj_info = ObjectInfo {
             mod_time: Some(OffsetDateTime::now_utc()),
             ..Default::default()
         };
 
         assert!(matches!(
-            check_object_lock_for_deletion_with_config(Some(&config), &obj_info, true),
+            check_object_lock_for_deletion_with_default_retention(Some(&retention), &obj_info, true),
             Ok(None)
         ));
     }
 
     #[test]
     fn deletion_with_default_retention_rejects_missing_object_mod_time() {
-        let config = default_retention_config(ObjectLockRetentionMode::COMPLIANCE);
+        let retention = default_retention(RetentionMode::Compliance);
 
-        let err = check_object_lock_for_deletion_with_config(Some(&config), &ObjectInfo::default(), false)
+        let err = check_object_lock_for_deletion_with_default_retention(Some(&retention), &ObjectInfo::default(), false)
             .expect_err("default retention needs an authoritative object modification time");
 
         assert!(err.to_string().contains("modification time"));
@@ -465,7 +451,7 @@ mod tests {
     fn deletion_with_confirmed_absence_still_blocks_explicit_compliance() {
         let retain_until = OffsetDateTime::now_utc() + time::Duration::days(30);
         let mut user_defined = std::collections::HashMap::new();
-        user_defined.insert("x-amz-object-lock-mode".to_string(), ObjectLockRetentionMode::COMPLIANCE.to_string());
+        user_defined.insert("x-amz-object-lock-mode".to_string(), RetentionMode::COMPLIANCE.to_string());
         user_defined.insert(
             "x-amz-object-lock-retain-until-date".to_string(),
             retain_until
@@ -477,7 +463,7 @@ mod tests {
             ..Default::default()
         };
 
-        let result = check_object_lock_for_deletion_with_config(None, &obj_info, true);
+        let result = check_object_lock_for_deletion_with_default_retention(None, &obj_info, true);
 
         assert!(matches!(result, Ok(Some(ObjectLockBlockReason::Retention { .. }))));
     }
@@ -501,16 +487,13 @@ mod tests {
     #[test]
     fn deletion_rejects_incomplete_persisted_retention_metadata() {
         let mut user_defined = std::collections::HashMap::new();
-        user_defined.insert(
-            X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(),
-            ObjectLockRetentionMode::COMPLIANCE.to_string(),
-        );
+        user_defined.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), RetentionMode::COMPLIANCE.to_string());
         let obj_info = ObjectInfo {
             user_defined: Arc::new(user_defined),
             ..Default::default()
         };
 
-        let err = check_object_lock_for_deletion_with_config(None, &obj_info, false)
+        let err = check_object_lock_for_deletion_with_default_retention(None, &obj_info, false)
             .expect_err("mode without retain-until date must fail closed");
 
         assert!(err.to_string().contains("incomplete"));
@@ -523,29 +506,24 @@ mod tests {
             .expect("retain-until date should format");
         let cases = [
             ("invalid mode", Some("INVALID"), Some(valid_date.as_str()), "retention mode"),
-            (
-                "invalid date",
-                Some(ObjectLockRetentionMode::COMPLIANCE),
-                Some("not-a-date"),
-                "retention date",
-            ),
+            ("invalid date", Some(RetentionMode::COMPLIANCE), Some("not-a-date"), "retention date"),
             ("date only", None, Some(valid_date.as_str()), "incomplete"),
         ];
 
         for (case, mode, retain_until, expected) in cases {
             let mut user_defined = std::collections::HashMap::new();
             if let Some(mode) = mode {
-                user_defined.insert(X_AMZ_OBJECT_LOCK_MODE.as_str().to_string(), mode.to_string());
+                user_defined.insert(AMZ_OBJECT_LOCK_MODE_LOWER.to_string(), mode.to_string());
             }
             if let Some(retain_until) = retain_until {
-                user_defined.insert(X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE.as_str().to_string(), retain_until.to_string());
+                user_defined.insert(AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER.to_string(), retain_until.to_string());
             }
             let obj_info = ObjectInfo {
                 user_defined: Arc::new(user_defined),
                 ..Default::default()
             };
 
-            let err = check_object_lock_for_deletion_with_config(None, &obj_info, false).expect_err(case);
+            let err = check_object_lock_for_deletion_with_default_retention(None, &obj_info, false).expect_err(case);
             assert!(err.to_string().contains(expected), "unexpected {case} error: {err}");
         }
     }
@@ -579,10 +557,6 @@ mod tests {
     /// source timestamp of every category that currently locks the version.
     #[test]
     fn replication_write_passes_worm_gate_only_with_every_locking_category_timestamp() {
-        use rustfs_utils::http::headers::{
-            AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER,
-        };
-
         let hold = [(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, "ON")];
         let retention = [
             (AMZ_OBJECT_LOCK_MODE_LOWER, "GOVERNANCE"),
@@ -623,19 +597,15 @@ mod tests {
     }
 
     /// The bucket default retention locks a version that carries no explicit
-    /// retention keys (`check_object_lock_for_deletion_with_config` judges it
-    /// from the modification time), so the replication bypass must demand the
-    /// retention source timestamp for it too — a tagging-only replication
-    /// write must not overwrite the default-protected version unjudged.
+    /// retention keys (`check_object_lock_for_deletion_with_default_retention`
+    /// judges it from the modification time), so the replication bypass must
+    /// demand the retention source timestamp for it too — a tagging-only
+    /// replication write must not overwrite the default-protected version
+    /// unjudged.
     #[test]
     fn replication_write_under_bucket_default_retention_requires_retention_timestamp() {
-        use rustfs_utils::http::headers::{AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER};
-
-        for mode in [ObjectLockRetentionMode::COMPLIANCE, ObjectLockRetentionMode::GOVERNANCE] {
-            let state = ObjectLockConfigState::Configured {
-                config: default_retention_config(mode),
-                updated_at: OffsetDateTime::now_utc(),
-            };
+        for mode in [RetentionMode::Compliance, RetentionMode::Governance] {
+            let state = configured_object_lock_state_for_tests(mode, 30);
             let no_keys = lock_object_info(std::collections::HashMap::new());
             assert!(
                 check_object_lock_for_deletion_with_state(&state, &no_keys, false)
@@ -689,8 +659,6 @@ mod tests {
     /// state or malformed persisted lock metadata; both are errors, not a pass.
     #[test]
     fn replication_write_worm_gate_fails_closed_on_unverifiable_lock_state() {
-        use rustfs_utils::http::headers::AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER;
-
         let opts = replication_opts(true, true);
         let err = replication_write_may_pass_worm_gate(
             &ObjectLockConfigState::Fabricated,
@@ -705,10 +673,7 @@ mod tests {
             .expect_err("malformed legal hold must not be judged");
         assert!(err.to_string().contains("legal-hold"));
 
-        let state = ObjectLockConfigState::Configured {
-            config: default_retention_config(ObjectLockRetentionMode::COMPLIANCE),
-            updated_at: OffsetDateTime::now_utc(),
-        };
+        let state = configured_object_lock_state_for_tests(RetentionMode::Compliance, 30);
         let no_mod_time = ObjectInfo::default();
         let err = replication_write_may_pass_worm_gate(&state, &no_mod_time, &opts)
             .expect_err("default retention without a modification time must not be judged");
@@ -722,10 +687,6 @@ mod tests {
     /// (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",
@@ -749,7 +710,7 @@ mod tests {
                 ..Default::default()
             };
 
-            let result = check_object_lock_for_deletion_with_config(None, &obj_info, false);
+            let result = check_object_lock_for_deletion_with_default_retention(None, &obj_info, false);
             assert!(matches!(result, Ok(None)), "{case}: empty lock keys must read as unlocked: {result:?}");
         }
     }
@@ -757,13 +718,13 @@ mod tests {
     #[test]
     fn deletion_rejects_invalid_persisted_legal_hold_metadata() {
         let mut user_defined = std::collections::HashMap::new();
-        user_defined.insert(X_AMZ_OBJECT_LOCK_LEGAL_HOLD.as_str().to_string(), "INVALID".to_string());
+        user_defined.insert(AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER.to_string(), "INVALID".to_string());
         let obj_info = ObjectInfo {
             user_defined: Arc::new(user_defined),
             ..Default::default()
         };
 
-        let err = check_object_lock_for_deletion_with_config(None, &obj_info, false)
+        let err = check_object_lock_for_deletion_with_default_retention(None, &obj_info, false)
             .expect_err("invalid legal-hold value must fail closed");
 
         assert!(err.to_string().contains("legal-hold"));
@@ -829,42 +790,29 @@ mod tests {
         assert_eq!(result.day(), 4);
     }
 
-    #[test]
-    fn test_is_retention_active_invalid_mode() {
-        // Invalid mode should return false
-        assert!(!is_retention_active("INVALID", None));
-        assert!(!is_retention_active("", None));
-    }
-
     #[test]
     fn test_is_retention_active_no_date() {
         // Valid mode but no retain_until_date should return false
-        assert!(!is_retention_active(ObjectLockRetentionMode::COMPLIANCE, None));
-        assert!(!is_retention_active(ObjectLockRetentionMode::GOVERNANCE, None));
+        assert!(!is_retention_active(RetentionMode::Compliance, None));
+        assert!(!is_retention_active(RetentionMode::Governance, None));
     }
 
     #[test]
     fn test_is_retention_active_future_date() {
         // Valid mode with future retain_until_date should return true
         let future_date = OffsetDateTime::now_utc() + time::Duration::days(30);
-        let s3_date = s3s::dto::Date::from(future_date);
 
-        assert!(is_retention_active(ObjectLockRetentionMode::COMPLIANCE, Some(&s3_date)));
-        let future_date = OffsetDateTime::now_utc() + time::Duration::days(30);
-        let s3_date = s3s::dto::Date::from(future_date);
-        assert!(is_retention_active(ObjectLockRetentionMode::GOVERNANCE, Some(&s3_date)));
+        assert!(is_retention_active(RetentionMode::Compliance, Some(future_date)));
+        assert!(is_retention_active(RetentionMode::Governance, Some(future_date)));
     }
 
     #[test]
     fn test_is_retention_active_past_date() {
         // Valid mode with past retain_until_date should return false
         let past_date = OffsetDateTime::now_utc() - time::Duration::days(30);
-        let s3_date = s3s::dto::Date::from(past_date);
 
-        assert!(!is_retention_active(ObjectLockRetentionMode::COMPLIANCE, Some(&s3_date)));
-        let past_date = OffsetDateTime::now_utc() - time::Duration::days(30);
-        let s3_date = s3s::dto::Date::from(past_date);
-        assert!(!is_retention_active(ObjectLockRetentionMode::GOVERNANCE, Some(&s3_date)));
+        assert!(!is_retention_active(RetentionMode::Compliance, Some(past_date)));
+        assert!(!is_retention_active(RetentionMode::Governance, Some(past_date)));
     }
 
     #[test]
@@ -890,10 +838,7 @@ mod tests {
 
         // Extending by another 30 days should be allowed
         let new_retain = Some(existing_retain + time::Duration::days(30));
-        assert!(
-            check_retention_for_modification(&user_defined, Some(ObjectLockRetentionMode::COMPLIANCE), new_retain, false)
-                .is_none()
-        );
+        assert!(check_retention_for_modification(&user_defined, Some(RetentionMode::Compliance), new_retain, false).is_none());
     }
 
     #[test]
@@ -911,8 +856,7 @@ mod tests {
 
         // Shortening to 30 days should be blocked
         let new_retain = Some(OffsetDateTime::now_utc() + time::Duration::days(30));
-        let result =
-            check_retention_for_modification(&user_defined, Some(ObjectLockRetentionMode::COMPLIANCE), new_retain, false);
+        let result = check_retention_for_modification(&user_defined, Some(RetentionMode::Compliance), new_retain, false);
         assert!(result.is_some());
         assert!(matches!(result, Some(ObjectLockBlockReason::Retention { .. })));
     }
@@ -950,8 +894,7 @@ mod tests {
 
         // Shortening from 30 days to 15 days without bypass should be blocked
         let new_retain = Some(OffsetDateTime::now_utc() + time::Duration::days(15));
-        let result =
-            check_retention_for_modification(&user_defined, Some(ObjectLockRetentionMode::GOVERNANCE), new_retain, false);
+        let result = check_retention_for_modification(&user_defined, Some(RetentionMode::Governance), new_retain, false);
         assert!(result.is_some());
     }
 
@@ -971,10 +914,7 @@ mod tests {
 
         // Extending from 30 days to 60 days without bypass should be allowed
         let new_retain = Some(OffsetDateTime::now_utc() + time::Duration::days(60));
-        assert!(
-            check_retention_for_modification(&user_defined, Some(ObjectLockRetentionMode::GOVERNANCE), new_retain, false)
-                .is_none()
-        );
+        assert!(check_retention_for_modification(&user_defined, Some(RetentionMode::Governance), new_retain, false).is_none());
     }
 
     #[test]
@@ -992,10 +932,7 @@ mod tests {
 
         // Shortening from 30 days to 15 days with bypass should be allowed
         let new_retain = Some(OffsetDateTime::now_utc() + time::Duration::days(15));
-        assert!(
-            check_retention_for_modification(&user_defined, Some(ObjectLockRetentionMode::GOVERNANCE), new_retain, true)
-                .is_none()
-        );
+        assert!(check_retention_for_modification(&user_defined, Some(RetentionMode::Governance), new_retain, true).is_none());
     }
 
     #[test]
@@ -1010,12 +947,8 @@ mod tests {
                 .unwrap(),
         );
 
-        let result = check_retention_for_modification(
-            &user_defined,
-            Some(ObjectLockRetentionMode::COMPLIANCE),
-            Some(existing_retain),
-            false,
-        );
+        let result =
+            check_retention_for_modification(&user_defined, Some(RetentionMode::Compliance), Some(existing_retain), false);
         assert!(result.is_some());
     }
 
@@ -1032,13 +965,8 @@ mod tests {
         );
 
         assert!(
-            check_retention_for_modification(
-                &user_defined,
-                Some(ObjectLockRetentionMode::COMPLIANCE),
-                Some(existing_retain),
-                true,
-            )
-            .is_none()
+            check_retention_for_modification(&user_defined, Some(RetentionMode::Compliance), Some(existing_retain), true)
+                .is_none()
         );
     }
 
@@ -1054,12 +982,8 @@ mod tests {
                 .unwrap(),
         );
 
-        let result = check_retention_for_modification(
-            &user_defined,
-            Some(ObjectLockRetentionMode::GOVERNANCE),
-            Some(existing_retain),
-            true,
-        );
+        let result =
+            check_retention_for_modification(&user_defined, Some(RetentionMode::Governance), Some(existing_retain), true);
         assert!(result.is_some());
     }
 

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

@@ -0,0 +1,179 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Storage-level Object Lock types (rustfs/backlog#1842).
+//!
+//! The engine evaluates WORM state from persisted object metadata and the
+//! bucket default retention; none of that needs S3 wire/DTO types. The
+//! serving layer converts to/from its wire DTOs at its own boundary, and the
+//! bucket-metadata module converts the persisted `ObjectLockConfiguration`
+//! into [`DefaultRetention`] when handing it to the evaluation code here.
+
+use std::fmt;
+use time::OffsetDateTime;
+
+/// Object Lock retention mode. Persisted metadata and the bucket default
+/// retention only ever carry these two values; anything else is either
+/// malformed metadata (fail-closed at the parse site) or an inactive
+/// configuration (ignored at the conversion site).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum RetentionMode {
+    Governance,
+    Compliance,
+}
+
+impl RetentionMode {
+    pub const GOVERNANCE: &'static str = "GOVERNANCE";
+    pub const COMPLIANCE: &'static str = "COMPLIANCE";
+
+    /// Parse the canonical S3 wire spelling, case-insensitively (matching the
+    /// historical `parse_ret_mode` behavior). Returns `None` for anything
+    /// that is not GOVERNANCE/COMPLIANCE.
+    pub fn parse(value: &str) -> Option<Self> {
+        if value.eq_ignore_ascii_case(Self::GOVERNANCE) {
+            Some(Self::Governance)
+        } else if value.eq_ignore_ascii_case(Self::COMPLIANCE) {
+            Some(Self::Compliance)
+        } else {
+            None
+        }
+    }
+
+    /// Parse only the exact canonical wire spelling. Use this for a mode a
+    /// caller supplies in a *request*: the retention-modification gate has
+    /// always compared the requested mode literally against the canonical
+    /// persisted mode, so a non-canonical spelling must stay "not the same
+    /// mode" (and therefore blocked), not be normalized into a match.
+    pub fn parse_exact(value: &str) -> Option<Self> {
+        match value {
+            Self::GOVERNANCE => Some(Self::Governance),
+            Self::COMPLIANCE => Some(Self::Compliance),
+            _ => None,
+        }
+    }
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::Governance => Self::GOVERNANCE,
+            Self::Compliance => Self::COMPLIANCE,
+        }
+    }
+}
+
+impl fmt::Display for RetentionMode {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+/// Object Lock legal hold status (ON/OFF).
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum LegalHoldStatus {
+    On,
+    Off,
+}
+
+impl LegalHoldStatus {
+    pub const ON: &'static str = "ON";
+    pub const OFF: &'static str = "OFF";
+
+    /// Parse the canonical S3 wire spelling, case-insensitively (matching the
+    /// historical `parse_legalhold_status` behavior).
+    pub fn parse(value: &str) -> Option<Self> {
+        if value.eq_ignore_ascii_case(Self::ON) {
+            Some(Self::On)
+        } else if value.eq_ignore_ascii_case(Self::OFF) {
+            Some(Self::Off)
+        } else {
+            None
+        }
+    }
+
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::On => Self::ON,
+            Self::Off => Self::OFF,
+        }
+    }
+}
+
+impl fmt::Display for LegalHoldStatus {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.write_str(self.as_str())
+    }
+}
+
+/// An object version's retention as read from persisted metadata. `mode` is
+/// `None` when the metadata carries no (or an unparsable) retention mode.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ObjectRetention {
+    pub mode: Option<RetentionMode>,
+    pub retain_until_date: Option<OffsetDateTime>,
+}
+
+/// An object version's legal hold as read from persisted metadata. `status`
+/// is `None` when the metadata carries no (or an unparsable) legal hold.
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct ObjectLegalHold {
+    pub status: Option<LegalHoldStatus>,
+}
+
+impl ObjectLegalHold {
+    pub fn is_on(&self) -> bool {
+        self.status == Some(LegalHoldStatus::On)
+    }
+}
+
+/// The bucket's default Object Lock retention, converted from the persisted
+/// configuration. Conversion only yields a value for an active default
+/// retention (a valid GOVERNANCE/COMPLIANCE mode); a rule without a usable
+/// mode converts to `None`, matching how the evaluation code has always
+/// ignored such rules.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub struct DefaultRetention {
+    pub mode: RetentionMode,
+    pub days: Option<i32>,
+    pub years: Option<i32>,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    /// The modification gate compares a *requested* mode against the canonical
+    /// persisted mode literally: a non-canonical spelling must not normalize
+    /// into a match, or a client could shorten GOVERNANCE retention without
+    /// bypass by spelling the mode differently. `parse_exact` is that pin.
+    #[test]
+    fn parse_exact_accepts_only_canonical_spellings() {
+        assert_eq!(RetentionMode::parse_exact("GOVERNANCE"), Some(RetentionMode::Governance));
+        assert_eq!(RetentionMode::parse_exact("COMPLIANCE"), Some(RetentionMode::Compliance));
+        for non_canonical in ["governance", "Governance", "compliance", "Compliance", "", "INVALID"] {
+            assert_eq!(RetentionMode::parse_exact(non_canonical), None, "{non_canonical:?} must not parse");
+        }
+    }
+
+    /// Persisted metadata parsing stays case-insensitive (the historical
+    /// `parse_ret_mode` / `parse_legalhold_status` behavior): on-disk values
+    /// written by older builds must keep locking.
+    #[test]
+    fn parse_is_case_insensitive_for_persisted_values() {
+        assert_eq!(RetentionMode::parse("governance"), Some(RetentionMode::Governance));
+        assert_eq!(RetentionMode::parse("Compliance"), Some(RetentionMode::Compliance));
+        assert_eq!(LegalHoldStatus::parse("on"), Some(LegalHoldStatus::On));
+        assert_eq!(LegalHoldStatus::parse("Off"), Some(LegalHoldStatus::Off));
+        assert_eq!(RetentionMode::parse("INVALID"), None);
+        assert_eq!(LegalHoldStatus::parse("MAYBE"), None);
+    }
+}

crates/ecstore/src/set_disk/mod.rs

@@ -46,8 +46,8 @@ use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
 use crate::bucket::metadata_sys;
 use crate::bucket::metadata_sys::ObjectLockConfigState;
 use crate::bucket::object_lock::objectlock_sys::{
-    check_object_lock_for_deletion_with_config, check_object_lock_for_deletion_with_state, check_retention_for_modification,
-    replication_write_may_pass_worm_gate,
+    check_object_lock_for_deletion_with_default_retention, check_object_lock_for_deletion_with_state,
+    check_retention_for_modification, replication_write_may_pass_worm_gate,
 };
 use crate::bucket::replication::{
     ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
@@ -4688,7 +4688,10 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
     if let Some(retention) = &opts.object_lock_retention
         && check_retention_for_modification(
             &obj_info.user_defined,
-            retention.mode.as_deref(),
+            retention
+                .mode
+                .as_deref()
+                .and_then(crate::bucket::object_lock::types::RetentionMode::parse_exact),
             retention.retain_until,
             retention.bypass_governance,
         )

rustfs/src/app/object_usecase.rs

@@ -40,6 +40,7 @@ use super::storage_api::object_usecase::bucket::{
     object_lock::{
         objectlock::{get_object_legalhold_meta, get_object_retention_meta},
         objectlock_sys::{check_object_lock_for_deletion, is_retention_active, replication_write_may_pass_worm_gate},
+        types::RetentionMode,
     },
     predict_lifecycle_expiration,
     quota::{QuotaCheckResult, QuotaError, QuotaOperation},
@@ -4045,11 +4046,7 @@ pub(crate) fn validate_existing_object_lock_for_write(
     }
 
     let legal_hold = get_object_legalhold_meta(&existing_obj_info.user_defined);
-    if legal_hold
-        .status
-        .as_ref()
-        .is_some_and(|status| status.as_str() == ObjectLockLegalHoldStatus::ON)
-    {
+    if legal_hold.is_on() {
         return Err(S3Error::with_message(
             S3ErrorCode::AccessDenied,
             "Object has a legal hold and cannot be overwritten. Remove the legal hold first.".to_string(),
@@ -4057,9 +4054,9 @@ pub(crate) fn validate_existing_object_lock_for_write(
     }
 
     let retention = get_object_retention_meta(&existing_obj_info.user_defined);
-    if let Some(mode) = retention.mode.as_ref()
-        && mode.as_str() == ObjectLockRetentionMode::COMPLIANCE
-        && is_retention_active(mode.as_str(), retention.retain_until_date.as_ref())
+    if let Some(mode) = retention.mode
+        && mode == RetentionMode::Compliance
+        && is_retention_active(mode, retention.retain_until_date)
     {
         return Err(S3Error::with_message(
             S3ErrorCode::AccessDenied,

rustfs/src/app/storage_api.rs

@@ -559,16 +559,20 @@ pub(crate) mod bucket {
     }
 
     pub(crate) mod object_lock {
+        pub(crate) mod types {
+            pub(crate) use crate::storage::storage_api::ecstore_bucket::object_lock::types::RetentionMode;
+        }
+
         pub(crate) mod objectlock {
             pub(crate) fn get_object_legalhold_meta(
                 meta: &std::collections::HashMap<String, String>,
-            ) -> s3s::dto::ObjectLockLegalHold {
+            ) -> crate::storage::storage_api::ecstore_bucket::object_lock::types::ObjectLegalHold {
                 crate::storage::storage_api::ecstore_bucket::object_lock::objectlock::get_object_legalhold_meta(meta)
             }
 
             pub(crate) fn get_object_retention_meta(
                 meta: &std::collections::HashMap<String, String>,
-            ) -> s3s::dto::ObjectLockRetention {
+            ) -> crate::storage::storage_api::ecstore_bucket::object_lock::types::ObjectRetention {
                 crate::storage::storage_api::ecstore_bucket::object_lock::objectlock::get_object_retention_meta(meta)
             }
         }
@@ -587,7 +591,10 @@ pub(crate) mod bucket {
                 .await
             }
 
-            pub(crate) fn is_retention_active(mode: &str, retain_until_date: Option<&s3s::dto::Date>) -> bool {
+            pub(crate) fn is_retention_active(
+                mode: crate::storage::storage_api::ecstore_bucket::object_lock::types::RetentionMode,
+                retain_until_date: Option<time::OffsetDateTime>,
+            ) -> bool {
                 crate::storage::storage_api::ecstore_bucket::object_lock::objectlock_sys::is_retention_active(
                     mode,
                     retain_until_date,

rustfs/src/storage/storage_api.rs

@@ -1745,6 +1745,10 @@ pub(crate) fn check_retention_for_modification(
     new_retain_until: Option<time::OffsetDateTime>,
     bypass_governance: bool,
 ) -> Option<ObjectLockBlockReason> {
+    // The gate compares the requested mode literally against the canonical
+    // persisted mode, so only the exact canonical spelling maps to a typed
+    // mode; anything else stays `None` and is judged as a mode change.
+    let new_mode = new_mode.and_then(ecstore_bucket::object_lock::types::RetentionMode::parse_exact);
     ecstore_bucket::object_lock::objectlock_sys::check_retention_for_modification(
         user_defined,
         new_mode,

scripts/check_s3s_footprint.sh

@@ -25,14 +25,14 @@ cd "$(dirname "$0")/.."
 # Baselines verified on 2026-08-11. Lower-only; see header.
 # Excludes crates/e2e_test/ — test infrastructure legitimately uses s3s
 # to verify S3 behavior and does not widen the production s3s surface.
-S3S_IMPORT_FILES_BASELINE=211
+S3S_IMPORT_FILES_BASELINE=208
 S3_ERROR_LINES_BASELINE=1620
 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
 # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
 # client was extracted to crates/s3-client, where s3s usage is legitimate;
 # this counter ratchets the remaining serving-side s3s references out of
 # crates/ecstore. Baseline verified on 2026-08-26.
-S3S_ECSTORE_FILES_BASELINE=42
+S3S_ECSTORE_FILES_BASELINE=39
 S3S_PATH_PATTERN='(^|[^"[:alnum:]_])s3s::'
 E2E_TEST_GLOB='--glob=!crates/e2e_test/**'