fix(bucket-repl): honor op_type in replicate_object so ExistingObject… (#3420)

fix(bucket-repl): honor op_type in replicate_object so ExistingObject resync respects DISABLED targets

replicate_object was calling filter_target_arns with hard-coded
op_type: Object and existing_object: false regardless of what was
stored in roi.op_type. This meant that a resync worker setting
roi.op_type = ExistingObject (resync_bucket, line 889) had no effect
on target filtering: all configured targets were included, even ones
whose rule had ExistingObjectReplicationStatus::DISABLED.

Fix: pass op_type: roi.op_type and derive existing_object from it
(true only for ExistingObject, not Heal — Heal intentionally bypasses
the existing-object opt-out to repair past failures).

Also add warn! logs at all four MRF channel-overflow sites that were
previously silently returning Missed with no observability.

Verified with a live two-instance test: after resync, objects reached
the ENABLED target and were correctly blocked from the DISABLED target.

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

crates/ecstore/src/bucket/replication/config.rs

@@ -318,4 +318,77 @@ mod tests {
 
         assert_eq!(arns, vec!["arn:legacy:target".to_string()]);
     }
+
+    fn replication_rule_existing_object_disabled(id: &str, arn: &str) -> ReplicationRule {
+        ReplicationRule {
+            delete_marker_replication: Some(DeleteMarkerReplication::default()),
+            delete_replication: None,
+            destination: Destination {
+                bucket: arn.to_string(),
+                ..Default::default()
+            },
+            existing_object_replication: Some(ExistingObjectReplication {
+                status: ExistingObjectReplicationStatus::from_static(ExistingObjectReplicationStatus::DISABLED),
+            }),
+            filter: None,
+            id: Some(id.to_string()),
+            prefix: Some(String::new()),
+            priority: Some(1),
+            source_selection_criteria: None,
+            status: ReplicationRuleStatus::from_static(ReplicationRuleStatus::ENABLED),
+        }
+    }
+
+    // Regression test for BUG-3: replicate_object was calling filter_target_arns with
+    // existing_object:false regardless of op_type, letting ExistingObject resync operations
+    // fan out to targets whose rule has ExistingObjectReplicationStatus::DISABLED.
+    #[test]
+    fn filter_target_arns_excludes_disabled_existing_object_target_for_existing_object_op() {
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule("rule-enabled", "arn:target:enabled"),
+                replication_rule_existing_object_disabled("rule-disabled", "arn:target:disabled"),
+            ],
+        };
+
+        let arns = config.filter_target_arns(&ObjectOpts {
+            name: "object".to_string(),
+            op_type: ReplicationType::ExistingObject,
+            existing_object: true,
+            ..Default::default()
+        });
+
+        assert_eq!(arns.len(), 1, "only the ENABLED target should be returned for ExistingObject ops");
+        assert!(arns.contains(&"arn:target:enabled".to_string()));
+        assert!(!arns.contains(&"arn:target:disabled".to_string()));
+    }
+
+    // Heal operations intentionally bypass ExistingObjectReplicationStatus — healing a past
+    // failure is not subject to the existing-object opt-out.
+    #[test]
+    fn filter_target_arns_includes_disabled_existing_object_target_for_heal_op() {
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![
+                replication_rule("rule-enabled", "arn:target:enabled"),
+                replication_rule_existing_object_disabled("rule-disabled", "arn:target:disabled"),
+            ],
+        };
+
+        let arns = config.filter_target_arns(&ObjectOpts {
+            name: "object".to_string(),
+            op_type: ReplicationType::Heal,
+            existing_object: false,
+            ..Default::default()
+        });
+
+        assert_eq!(
+            arns.len(),
+            2,
+            "Heal ops must reach all targets regardless of existing_object_replication setting"
+        );
+        assert!(arns.contains(&"arn:target:enabled".to_string()));
+        assert!(arns.contains(&"arn:target:disabled".to_string()));
+    }
 }

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -56,7 +56,7 @@ use tokio::sync::mpsc::Sender;
 use tokio::task::JoinHandle;
 use tokio::time::Duration;
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, info, instrument};
+use tracing::{debug, info, instrument, warn};
 
 const LOG_COMPONENT_ECSTORE: &str = "ecstore";
 const LOG_SUBSYSTEM_REPLICATION: &str = "replication";
@@ -65,6 +65,7 @@ const EVENT_REPLICATION_WORKER_RESIZED: &str = "replication_worker_resized";
 const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure";
 const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped";
 const EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED: &str = "replication_config_lookup_skipped";
+const EVENT_REPLICATION_MRF_QUEUE_OVERFLOW: &str = "replication_mrf_queue_overflow";
 
 // Worker limits
 pub const WORKER_MAX_LIMIT: usize = 500;
@@ -588,6 +589,14 @@ impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
                     let admission = if self.mrf_save_tx.try_send(ri.to_mrf_entry()).is_ok() {
                         ReplicationQueueAdmission::Queued
                     } else {
+                        warn!(
+                            event = EVENT_REPLICATION_MRF_QUEUE_OVERFLOW,
+                            component = LOG_COMPONENT_ECSTORE,
+                            subsystem = LOG_SUBSYSTEM_REPLICATION,
+                            bucket = %ri.bucket,
+                            object = %ri.name,
+                            "MRF queue full — large-worker replication failure entry dropped and will not be retried"
+                        );
                         ReplicationQueueAdmission::Missed
                     };
 
@@ -627,6 +636,14 @@ impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
         let admission = if self.mrf_save_tx.try_send(ri.to_mrf_entry()).is_ok() {
             ReplicationQueueAdmission::Queued
         } else {
+            warn!(
+                event = EVENT_REPLICATION_MRF_QUEUE_OVERFLOW,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %ri.bucket,
+                object = %ri.name,
+                "MRF queue full — replication failure entry dropped and will not be retried"
+            );
             ReplicationQueueAdmission::Missed
         };
 
@@ -703,6 +720,14 @@ impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
         let admission = if self.mrf_save_tx.try_send(doi.to_mrf_entry()).is_ok() {
             ReplicationQueueAdmission::Queued
         } else {
+            warn!(
+                event = EVENT_REPLICATION_MRF_QUEUE_OVERFLOW,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %doi.bucket,
+                object = %doi.delete_object.object_name,
+                "MRF queue full — delete replication failure entry dropped and will not be retried"
+            );
             ReplicationQueueAdmission::Missed
         };
 
@@ -749,7 +774,14 @@ impl<S: StorageAPI + NamespaceLocking> ReplicationPool<S> {
 
     /// Queues an MRF save operation
     async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
-        let _ = self.mrf_save_tx.try_send(entry);
+        if self.mrf_save_tx.try_send(entry).is_err() {
+            warn!(
+                event = EVENT_REPLICATION_MRF_QUEUE_OVERFLOW,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                "MRF queue full — replication failure entry dropped and will not be retried"
+            );
+        }
     }
 
     /// Starts the MRF processor background task

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -2593,6 +2593,10 @@ pub async fn replicate_object<S: StorageAPI>(roi: ReplicateObjectInfo, storage:
         name: object.clone(),
         user_tags: roi.user_tags.clone(),
         ssec: roi.ssec,
+        op_type: roi.op_type,
+        // ExistingObject ops must respect per-rule ExistingObjectReplicationStatus.
+        // Heal ops intentionally bypass it (repairing a past failure is not an initial sync).
+        existing_object: roi.op_type == ReplicationType::ExistingObject,
         ..Default::default()
     });