refactor(replication): own utility wire contracts (#4255)

Cargo.lock

@@ -9621,7 +9621,6 @@ dependencies = [
  "regex",
  "rmp",
  "rmp-serde",
- "rustfs-utils",
  "s3s",
  "serde",
  "time",

crates/replication/Cargo.toml

@@ -31,7 +31,6 @@ byteorder.workspace = true
 regex.workspace = true
 rmp.workspace = true
 rmp-serde.workspace = true
-rustfs-utils = { workspace = true, features = ["http", "path", "string"] }
 s3s.workspace = true
 serde.workspace = true
 time.workspace = true

crates/replication/src/filemeta.rs

@@ -12,10 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::http::internal_key_rustfs;
 use bytes::Bytes;
 use core::fmt;
 use regex::Regex;
-use rustfs_utils::http::internal_key_rustfs;
 use serde::{Deserialize, Serialize};
 use std::any::Any;
 use std::collections::HashMap;

crates/replication/src/http.rs

@@ -0,0 +1,147 @@
+// 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.
+
+use std::collections::HashMap;
+
+const RUSTFS_INTERNAL_PREFIX: &str = "x-rustfs-internal-";
+const MINIO_INTERNAL_PREFIX: &str = "x-minio-internal-";
+const RUSTFS_HEADER_PREFIX: &str = "x-rustfs-";
+const MINIO_HEADER_PREFIX: &str = "x-minio-";
+
+pub(crate) const AMZ_BUCKET_REPLICATION_STATUS: &str = "X-Amz-Replication-Status";
+pub(crate) const AMZ_OBJECT_LOCK_LEGAL_HOLD: &str = "X-Amz-Object-Lock-Legal-Hold";
+pub(crate) const AMZ_OBJECT_LOCK_MODE: &str = "X-Amz-Object-Lock-Mode";
+pub(crate) const AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE: &str = "X-Amz-Object-Lock-Retain-Until-Date";
+pub(crate) const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
+pub(crate) const AMZ_WEBSITE_REDIRECT_LOCATION: &str = "x-amz-website-redirect-location";
+pub(crate) const CACHE_CONTROL: &str = "Cache-Control";
+pub(crate) const CONTENT_DISPOSITION: &str = "Content-Disposition";
+pub(crate) const CONTENT_ENCODING: &str = "Content-Encoding";
+pub(crate) const CONTENT_LANGUAGE: &str = "Content-Language";
+pub(crate) const EXPIRES: &str = "Expires";
+pub(crate) const SSEC_ALGORITHM_HEADER: &str = "x-amz-server-side-encryption-customer-algorithm";
+pub(crate) const SSEC_KEY_HEADER: &str = "x-amz-server-side-encryption-customer-key";
+pub(crate) const SSEC_KEY_MD5_HEADER: &str = "x-amz-server-side-encryption-customer-key-md5";
+pub(crate) const SUFFIX_ACTUAL_SIZE: &str = "actual-size";
+pub(crate) const SUFFIX_REPLICATION_RESET_STATUS: &str = "replication-reset-status";
+
+fn internal_keys(suffix: &str) -> (String, String) {
+    (format!("{RUSTFS_INTERNAL_PREFIX}{suffix}"), format!("{MINIO_INTERNAL_PREFIX}{suffix}"))
+}
+
+fn rustfs_header_key(suffix: &str) -> String {
+    format!("{RUSTFS_HEADER_PREFIX}{suffix}")
+}
+
+fn minio_header_key(suffix: &str) -> String {
+    format!("{MINIO_HEADER_PREFIX}{suffix}")
+}
+
+pub(crate) fn internal_key_rustfs(suffix: &str) -> String {
+    format!("{RUSTFS_INTERNAL_PREFIX}{suffix}")
+}
+
+pub(crate) fn get_internal_metadata(map: &HashMap<String, String>, suffix: &str) -> Option<String> {
+    let (rustfs_key, minio_key) = internal_keys(suffix);
+    map.get(&rustfs_key)
+        .cloned()
+        .or_else(|| map.get(&minio_key).cloned())
+        .or_else(|| {
+            map.iter()
+                .find(|(key, _)| key.eq_ignore_ascii_case(&rustfs_key) || key.eq_ignore_ascii_case(&minio_key))
+                .map(|(_, value)| value.clone())
+        })
+}
+
+pub(crate) fn get_header_metadata(map: &HashMap<String, String>, suffix: &str) -> Option<String> {
+    let rustfs_key = rustfs_header_key(suffix);
+    let minio_key = minio_header_key(suffix);
+    map.get(&rustfs_key).cloned().or_else(|| map.get(&minio_key).cloned())
+}
+
+pub(crate) fn has_prefix_fold(s: &str, prefix: &str) -> bool {
+    if s.starts_with(prefix) {
+        return true;
+    }
+
+    s.get(..prefix.len())
+        .is_some_and(|s_prefix| s_prefix.eq_ignore_ascii_case(prefix))
+}
+
+pub(crate) fn trim_etag(etag: &str) -> String {
+    etag.trim_matches('"').to_string()
+}
+
+#[cfg(test)]
+pub(crate) fn insert_internal_metadata(map: &mut HashMap<String, String>, suffix: &str, value: String) {
+    let (rustfs_key, minio_key) = internal_keys(suffix);
+    map.insert(rustfs_key, value.clone());
+    map.insert(minio_key, value);
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        SUFFIX_ACTUAL_SIZE, SUFFIX_REPLICATION_RESET_STATUS, get_header_metadata, get_internal_metadata, has_prefix_fold,
+        insert_internal_metadata, internal_key_rustfs, trim_etag,
+    };
+    use std::collections::HashMap;
+
+    #[test]
+    fn internal_metadata_prefers_rustfs_and_falls_back_to_minio() {
+        let mut metadata = HashMap::from([("x-minio-internal-actual-size".to_string(), "10".to_string())]);
+
+        assert_eq!(get_internal_metadata(&metadata, SUFFIX_ACTUAL_SIZE).as_deref(), Some("10"));
+
+        metadata.insert("x-rustfs-internal-actual-size".to_string(), "11".to_string());
+        assert_eq!(get_internal_metadata(&metadata, SUFFIX_ACTUAL_SIZE).as_deref(), Some("11"));
+    }
+
+    #[test]
+    fn internal_metadata_keeps_case_insensitive_lookup_compatibility() {
+        let metadata = HashMap::from([("X-RustFS-Internal-Actual-Size".to_string(), "12".to_string())]);
+
+        assert_eq!(get_internal_metadata(&metadata, SUFFIX_ACTUAL_SIZE).as_deref(), Some("12"));
+    }
+
+    #[test]
+    fn internal_metadata_insert_writes_rustfs_and_minio_keys() {
+        let mut metadata = HashMap::new();
+        insert_internal_metadata(&mut metadata, SUFFIX_ACTUAL_SIZE, "13".to_string());
+
+        assert_eq!(metadata.get("x-rustfs-internal-actual-size").map(String::as_str), Some("13"));
+        assert_eq!(metadata.get("x-minio-internal-actual-size").map(String::as_str), Some("13"));
+    }
+
+    #[test]
+    fn header_metadata_prefers_rustfs_then_minio() {
+        let mut metadata = HashMap::from([("x-minio-replication-reset-status".to_string(), "old".to_string())]);
+
+        assert_eq!(get_header_metadata(&metadata, SUFFIX_REPLICATION_RESET_STATUS).as_deref(), Some("old"));
+
+        metadata.insert("x-rustfs-replication-reset-status".to_string(), "new".to_string());
+        assert_eq!(get_header_metadata(&metadata, SUFFIX_REPLICATION_RESET_STATUS).as_deref(), Some("new"));
+    }
+
+    #[test]
+    fn helper_contracts_match_replication_wire_rules() {
+        assert_eq!(
+            internal_key_rustfs("replication-reset-arn:target"),
+            "x-rustfs-internal-replication-reset-arn:target"
+        );
+        assert_eq!(trim_etag("\"abc\""), "abc");
+        assert!(has_prefix_fold("X-Amz-Meta-Foo", "x-amz-meta-"));
+        assert!(!has_prefix_fold("X-Amz-Meta-Foo", "amz-meta"));
+    }
+}

crates/replication/src/lib.rs

@@ -15,6 +15,7 @@
 pub mod config;
 pub mod delete;
 mod filemeta;
+mod http;
 pub mod mrf;
 pub mod multipart;
 pub mod object;

crates/replication/src/multipart.rs

@@ -15,7 +15,7 @@
 use std::collections::HashMap;
 use std::fmt;
 
-use rustfs_utils::http::{SUFFIX_ACTUAL_SIZE, get_str};
+use crate::http::{SUFFIX_ACTUAL_SIZE, get_internal_metadata};
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub struct ReplicationMultipartPartInput {
@@ -106,7 +106,7 @@ pub fn replication_multipart_part_plan(
 }
 
 pub fn replication_multipart_complete_actual_size(user_defined: &HashMap<String, String>) -> String {
-    get_str(user_defined, SUFFIX_ACTUAL_SIZE).unwrap_or_default()
+    get_internal_metadata(user_defined, SUFFIX_ACTUAL_SIZE).unwrap_or_default()
 }
 
 #[cfg(test)]
@@ -115,7 +115,7 @@ mod tests {
         ReplicationMultipartPartInput, ReplicationMultipartPartPlan, ReplicationMultipartPlanError, ReplicationMultipartRange,
         replication_multipart_complete_actual_size, replication_multipart_part_plan,
     };
-    use rustfs_utils::http::{SUFFIX_ACTUAL_SIZE, insert_str};
+    use crate::http::{SUFFIX_ACTUAL_SIZE, insert_internal_metadata};
     use std::collections::HashMap;
 
     #[test]
@@ -214,7 +214,7 @@ mod tests {
     #[test]
     fn multipart_complete_actual_size_reads_compatible_metadata() {
         let mut user_defined = HashMap::new();
-        insert_str(&mut user_defined, SUFFIX_ACTUAL_SIZE, "123".to_string());
+        insert_internal_metadata(&mut user_defined, SUFFIX_ACTUAL_SIZE, "123".to_string());
 
         assert_eq!(replication_multipart_complete_actual_size(&user_defined), "123");
         assert!(replication_multipart_complete_actual_size(&HashMap::new()).is_empty());

crates/replication/src/object.rs

@@ -13,13 +13,12 @@
 // limitations under the License.
 
 use crate::filemeta::{ReplicationAction, ReplicationType};
-use crate::tagging::ReplicationTagFilter;
-use rustfs_utils::http::{
+use crate::http::{
     AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, AMZ_OBJECT_TAGGING,
     AMZ_WEBSITE_REDIRECT_LOCATION, CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, EXPIRES,
+    has_prefix_fold, trim_etag,
 };
-use rustfs_utils::path::trim_etag;
-use rustfs_utils::string::strings_has_prefix_fold;
+use crate::tagging::ReplicationTagFilter;
 use std::collections::HashMap;
 use time::OffsetDateTime;
 
@@ -150,7 +149,7 @@ fn comparable_metadata(metadata: Option<&HashMap<String, String>>) -> HashMap<St
     for (key, value) in metadata.into_iter().flatten() {
         if REPLICATION_METADATA_COMPARE_KEYS
             .iter()
-            .any(|prefix| strings_has_prefix_fold(key, prefix))
+            .any(|prefix| has_prefix_fold(key, prefix))
         {
             comparable.insert(key.to_lowercase(), value.clone());
         }

crates/replication/src/operation.rs

@@ -14,9 +14,9 @@
 
 use std::collections::HashMap;
 
-use rustfs_utils::http::{
+use crate::http::{
     AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, SSEC_ALGORITHM_HEADER, SSEC_KEY_HEADER, SSEC_KEY_MD5_HEADER,
-    SUFFIX_REPLICATION_RESET_STATUS, get_header_map,
+    SUFFIX_REPLICATION_RESET_STATUS, get_header_metadata,
 };
 use s3s::dto::ReplicationConfiguration;
 use time::OffsetDateTime;
@@ -266,7 +266,7 @@ pub fn resync_target_for_object(
         .user_defined
         .get(target_reset_header(arn).as_str())
         .cloned()
-        .or_else(|| get_header_map(object.user_defined, SUFFIX_REPLICATION_RESET_STATUS));
+        .or_else(|| get_header_metadata(object.user_defined, SUFFIX_REPLICATION_RESET_STATUS));
 
     let mut dec = ResyncTargetDecision::default();
 
@@ -318,9 +318,9 @@ mod tests {
         is_ssec_encrypted, resync_target_for_object, should_schedule_delete_replication,
         should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
     };
+    use crate::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER};
     use crate::storage_api::ObjectToDelete;
     use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header};
-    use rustfs_utils::http::{AMZ_BUCKET_REPLICATION_STATUS, SSEC_ALGORITHM_HEADER};
     use s3s::dto::{
         DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication,
         ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration,

docs/architecture/ecstore-module-split-plan.md

@@ -131,6 +131,10 @@ Current coupling:
   `rustfs-replication`; ECStore converts storage-api delete DTOs at the
   replication storage boundary instead of `rustfs-replication` importing
   `rustfs-storage-api`;
+- `ReplicationCrateUtilsIndependence`: HTTP metadata keys, S3 header labels,
+  ETag trimming, and case-insensitive prefix matching used by replication wire
+  contracts are owned inside `rustfs-replication` instead of importing
+  `rustfs-utils`;
 - direct ECStore replication imports from `rustfs-replication` are limited to
   `*_boundary.rs` modules;
 - storage-api delete replication status/state helpers use the local
@@ -188,6 +192,10 @@ Required contracts before crate movement:
   replication delete/queue/operation helpers are owned in
   `crates/replication/src/storage_api.rs`, and `rustfs-replication` must not
   import or depend on `rustfs-storage-api`.
+- `ReplicationCrateUtilsIndependence`: replication-specific HTTP metadata,
+  header, ETag, and prefix helper contracts are owned in
+  `crates/replication/src/http.rs`, and `rustfs-replication` must not import or
+  depend on `rustfs-utils`.
 - `EcstoreReplicationBoundaryImports`: ECStore-side imports from
   `rustfs-replication` are concentrated in replication `*_boundary.rs` modules.
 - `RuntimeReplicationFacadeConsumers`: scanner, admin, storage-owner, and app

scripts/check_architecture_migration_rules.sh

@@ -88,6 +88,7 @@ require_source_contains "docs/architecture/ecstore-module-split-plan.md" "Ecstor
 require_source_contains "docs/architecture/ecstore-module-split-plan.md" "RuntimeReplicationFacadeConsumers" "ECStore split plan runtime replication facade consumer section"
 require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateFileMetaIndependence" "ECStore split plan replication crate filemeta independence section"
 require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateStorageApiIndependence" "ECStore split plan replication crate storage-api independence section"
+require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateUtilsIndependence" "ECStore split plan replication crate utils independence section"
 require_source_contains "docs/architecture/ecstore-module-split-plan.md" "StorageApiReplicationContracts" "ECStore split plan storage-api replication contract section"
 require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "## Facade Group Inventory" "ECStore facade inventory group section"
 require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "## External Consumer Boundaries" "ECStore facade inventory consumer boundary section"
@@ -213,6 +214,7 @@ RUSTFS_REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/rustfs_replication_facade
 RUNTIME_REPLICATION_DEPENDENCY_BYPASS_HITS_FILE="${TMP_DIR}/runtime_replication_dependency_bypass_hits.txt"
 REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE="${TMP_DIR}/replication_crate_filemeta_bypass_hits.txt"
 REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE="${TMP_DIR}/replication_crate_storage_api_bypass_hits.txt"
+REPLICATION_CRATE_UTILS_BYPASS_HITS_FILE="${TMP_DIR}/replication_crate_utils_bypass_hits.txt"
 REPLICATION_CONFIG_RULE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_config_rule_contract_backslide_hits.txt"
 REPLICATION_DELETE_WORKER_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_delete_worker_contract_backslide_hits.txt"
 REPLICATION_OPERATION_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_operation_contract_backslide_hits.txt"
@@ -2687,6 +2689,18 @@ if [[ -s "$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE" ]]; then
   report_failure "replication crate delete DTO contracts must not import or depend on rustfs-storage-api: $(paste -sd '; ' "$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE")"
 fi
 
+(
+  cd "$ROOT_DIR"
+  rg -n --with-filename 'rustfs_utils::|use\s+rustfs_utils\b|^rustfs-utils\b' \
+    crates/replication/src \
+    --glob '*.rs' \
+    crates/replication/Cargo.toml || true
+) >"$REPLICATION_CRATE_UTILS_BYPASS_HITS_FILE"
+
+if [[ -s "$REPLICATION_CRATE_UTILS_BYPASS_HITS_FILE" ]]; then
+  report_failure "replication crate HTTP/helper contracts must not import or depend on rustfs-utils: $(paste -sd '; ' "$REPLICATION_CRATE_UTILS_BYPASS_HITS_FILE")"
+fi
+
 (
   cd "$ROOT_DIR"
   replication_config_rule_status=0