From 22460243bf08eef3f8392453f4f6323b4cd4f0d2 Mon Sep 17 00:00:00 2001 From: abdullahnah92 <157595835+abdullahnah92@users.noreply.github.com> Date: Sun, 14 Jun 2026 02:14:11 +0300 Subject: [PATCH] =?UTF-8?q?fix(bucket-repl):=20honor=20op=5Ftype=20in=20re?= =?UTF-8?q?plicate=5Fobject=20so=20ExistingObject=E2=80=A6=20(#3420)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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 Co-authored-by: houseme --- .../ecstore/src/bucket/replication/config.rs | 73 +++++++++++++++++++ .../bucket/replication/replication_pool.rs | 36 ++++++++- .../replication/replication_resyncer.rs | 4 + 3 files changed, 111 insertions(+), 2 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/config.rs b/crates/ecstore/src/bucket/replication/config.rs index 3eea72b16..335bc5aa4 100644 --- a/crates/ecstore/src/bucket/replication/config.rs +++ b/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())); + } } diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 241c13972..50c8dcdec 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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 ReplicationPool { 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 ReplicationPool { 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 ReplicationPool { 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 ReplicationPool { /// 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 diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 54b3d7e53..0dd5d5d09 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2593,6 +2593,10 @@ pub async fn replicate_object(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() });