refactor(replication): move mrf wire format to replication crate (#4157)

* refactor(replication): move mrf wire format to replication crate * refactor(replication): match re-exported error variants
2026-07-26 08:18:18 +00:00 · 2026-07-02 04:27:53 +08:00
parent 8f01012071
commit 2de38d3e5c
7 changed files with 172 additions and 40 deletions
@@ -9895,8 +9895,10 @@ dependencies = [
 "byteorder",
 "rmp",
 "rmp-serde",
+ "rustfs-filemeta",
 "serde",
 "time",
+ "uuid",
 ]

 [[package]]
@@ -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};
@@ -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<Vec<u8>> {
-    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<BucketReplicationResyncStatus> {
-    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<Vec<u8>> {
-    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<Vec<MrfReplicateEntry>> {
-    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;
@@ -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

@@ -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,
 };
@@ -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<Vec<u8>> {
+    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<Vec<MrfReplicateEntry>> {
+    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)));
+    }
+}
@@ -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"
  {