diff --git a/Cargo.lock b/Cargo.lock index e94d4c69b..7b5830252 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9895,8 +9895,10 @@ dependencies = [ "byteorder", "rmp", "rmp-serde", + "rustfs-filemeta", "serde", "time", + "uuid", ] [[package]] diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index d51a6a26c..80622d2a6 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -13,9 +13,10 @@ // limitations under the License. pub(crate) use rustfs_filemeta::{ - MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, - REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, - ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, - ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, - parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, + REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE, + ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, + ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, + VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, + version_purge_statuses_map, }; +pub(crate) use rustfs_replication::{MrfOpKind, MrfReplicateEntry}; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 1e72818fb..1da8b7175 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -44,7 +44,6 @@ use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput}; use aws_sdk_s3::primitives::ByteStream; use aws_sdk_s3::types::{CompletedPart, ObjectLockLegalHoldStatus}; use aws_smithy_types::body::SdkBody; -use byteorder::ByteOrder; use futures::future::join_all; use futures::stream::StreamExt; use headers::{ @@ -55,7 +54,6 @@ use http::HeaderMap; use http_body::Frame; use http_body_util::StreamBody; use regex::Regex; -use rmp_serde; use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use rustfs_s3_types::EventName; use rustfs_utils::http::{ @@ -101,8 +99,8 @@ const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_ch pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT; pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION; -const MRF_META_FORMAT: u16 = 1; -const MRF_META_VERSION: u16 = 1; +pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT; +pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION; const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60); static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); @@ -189,49 +187,27 @@ fn content_matches(src: &ObjectInfo, tgt: &HeadObjectOutput) -> bool { src_etag.is_some() && src_etag == tgt_etag } -fn map_resync_error(err: rustfs_replication::resync::Error) -> Error { +fn map_replication_error(err: rustfs_replication::Error) -> Error { match err { - rustfs_replication::resync::Error::CorruptedFormat => Error::CorruptedFormat, - rustfs_replication::resync::Error::Other(err) => Error::other(err), + rustfs_replication::Error::CorruptedFormat => Error::CorruptedFormat, + rustfs_replication::Error::Other(err) => Error::other(err), } } pub(crate) fn encode_resync_file(status: &BucketReplicationResyncStatus) -> Result> { - rustfs_replication::encode_resync_file(status).map_err(map_resync_error) + rustfs_replication::encode_resync_file(status).map_err(map_replication_error) } pub(crate) fn decode_resync_file(data: &[u8]) -> Result { - rustfs_replication::decode_resync_file(data).map_err(map_resync_error) + rustfs_replication::decode_resync_file(data).map_err(map_replication_error) } pub(crate) fn encode_mrf_file(entries: &[MrfReplicateEntry]) -> Result> { - let payload = rmp_serde::to_vec_named(entries).map_err(|e| Error::other(e.to_string()))?; - let mut data = Vec::with_capacity(4 + payload.len()); - let mut fmt = [0u8; 2]; - byteorder::LittleEndian::write_u16(&mut fmt, MRF_META_FORMAT); - data.extend_from_slice(&fmt); - let mut ver = [0u8; 2]; - byteorder::LittleEndian::write_u16(&mut ver, MRF_META_VERSION); - data.extend_from_slice(&ver); - data.extend_from_slice(&payload); - Ok(data) + rustfs_replication::encode_mrf_file(entries).map_err(map_replication_error) } pub(crate) fn decode_mrf_file(data: &[u8]) -> Result> { - if data.len() <= 4 { - return Err(Error::CorruptedFormat); - } - let mut fmt = [0u8; 2]; - fmt.copy_from_slice(&data[0..2]); - if byteorder::LittleEndian::read_u16(&fmt) != MRF_META_FORMAT { - return Err(Error::CorruptedFormat); - } - let mut ver = [0u8; 2]; - ver.copy_from_slice(&data[2..4]); - if byteorder::LittleEndian::read_u16(&ver) != MRF_META_VERSION { - return Err(Error::CorruptedFormat); - } - rmp_serde::from_slice(&data[4..]).map_err(|e| Error::other(e.to_string())) + rustfs_replication::decode_mrf_file(data).map_err(map_replication_error) } static RESYNC_WORKER_COUNT: usize = 10; diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index dd61297b3..4e49b89bb 100644 --- a/crates/replication/Cargo.toml +++ b/crates/replication/Cargo.toml @@ -29,9 +29,13 @@ documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/" byteorder.workspace = true rmp.workspace = true rmp-serde.workspace = true +rustfs-filemeta.workspace = true serde.workspace = true time.workspace = true +[dev-dependencies] +uuid = { workspace = true, features = ["v4"] } + [lints] workspace = true diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index fce854ad6..0fc000147 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -12,9 +12,11 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod mrf; pub mod resync; +pub use mrf::{MrfOpKind, MrfReplicateEntry, decode_mrf_file, encode_mrf_file}; pub use resync::{ - BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, decode_resync_file, - encode_resync_file, + BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, + decode_resync_file, encode_resync_file, }; diff --git a/crates/replication/src/mrf.rs b/crates/replication/src/mrf.rs new file mode 100644 index 000000000..a125a846e --- /dev/null +++ b/crates/replication/src/mrf.rs @@ -0,0 +1,135 @@ +// 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 byteorder::{ByteOrder, LittleEndian}; + +use crate::{Error, Result}; + +pub use rustfs_filemeta::{MrfOpKind, MrfReplicateEntry}; + +pub const MRF_META_FORMAT: u16 = 1; +pub const MRF_META_VERSION: u16 = 1; + +pub fn encode_mrf_file(entries: &[MrfReplicateEntry]) -> Result> { + let payload = rmp_serde::to_vec_named(entries).map_err(|e| Error::Other(e.to_string()))?; + let mut data = Vec::with_capacity(4 + payload.len()); + let mut fmt = [0u8; 2]; + LittleEndian::write_u16(&mut fmt, MRF_META_FORMAT); + data.extend_from_slice(&fmt); + let mut ver = [0u8; 2]; + LittleEndian::write_u16(&mut ver, MRF_META_VERSION); + data.extend_from_slice(&ver); + data.extend_from_slice(&payload); + Ok(data) +} + +pub fn decode_mrf_file(data: &[u8]) -> Result> { + if data.len() <= 4 { + return Err(Error::CorruptedFormat); + } + let mut fmt = [0u8; 2]; + fmt.copy_from_slice(&data[0..2]); + if LittleEndian::read_u16(&fmt) != MRF_META_FORMAT { + return Err(Error::CorruptedFormat); + } + let mut ver = [0u8; 2]; + ver.copy_from_slice(&data[2..4]); + if LittleEndian::read_u16(&ver) != MRF_META_VERSION { + return Err(Error::CorruptedFormat); + } + rmp_serde::from_slice(&data[4..]).map_err(|e| Error::Other(e.to_string())) +} + +#[cfg(test)] +mod tests { + use super::*; + use uuid::Uuid; + + #[test] + fn mrf_file_round_trips_object_and_delete_entries() { + let obj_vid = Uuid::new_v4(); + let del_vid = Uuid::new_v4(); + let entries = vec![ + MrfReplicateEntry { + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + version_id: Some(obj_vid), + retry_count: 2, + size: 1024, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + }, + MrfReplicateEntry { + bucket: "bucket-a".to_string(), + object: "delete-a".to_string(), + version_id: None, + retry_count: 0, + size: 0, + op: MrfOpKind::Delete, + delete_marker_version_id: Some(del_vid), + delete_marker: true, + }, + ]; + + let encoded = encode_mrf_file(&entries).expect("mrf file should encode"); + let decoded = decode_mrf_file(&encoded).expect("mrf file should decode"); + + assert_eq!(decoded.len(), 2); + assert_eq!(decoded[0].version_id, Some(obj_vid)); + assert_eq!(decoded[0].op, MrfOpKind::Object); + assert_eq!(decoded[1].delete_marker_version_id, Some(del_vid)); + assert_eq!(decoded[1].op, MrfOpKind::Delete); + assert!(decoded[1].delete_marker); + } + + #[test] + fn mrf_legacy_file_without_op_decodes_as_object() { + let mut payload = Vec::new(); + rmp::encode::write_array_len(&mut payload, 1).expect("array len should encode"); + rmp::encode::write_map_len(&mut payload, 4).expect("map len should encode"); + rmp::encode::write_str(&mut payload, "bucket").expect("bucket key should encode"); + rmp::encode::write_str(&mut payload, "old-bucket").expect("bucket value should encode"); + rmp::encode::write_str(&mut payload, "object").expect("object key should encode"); + rmp::encode::write_str(&mut payload, "old-key").expect("object value should encode"); + rmp::encode::write_str(&mut payload, "retryCount").expect("retry key should encode"); + rmp::encode::write_i32(&mut payload, 2).expect("retry value should encode"); + rmp::encode::write_str(&mut payload, "size").expect("size key should encode"); + rmp::encode::write_i64(&mut payload, 100).expect("size value should encode"); + + let mut data = Vec::with_capacity(4 + payload.len()); + data.extend_from_slice(&MRF_META_FORMAT.to_le_bytes()); + data.extend_from_slice(&MRF_META_VERSION.to_le_bytes()); + data.extend_from_slice(&payload); + + let decoded = decode_mrf_file(&data).expect("legacy mrf file should decode"); + + assert_eq!(decoded.len(), 1); + assert_eq!(decoded[0].bucket, "old-bucket"); + assert_eq!(decoded[0].object, "old-key"); + assert_eq!(decoded[0].retry_count, 2); + assert_eq!(decoded[0].size, 100); + assert_eq!(decoded[0].op, MrfOpKind::Object); + } + + #[test] + fn mrf_file_rejects_invalid_header() { + let mut data = Vec::new(); + data.extend_from_slice(&2u16.to_le_bytes()); + data.extend_from_slice(&MRF_META_VERSION.to_le_bytes()); + data.push(0x90); + + assert!(matches!(decode_mrf_file(&data), Err(Error::CorruptedFormat))); + } +} diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index a9ff37bc0..9918a5963 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -200,6 +200,7 @@ EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundar REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt" REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt" REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_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" APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/app_replication_dto_boundary_bypass_hits.txt" @@ -2552,6 +2553,17 @@ 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" + 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' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' || true +) >"$REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE" + +if [[ -s "$REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE" ]]; then + report_failure "MRF wire format must stay in crates/replication with only ECStore error-mapping wrappers: $(paste -sd '; ' "$REPLICATION_MRF_WIRE_FORMAT_BACKSLIDE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" {