diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index cc325eb93..e2b004a2c 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -56,9 +56,11 @@ use http_body_util::StreamBody; #[cfg(test)] use rmp_serde; use rustfs_replication::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ResyncOpts, TargetReplicationResyncStatus, + BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationSourceObject, + ReplicationTargetObject, ResyncOpts, TargetReplicationResyncStatus, content_matches_by_etag, is_retryable_delete_replication_head_error, is_version_delete_replication, is_version_id_mismatch, - resync_state_accepts_update, should_count_head_proxy_failure, should_retry_delete_marker_purge, + replication_action_for_target, resync_state_accepts_update, should_count_head_proxy_failure, + should_retry_delete_marker_purge, target_is_newer_than_source_null_version, }; use rustfs_s3_types::EventName; use rustfs_utils::http::{ @@ -70,7 +72,6 @@ use rustfs_utils::http::{ SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_SSEC_CRC, get_header_map, get_str, has_internal_suffix, insert_header_map, insert_str, is_internal_key, }; -use rustfs_utils::string::strings_has_prefix_fold; use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; use s3s::dto::ReplicationConfiguration; use serde::Deserialize; @@ -158,12 +159,36 @@ async fn head_object_fallback( } } -// Version IDs differ by design on this path (RustFS UUID vs AWS alphanumeric), so -// compare only ETags. Equal ETags mean identical content; version ID is irrelevant. -fn content_matches(src: &ObjectInfo, tgt: &HeadObjectOutput) -> bool { - let src_etag = src.etag.as_deref().map(rustfs_utils::path::trim_etag); - let tgt_etag = tgt.e_tag.as_deref().map(rustfs_utils::path::trim_etag); - src_etag.is_some() && src_etag == tgt_etag +fn head_object_last_modified(oi: &HeadObjectOutput) -> Option { + oi.last_modified + .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) +} + +fn replication_source_object(oi: &ObjectInfo) -> ReplicationSourceObject<'_> { + ReplicationSourceObject { + mod_time: oi.mod_time, + version_id: oi.version_id.map(|version_id| version_id.to_string()), + etag: oi.etag.as_deref(), + actual_size: oi.get_actual_size().unwrap_or_default(), + delete_marker: oi.delete_marker, + content_type: oi.content_type.as_deref(), + content_encoding: oi.content_encoding.as_deref(), + user_tags: oi.user_tags.as_str(), + user_defined: oi.user_defined.as_ref(), + } +} + +fn replication_target_object(oi: &HeadObjectOutput) -> ReplicationTargetObject<'_> { + ReplicationTargetObject { + last_modified: head_object_last_modified(oi), + version_id: oi.version_id.as_deref(), + etag: oi.e_tag.as_deref(), + content_length: oi.content_length.unwrap_or_default(), + delete_marker: oi.delete_marker.unwrap_or_default(), + content_type: oi.content_type.as_deref(), + metadata: oi.metadata.as_ref(), + tag_count: oi.tag_count.unwrap_or_default(), + } } fn map_replication_error(err: rustfs_replication::Error) -> Error { @@ -2557,7 +2582,11 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await { Ok(oi) => { - replication_action = get_replication_action(&object_info, &oi, self.op_type); + replication_action = replication_action_for_target( + &replication_source_object(&object_info), + &replication_target_object(&oi), + self.op_type, + ); if replication_action == ReplicationAction::None { rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; @@ -2572,7 +2601,12 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } else if is_version_id_format_mismatch(&e) { // Version-ID format mismatch: retry without versionId and compare ETags. match head_object_fallback(&bucket, &tgt_client, &object).await { - Ok(Some(oi)) if content_matches(&object_info, &oi) => { + Ok(Some(oi)) + if content_matches_by_etag( + &replication_source_object(&object_info), + &replication_target_object(&oi), + ) => + { rinfo.replication_status = ReplicationStatusType::Completed; rinfo.replication_resynced = true; rinfo.replication_action = ReplicationAction::None; @@ -2874,11 +2908,18 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await { Ok(oi) => { - replication_action = get_replication_action(&object_info, &oi, self.op_type); + replication_action = replication_action_for_target( + &replication_source_object(&object_info), + &replication_target_object(&oi), + self.op_type, + ); rinfo.replication_status = ReplicationStatusType::Completed; if replication_action == ReplicationAction::None { if self.op_type == ReplicationType::ExistingObject - && target_is_newer_than_source_null_version(&object_info, &oi) + && target_is_newer_than_source_null_version( + &replication_source_object(&object_info), + &replication_target_object(&oi), + ) { warn!( event = EVENT_RESYNC_RUNTIME_SKIPPED, @@ -2932,7 +2973,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // Version-ID format mismatch: retry without versionId and compare ETags. match head_object_fallback(&bucket, &tgt_client, &object).await { Ok(Some(oi)) => { - replication_action = if content_matches(&object_info, &oi) { + replication_action = if content_matches_by_etag( + &replication_source_object(&object_info), + &replication_target_object(&oi), + ) { ReplicationAction::None } else { ReplicationAction::All @@ -3520,115 +3564,6 @@ async fn replicate_object_with_multipart(ctx: MultipartR Ok(()) } -fn target_is_newer_than_source_null_version(oi1: &ObjectInfo, oi2: &HeadObjectOutput) -> bool { - oi2.last_modified - .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) - .is_some_and(|target_mod_time| target_mod_time > oi1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)) - && oi1.version_id.is_none() -} - -fn get_replication_action(oi1: &ObjectInfo, oi2: &HeadObjectOutput, op_type: ReplicationType) -> ReplicationAction { - if op_type == ReplicationType::ExistingObject && target_is_newer_than_source_null_version(oi1, oi2) { - return ReplicationAction::None; - } - - let size = oi1.get_actual_size().unwrap_or_default(); - - // Normalize ETags by removing quotes before comparison (PR #592 compatibility) - let oi1_etag = oi1.etag.as_ref().map(|e| rustfs_utils::path::trim_etag(e)); - let oi2_etag = oi2.e_tag.as_ref().map(|e| rustfs_utils::path::trim_etag(e)); - - if oi1_etag != oi2_etag - || oi1.version_id.map(|v| v.to_string()) != oi2.version_id - || size != oi2.content_length.unwrap_or_default() - || oi1.delete_marker != oi2.delete_marker.unwrap_or_default() - || oi1.mod_time - != oi2 - .last_modified - .map(|dt| OffsetDateTime::from_unix_timestamp(dt.secs()).unwrap_or(OffsetDateTime::UNIX_EPOCH)) - { - return ReplicationAction::All; - } - - if oi1.content_type != oi2.content_type { - return ReplicationAction::Metadata; - } - - let empty_metadata = HashMap::new(); - let metadata = oi2.metadata.as_ref().unwrap_or(&empty_metadata); - - if let Some(content_encoding) = &oi1.content_encoding { - if let Some(enc) = metadata - .get(CONTENT_ENCODING) - .or_else(|| metadata.get(&CONTENT_ENCODING.to_lowercase())) - { - if enc != content_encoding { - return ReplicationAction::Metadata; - } - } else { - return ReplicationAction::Metadata; - } - } - - let oi1_tags = ReplicationTagFilter::decode_tags_to_map(&oi1.user_tags); - let oi2_tags = - ReplicationTagFilter::decode_tags_to_map(metadata.get(AMZ_OBJECT_TAGGING).cloned().unwrap_or_default().as_str()); - - if (oi2.tag_count.unwrap_or_default() > 0 && oi1_tags != oi2_tags) - || oi2.tag_count.unwrap_or_default() != oi1_tags.len() as i32 - { - return ReplicationAction::Metadata; - } - - // Compare only necessary headers - let compare_keys = vec![ - "Expires", - "Cache-Control", - "Content-Language", - "Content-Disposition", - "X-Amz-Object-Lock-Mode", - "X-Amz-Object-Lock-Retain-Until-Date", - "X-Amz-Object-Lock-Legal-Hold", - "X-Amz-Website-Redirect-Location", - "X-Amz-Meta-", - ]; - - // compare metadata on both maps to see if meta is identical - let mut compare_meta1 = HashMap::new(); - for (k, v) in oi1.user_defined.iter() { - let mut found = false; - for prefix in &compare_keys { - if strings_has_prefix_fold(k, prefix) { - found = true; - break; - } - } - if found { - compare_meta1.insert(k.to_lowercase(), v.clone()); - } - } - - let mut compare_meta2 = HashMap::new(); - for (k, v) in metadata { - let mut found = false; - for prefix in &compare_keys { - if strings_has_prefix_fold(k.to_string().as_str(), prefix) { - found = true; - break; - } - } - if found { - compare_meta2.insert(k.to_lowercase(), v.clone()); - } - } - - if compare_meta1 != compare_meta2 { - return ReplicationAction::Metadata; - } - - ReplicationAction::None -} - #[cfg(test)] mod tests { use super::*; @@ -3907,7 +3842,11 @@ mod tests { let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(10)).build(); assert_eq!( - get_replication_action(&source, &target, ReplicationType::ExistingObject), + replication_action_for_target( + &replication_source_object(&source), + &replication_target_object(&target), + ReplicationType::ExistingObject, + ), ReplicationAction::All, "a newer source null version must not be skipped during existing-object replication" ); @@ -3923,7 +3862,11 @@ mod tests { let target = HeadObjectOutput::builder().last_modified(DateTime::from_secs(20)).build(); assert_eq!( - get_replication_action(&source, &target, ReplicationType::ExistingObject), + replication_action_for_target( + &replication_source_object(&source), + &replication_target_object(&target), + ReplicationType::ExistingObject, + ), ReplicationAction::None, "a newer target null-version object should not be overwritten by existing-object replication" ); @@ -4213,11 +4156,14 @@ mod tests { }; let tgt_match = HeadObjectOutput::builder().e_tag("\"abc123\"").build(); - assert!(content_matches(&src, &tgt_match), "identical ETags must match"); + assert!( + content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_match)), + "identical ETags must match" + ); let tgt_unquoted_match = HeadObjectOutput::builder().e_tag("abc123").build(); assert!( - content_matches(&src, &tgt_unquoted_match), + content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_unquoted_match)), "quoted and unquoted ETags with identical values must match" ); @@ -4227,21 +4173,30 @@ mod tests { .version_id("aws-alphanumeric-id") .build(); assert!( - content_matches(&src, &tgt_different_version), + content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_different_version)), "matching ETags with different version IDs must still match" ); let tgt_different_content = HeadObjectOutput::builder().e_tag("\"def456\"").build(); - assert!(!content_matches(&src, &tgt_different_content), "different ETags must not match"); + assert!( + !content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_different_content)), + "different ETags must not match" + ); let src_no_etag = ObjectInfo { etag: None, ..Default::default() }; - assert!(!content_matches(&src_no_etag, &tgt_match), "missing source ETag must not match"); + assert!( + !content_matches_by_etag(&replication_source_object(&src_no_etag), &replication_target_object(&tgt_match)), + "missing source ETag must not match" + ); let tgt_no_etag = HeadObjectOutput::builder().build(); - assert!(!content_matches(&src, &tgt_no_etag), "missing target ETag must not match"); + assert!( + !content_matches_by_etag(&replication_source_object(&src), &replication_target_object(&tgt_no_etag)), + "missing target ETag must not match" + ); } #[test] diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index 9407b2446..5a582c0b7 100644 --- a/crates/replication/Cargo.toml +++ b/crates/replication/Cargo.toml @@ -31,7 +31,7 @@ rmp.workspace = true rmp-serde.workspace = true rustfs-filemeta.workspace = true rustfs-storage-api.workspace = true -rustfs-utils = { workspace = true, features = ["http"] } +rustfs-utils = { workspace = true, features = ["http", "path", "string"] } s3s.workspace = true serde.workspace = true time.workspace = true diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 1f8984c3e..b5634c7fa 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -15,6 +15,7 @@ pub mod config; pub mod delete; pub mod mrf; +pub mod object; pub mod operation; pub mod queue; pub mod resync; @@ -29,6 +30,10 @@ pub use delete::{ should_retry_delete_marker_purge, }; pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file}; +pub use object::{ + ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, + target_is_newer_than_source_null_version, +}; pub use operation::{MustReplicateOptions, is_ssec_encrypted}; pub use queue::{ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission}; pub use resync::{ diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs new file mode 100644 index 000000000..7596df784 --- /dev/null +++ b/crates/replication/src/object.rs @@ -0,0 +1,271 @@ +// 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 crate::tagging::ReplicationTagFilter; +use rustfs_filemeta::{ReplicationAction, ReplicationType}; +use rustfs_utils::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, +}; +use rustfs_utils::path::trim_etag; +use rustfs_utils::string::strings_has_prefix_fold; +use std::collections::HashMap; +use time::OffsetDateTime; + +const AMZ_META_PREFIX: &str = "X-Amz-Meta-"; +const REPLICATION_METADATA_COMPARE_KEYS: [&str; 9] = [ + EXPIRES, + CACHE_CONTROL, + CONTENT_LANGUAGE, + CONTENT_DISPOSITION, + AMZ_OBJECT_LOCK_MODE, + AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, + AMZ_OBJECT_LOCK_LEGAL_HOLD, + AMZ_WEBSITE_REDIRECT_LOCATION, + AMZ_META_PREFIX, +]; + +#[derive(Debug, Clone)] +pub struct ReplicationSourceObject<'a> { + pub mod_time: Option, + pub version_id: Option, + pub etag: Option<&'a str>, + pub actual_size: i64, + pub delete_marker: bool, + pub content_type: Option<&'a str>, + pub content_encoding: Option<&'a str>, + pub user_tags: &'a str, + pub user_defined: &'a HashMap, +} + +#[derive(Debug, Clone)] +pub struct ReplicationTargetObject<'a> { + pub last_modified: Option, + pub version_id: Option<&'a str>, + pub etag: Option<&'a str>, + pub content_length: i64, + pub delete_marker: bool, + pub content_type: Option<&'a str>, + pub metadata: Option<&'a HashMap>, + pub tag_count: i32, +} + +pub fn content_matches_by_etag(source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>) -> bool { + let source_etag = source.etag.map(trim_etag); + let target_etag = target.etag.map(trim_etag); + source_etag.is_some() && source_etag == target_etag +} + +pub fn target_is_newer_than_source_null_version( + source: &ReplicationSourceObject<'_>, + target: &ReplicationTargetObject<'_>, +) -> bool { + target + .last_modified + .is_some_and(|target_mod_time| target_mod_time > source.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH)) + && source.version_id.is_none() +} + +pub fn replication_action_for_target( + source: &ReplicationSourceObject<'_>, + target: &ReplicationTargetObject<'_>, + op_type: ReplicationType, +) -> ReplicationAction { + if op_type == ReplicationType::ExistingObject && target_is_newer_than_source_null_version(source, target) { + return ReplicationAction::None; + } + + if source.etag.map(trim_etag) != target.etag.map(trim_etag) + || source.version_id.as_deref() != target.version_id + || source.actual_size != target.content_length + || source.delete_marker != target.delete_marker + || source.mod_time != target.last_modified + { + return ReplicationAction::All; + } + + if source.content_type != target.content_type { + return ReplicationAction::Metadata; + } + + if content_encoding_differs(source, target) { + return ReplicationAction::Metadata; + } + + if tag_metadata_differs(source, target) { + return ReplicationAction::Metadata; + } + + if comparable_metadata(Some(source.user_defined)) != comparable_metadata(target.metadata) { + return ReplicationAction::Metadata; + } + + ReplicationAction::None +} + +fn content_encoding_differs(source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>) -> bool { + if let Some(content_encoding) = source.content_encoding { + return target + .metadata + .and_then(|metadata| { + metadata + .get(CONTENT_ENCODING) + .or_else(|| metadata.get(&CONTENT_ENCODING.to_lowercase())) + }) + .is_none_or(|enc| enc != content_encoding); + } + false +} + +fn tag_metadata_differs(source: &ReplicationSourceObject<'_>, target: &ReplicationTargetObject<'_>) -> bool { + let source_tags = ReplicationTagFilter::decode_tags_to_map(source.user_tags); + let target_tags = ReplicationTagFilter::decode_tags_to_map( + target + .metadata + .and_then(|metadata| metadata.get(AMZ_OBJECT_TAGGING)) + .cloned() + .unwrap_or_default() + .as_str(), + ); + let source_tag_count = match i32::try_from(source_tags.len()) { + Ok(count) => count, + Err(_) => i32::MAX, + }; + + (target.tag_count > 0 && source_tags != target_tags) || target.tag_count != source_tag_count +} + +fn comparable_metadata(metadata: Option<&HashMap>) -> HashMap { + let mut comparable = HashMap::new(); + for (key, value) in metadata.into_iter().flatten() { + if REPLICATION_METADATA_COMPARE_KEYS + .iter() + .any(|prefix| strings_has_prefix_fold(key, prefix)) + { + comparable.insert(key.to_lowercase(), value.clone()); + } + } + comparable +} + +#[cfg(test)] +mod tests { + use super::{ + ReplicationSourceObject, ReplicationTargetObject, content_matches_by_etag, replication_action_for_target, + target_is_newer_than_source_null_version, + }; + use rustfs_filemeta::{ReplicationAction, ReplicationType}; + use std::collections::HashMap; + use time::{Duration, OffsetDateTime}; + + fn source_object(user_defined: &HashMap) -> ReplicationSourceObject<'_> { + ReplicationSourceObject { + mod_time: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + version_id: Some("source-version".to_string()), + etag: Some("\"abc\""), + actual_size: 10, + delete_marker: false, + content_type: Some("text/plain"), + content_encoding: None, + user_tags: "a=1", + user_defined, + } + } + + fn target_object(metadata: &HashMap) -> ReplicationTargetObject<'_> { + ReplicationTargetObject { + last_modified: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(10)), + version_id: Some("source-version"), + etag: Some("abc"), + content_length: 10, + delete_marker: false, + content_type: Some("text/plain"), + metadata: Some(metadata), + tag_count: 1, + } + } + + #[test] + fn content_matches_by_etag_ignores_version_ids() { + let source_metadata = HashMap::new(); + let target_metadata = HashMap::new(); + let source = ReplicationSourceObject { + version_id: Some("source-version".to_string()), + etag: Some("\"abc\""), + user_defined: &source_metadata, + ..source_object(&source_metadata) + }; + let target = ReplicationTargetObject { + version_id: Some("different-version"), + etag: Some("abc"), + metadata: Some(&target_metadata), + ..target_object(&target_metadata) + }; + + assert!(content_matches_by_etag(&source, &target)); + } + + #[test] + fn target_newer_null_version_skips_existing_object_replication() { + let source_metadata = HashMap::new(); + let target_metadata = HashMap::new(); + let source = ReplicationSourceObject { + version_id: None, + ..source_object(&source_metadata) + }; + let target = ReplicationTargetObject { + last_modified: Some(OffsetDateTime::UNIX_EPOCH + Duration::seconds(20)), + ..target_object(&target_metadata) + }; + + assert!(target_is_newer_than_source_null_version(&source, &target)); + assert_eq!( + replication_action_for_target(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::None + ); + } + + #[test] + fn replication_action_detects_content_and_metadata_differences() { + let mut source_metadata = HashMap::new(); + source_metadata.insert("Cache-Control".to_string(), "max-age=1".to_string()); + let mut target_metadata = HashMap::new(); + target_metadata.insert("X-Amz-Tagging".to_string(), "a=1".to_string()); + target_metadata.insert("Cache-Control".to_string(), "max-age=1".to_string()); + + let source = source_object(&source_metadata); + let target = target_object(&target_metadata); + assert_eq!( + replication_action_for_target(&source, &target, ReplicationType::ExistingObject), + ReplicationAction::None + ); + + let changed_content = ReplicationTargetObject { + content_length: 11, + ..target_object(&target_metadata) + }; + assert_eq!( + replication_action_for_target(&source, &changed_content, ReplicationType::ExistingObject), + ReplicationAction::All + ); + + let mut changed_target_metadata = target_metadata.clone(); + changed_target_metadata.insert("Cache-Control".to_string(), "max-age=2".to_string()); + let changed_metadata = target_object(&changed_target_metadata); + assert_eq!( + replication_action_for_target(&source, &changed_metadata, ReplicationType::ExistingObject), + ReplicationAction::Metadata + ); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 5f09d9fef..b182331eb 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -206,6 +206,7 @@ REPLICATION_QUEUE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_queue_con REPLICATION_STATS_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_stats_contract_backslide_hits.txt" REPLICATION_RUNTIME_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_runtime_contract_backslide_hits.txt" REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_contract_backslide_hits.txt" +REPLICATION_OBJECT_COMPARE_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_object_compare_contract_backslide_hits.txt" REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_mrf_wire_format_backslide_hits.txt" STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/storage_replication_handle_boundary_bypass_hits.txt" ADMIN_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/admin_replication_dto_boundary_bypass_hits.txt" @@ -2654,6 +2655,21 @@ if [[ -s "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then report_failure "resync DTO contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE")" fi +( + cd "$ROOT_DIR" + replication_object_compare_status=0 + rg -n --with-filename '^\s*(?:pub(?:\([^)]*\))?\s+)?(?:(?:struct)\s+(?:ReplicationSourceObject|ReplicationTargetObject)|fn\s+(?:content_matches_by_etag|content_matches|target_is_newer_than_source_null_version|replication_action_for_target|get_replication_action))\b' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' >"$REPLICATION_OBJECT_COMPARE_CONTRACT_BACKSLIDE_HITS_FILE" || replication_object_compare_status=$? + if [[ "$replication_object_compare_status" -ne 0 && "$replication_object_compare_status" -ne 1 ]]; then + exit "$replication_object_compare_status" + fi +) + +if [[ -s "$REPLICATION_OBJECT_COMPARE_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then + report_failure "replication object comparison contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_OBJECT_COMPARE_CONTRACT_BACKSLIDE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" rg -n --with-filename 'rmp_serde::(to_vec_named|from_slice)|LittleEndian::(write_u16|read_u16)|const\s+MRF_META_(FORMAT|VERSION):\s+u16\s*=\s*1\b' \