fix(bucket-repl): persist MRF retry queue to disk and reload on startup (#3456)

* fix(bucket-repl): persist MRF retry queue to disk and reload on startup

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* fix(bucket-repl): address three blocking MRF issues from review

1. MRF replay loses delete operations — add `MrfOpKind` discriminator to
   `MrfReplicateEntry` (Object | Delete, default=Object for backward
   compat).  `DeletedObjectReplicationInfo::to_mrf_entry` now persists
   `op=Delete`, `version_id`, `delete_marker_version_id`, and
   `delete_marker`.  `start_mrf_processor` branches on `op`: delete
   entries skip `get_object_info` and replay via
   `schedule_replication_delete` with `ReplicationType::Heal`; object
   entries follow the existing heal path.

2. `flush_mrf_to_disk` cleared the in-memory batch even on encode/write
   failure — changed return type to `bool` and callers now only
   `pending.clear()` on `true`, so a transient storage error retries
   on the next tick instead of silently dropping the batch.

3. Add focused tests: encode/decode roundtrips for object, delete-marker,
   versioned-delete, and mixed-batch entries; a routing test confirming
   op-kind propagates correctly and that the default is Object for
   legacy files; a legacy-compat test verifying old entries round-trip
   cleanly through the new format.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* style: fix clippy redundant-clone in MRF tests

Replace &[entry.clone()] with std::slice::from_ref(&entry) in two
encode_mrf_file call sites flagged by clippy's redundant_clone lint
under --all-targets --features rio-v2.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* test(bucket-repl): strengthen legacy MRF compat test with hand-built msgpack

The previous mrf_legacy_file_without_op_field_decoded_as_object test
round-tripped through encode_mrf_file, so it exercised the new format
and never touched a truly-legacy payload.

Replace it with a hand-built msgpack payload that genuinely omits the
"op", "deleteMarker", and "deleteMarkerVersionID" keys — exactly what
the old binary would have written before MrfOpKind existed.  The test
now fails if #[serde(default)] is removed from the op field, which
proves real backward compatibility rather than round-trip stability.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>
Co-authored-by: houseme <housemecn@gmail.com>

crates/filemeta/src/replication.rs

@@ -534,7 +534,20 @@ impl ReplicatedInfos {
     }
 }
 
-#[derive(Serialize, Deserialize, Debug)]
+/// Distinguishes the kind of operation stored in [`MrfReplicateEntry`].
+///
+/// Old serialized files lack the `op` key; `default` maps to `Object`, which preserves
+/// the pre-existing replay behaviour for entries written before this field existed.
+#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum MrfOpKind {
+    #[default]
+    #[serde(rename = "object")]
+    Object,
+    #[serde(rename = "delete")]
+    Delete,
+}
+
+#[derive(Serialize, Deserialize, Debug, Clone)]
 pub struct MrfReplicateEntry {
     #[serde(rename = "bucket")]
     pub bucket: String,
@@ -542,14 +555,30 @@ pub struct MrfReplicateEntry {
     #[serde(rename = "object")]
     pub object: String,
 
-    #[serde(skip_serializing, skip_deserializing)]
+    // Persisted so recovery after restart can replay the exact version.
+    // Old serialized files lack this key; `default` fills in None safely.
+    #[serde(rename = "versionID", skip_serializing_if = "Option::is_none", default)]
     pub version_id: Option<Uuid>,
 
     #[serde(rename = "retryCount")]
     pub retry_count: i32,
 
-    #[serde(skip_serializing, skip_deserializing)]
+    #[serde(rename = "size", default)]
     pub size: i64,
+
+    // Operation kind. Old files lack this key; default=Object preserves existing behaviour.
+    #[serde(rename = "op", default)]
+    pub op: MrfOpKind,
+
+    // For delete entries: the delete-marker version id (distinct from version_id, which is
+    // the version being purged). Old files lack this; default=None is correct.
+    #[serde(rename = "deleteMarkerVersionID", skip_serializing_if = "Option::is_none", default)]
+    pub delete_marker_version_id: Option<Uuid>,
+
+    // For delete entries: whether this is a delete-marker vs a versioned-object delete.
+    // Old files lack this; default=false is correct.
+    #[serde(rename = "deleteMarker", default)]
+    pub delete_marker: bool,
 }
 
 pub trait ReplicationWorkerOperation: Any + Send + Sync {
@@ -748,6 +777,9 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo {
             version_id: self.version_id,
             retry_count: self.retry_count as i32,
             size: self.size,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
         }
     }
 
@@ -795,6 +827,9 @@ impl ReplicateObjectInfo {
             version_id: self.version_id,
             retry_count: self.retry_count as i32,
             size: self.size,
+            op: MrfOpKind::Object,
+            delete_marker_version_id: None,
+            delete_marker: false,
         }
     }
 }