feat(replication): bound DeleteObjects queue admission (#5637)

feat(replication): batch DeleteObjects queue admission

crates/ecstore/src/api/mod.rs

@@ -180,18 +180,19 @@ pub mod bucket {
         pub use crate::bucket::replication::{
             BucketReplicationResyncStatus, BucketStats, DeleteReplicationConfigSnapshot, DeletedObjectReplicationInfo,
             DurableMrfBacklog, DynReplicationPool, MrfOpKind, MrfReplicateEntry, MustReplicateOptions, ObjectOpts,
-            REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationConfig, ReplicationConfigurationExt,
+            REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig,
-            ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge,
+            ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource,
-            ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission,
+            ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait,
-            ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage,
+            ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats,
-            ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog,
+            ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, ReplicationType, ResyncOpts,
-            TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config,
+            ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus, VersionPurgeStatusType,
-            delete_replication_version_id, get_global_replication_pool, get_global_replication_stats,
+            delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
-            init_background_replication, invalid_replication_config_status_field, read_durable_mrf_backlog,
+            get_global_replication_stats, init_background_replication, invalid_replication_config_status_field,
-            replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
+            read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map,
-            resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
+            replication_target_arns, resync_start_conflict_id, should_remove_replication_target,
-            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
+            should_schedule_delete_replication, should_use_existing_delete_replication_info,
-            unsupported_replication_config_field, validate_replication_config_target_arns, version_purge_status_to_filemeta,
+            should_use_existing_delete_replication_source, unsupported_replication_config_field,
+            validate_replication_config_target_arns, version_purge_status_to_filemeta,
         };
     }
 

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

@@ -72,8 +72,8 @@ pub use replication_pool::{
     init_background_replication, read_durable_mrf_backlog, resync_start_conflict_id,
 };
 pub use replication_queue_boundary::{
-    DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority,
+    DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation,
-    ReplicationQueueAdmission,
+    ReplicationPriority, ReplicationQueueAdmission,
 };
 pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 pub use replication_scanner_bridge::ReplicationScannerBridge;

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

@@ -15,7 +15,7 @@
 use std::{collections::HashMap, sync::Arc};
 
 use super::replication_error_boundary::Result;
-use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType};
+use super::replication_filemeta_boundary::{ReplicateDecision, ReplicatedTargetInfo, ReplicationStatusType, ReplicationType};
 use super::replication_metadata_boundary::ReplicationInstanceContext;
 use super::replication_object_config::{
     DeleteReplicationConfigSnapshot, check_replicate_delete, check_replicate_delete_strict, check_replicate_delete_with_snapshot,
@@ -112,6 +112,31 @@ impl ReplicationObjectBridge {
         schedule_replication_delete(delete_object).await;
     }
 
+    pub async fn schedule_deletes(delete_objects: &[DeletedObjectReplicationInfo]) {
+        if let Some(pool) = super::runtime_boundary::replication_pool() {
+            let _ = pool.queue_replica_delete_batch(delete_objects).await;
+        }
+
+        if let Some(stats) = super::runtime_boundary::replication_stats() {
+            for delete_object in delete_objects {
+                if let Some(rs) = &delete_object.delete_object.replication_state {
+                    for k in rs.targets.keys() {
+                        let ri = ReplicatedTargetInfo {
+                            arn: k.clone(),
+                            size: 0,
+                            duration: std::time::Duration::default(),
+                            op_type: ReplicationType::Delete,
+                            ..Default::default()
+                        };
+                        stats
+                            .update(&delete_object.bucket, &ri, ReplicationStatusType::Pending, ReplicationStatusType::Empty)
+                            .await;
+                    }
+                }
+            }
+        }
+    }
+
     pub async fn schedule_storage_delete(delete_object: DeletedObject, bucket: String, event_type: String) {
         Self::schedule_delete(DeletedObjectReplicationInfo {
             delete_object: deleted_object_for_replication(delete_object),
@@ -121,6 +146,19 @@ impl ReplicationObjectBridge {
         })
         .await;
     }
+
+    pub async fn schedule_storage_deletes(delete_objects: Vec<DeletedObject>, bucket: String, event_type: String) {
+        let delete_objects = delete_objects
+            .into_iter()
+            .map(|delete_object| DeletedObjectReplicationInfo {
+                delete_object: deleted_object_for_replication(delete_object),
+                bucket: bucket.clone(),
+                event_type: event_type.clone(),
+                ..Default::default()
+            })
+            .collect::<Vec<_>>();
+        Self::schedule_deletes(&delete_objects).await;
+    }
 }
 
 #[cfg(test)]

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

@@ -25,10 +25,11 @@ use super::replication_metadata_boundary::ReplicationMetadataStore;
 use super::replication_object_config::{ReplicationConfig, check_replicate_delete};
 use super::replication_queue_boundary::{
     DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationBackpressureRecommendation, ReplicationBackpressureState,
-    ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation,
+    ReplicationBatchAdmission, ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes,
-    ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, WORKER_MAX_LIMIT,
+    ReplicationOperation, ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue,
-    initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count, replication_backpressure_recommendation,
+    WORKER_MAX_LIMIT, initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count,
-    replication_heal_queue_action, resized_worker_counts, should_queue_large_object, worker_queue_for_replication_type,
+    replication_backpressure_recommendation, replication_heal_queue_action, resized_worker_counts, should_queue_large_object,
+    worker_queue_for_replication_type,
 };
 use super::replication_resync_boundary::ResyncStatusType;
 use super::replication_resync_boundary::{
@@ -45,6 +46,8 @@ use super::replication_storage_boundary::{
 use super::replication_target_boundary::{ReplicationTargetStore, replication_object_is_ssec_encrypted};
 use super::replication_versioning_boundary::ReplicationVersioningStore;
 use super::runtime_boundary as runtime_sources;
+use futures_util::stream::{self, StreamExt};
+use metrics::{counter, histogram};
 use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
 use std::collections::HashMap;
 use std::collections::hash_map::Entry;
@@ -72,6 +75,9 @@ const EVENT_REPLICATION_BACKPRESSURE: &str = "replication_backpressure";
 const EVENT_REPLICATION_RESYNC_LOAD_SKIPPED: &str = "replication_resync_load_skipped";
 const EVENT_REPLICATION_RESYNC_RECOVERED: &str = "replication_resync_recovered";
 const EVENT_REPLICATION_MRF_QUEUE_UNAVAILABLE: &str = "replication_mrf_queue_unavailable";
+const DELETE_BATCH_ADMISSION_CONCURRENCY: usize = 16;
+const METRIC_DELETE_BATCH_ITEMS_TOTAL: &str = "rustfs_replication_delete_batch_items_total";
+const METRIC_DELETE_BATCH_SIZE: &str = "rustfs_replication_delete_batch_size";
 
 #[derive(Debug, Default)]
 pub struct DurableMrfBacklog {
@@ -859,6 +865,40 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         admission
     }
 
+    /// Queues a DeleteObjects replication tail with a fixed concurrency window.
+    /// Each item retains the existing regular-worker to MRF fallback contract.
+    pub async fn queue_replica_delete_batch(&self, deletes: &[DeletedObjectReplicationInfo]) -> ReplicationBatchAdmission {
+        let mut summary = ReplicationBatchAdmission::default();
+        let mut admissions = stream::iter(
+            deletes
+                .iter()
+                .cloned()
+                .map(|delete| async move { self.queue_replica_delete_task(delete).await }),
+        )
+        .buffer_unordered(DELETE_BATCH_ADMISSION_CONCURRENCY);
+
+        while let Some(admission) = admissions.next().await {
+            summary.record(admission);
+        }
+
+        let outcome = summary.outcome();
+        let total = u64::try_from(summary.total).unwrap_or(u64::MAX);
+        let queued = u64::try_from(summary.queued).unwrap_or(u64::MAX);
+        let missed = u64::try_from(summary.missed).unwrap_or(u64::MAX);
+        histogram!(METRIC_DELETE_BATCH_SIZE).record(total as f64);
+        counter!(METRIC_DELETE_BATCH_ITEMS_TOTAL, "outcome" => outcome, "state" => "queued").increment(queued);
+        counter!(METRIC_DELETE_BATCH_ITEMS_TOTAL, "outcome" => outcome, "state" => "missed").increment(missed);
+        debug!(
+            event = "replication_delete_batch_admission",
+            batch_size = summary.total,
+            queued = summary.queued,
+            missed = summary.missed,
+            outcome,
+            "Admitted DeleteObjects replication batch"
+        );
+        summary
+    }
+
     /// Queues an MRF save operation
     async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
         let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
@@ -1712,6 +1752,7 @@ pub trait ReplicationPoolTrait: std::fmt::Debug {
     fn active_lrg_workers(&self) -> i32;
     async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission;
     async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission;
+    async fn queue_replica_delete_batch(&self, deletes: &[DeletedObjectReplicationInfo]) -> ReplicationBatchAdmission;
     async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize);
     async fn get_bucket_resync_status(&self, bucket: &str) -> Result<BucketReplicationResyncStatus, EcstoreError>;
     async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>;
@@ -1748,6 +1789,10 @@ impl<S: ReplicationStorage> ReplicationPoolTrait for ReplicationPool<S> {
         self.queue_replica_delete_task(ri).await
     }
 
+    async fn queue_replica_delete_batch(&self, deletes: &[DeletedObjectReplicationInfo]) -> ReplicationBatchAdmission {
+        self.queue_replica_delete_batch(deletes).await
+    }
+
     async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize) {
         self.resize(priority, max_workers, max_l_workers).await;
     }
@@ -3036,6 +3081,53 @@ mod tests {
         assert_eq!(received.object, "second");
     }
 
+    #[tokio::test]
+    async fn delete_batch_admission_reports_mrf_fallback_items() {
+        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
+        let (worker_tx, worker_rx) = mpsc::channel(1);
+        worker_tx
+            .try_send(ReplicationOperation::Delete(Box::new(DeletedObjectReplicationInfo {
+                bucket: "batch-backpressure".to_string(),
+                delete_object: ReplicationDeletedObject {
+                    object_name: "already-queued".to_string(),
+                    ..Default::default()
+                },
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            })))
+            .expect("test worker channel should be full");
+        pool.workers.write().await.push(worker_tx);
+
+        let mut mrf_rx = pool
+            .mrf_save_rx
+            .lock()
+            .await
+            .take()
+            .expect("test should own the MRF save receiver");
+        let deletes = (0..1)
+            .map(|index| DeletedObjectReplicationInfo {
+                bucket: "batch-backpressure".to_string(),
+                delete_object: ReplicationDeletedObject {
+                    object_name: format!("object-{index}"),
+                    ..Default::default()
+                },
+                op_type: ReplicationType::Delete,
+                ..Default::default()
+            })
+            .collect::<Vec<_>>();
+        let summary = pool.queue_replica_delete_batch(&deletes).await;
+        let entry = mrf_rx
+            .recv()
+            .await
+            .expect("MRF fallback entry should be queued after batch admission");
+        assert_eq!(entry.object, "object-0");
+        assert_eq!(summary.total, 1);
+        assert_eq!(summary.queued, 1);
+        assert_eq!(summary.missed, 0);
+        assert_eq!(summary.outcome(), "all_queued");
+        drop(worker_rx);
+    }
+
     #[tokio::test]
     async fn mrf_save_admission_records_missed_when_channel_is_closed() {
         let (tx, rx) = mpsc::channel(1);

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

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 pub use rustfs_replication::{
-    DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority,
+    DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation,
-    ReplicationQueueAdmission,
+    ReplicationPriority, ReplicationQueueAdmission,
 };
 pub(crate) use rustfs_replication::{
     LARGE_WORKER_COUNT, ReplicationBackpressureRecommendation, ReplicationBackpressureState, ReplicationHealQueueAction,

crates/replication/src/lib.rs

@@ -62,9 +62,9 @@ pub use operation::{
     should_use_existing_delete_replication_source,
 };
 pub use queue::{
-    ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation,
+    ReplicationBatchAdmission, ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes,
-    ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, mrf_save_admission, replication_heal_queue_action,
+    ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, mrf_save_admission,
-    worker_queue_for_replication_type,
+    replication_heal_queue_action, worker_queue_for_replication_type,
 };
 pub use resync::{
     BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus,

crates/replication/src/queue.rs

@@ -39,6 +39,32 @@ impl ReplicationQueueAdmission {
     }
 }
 
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+pub struct ReplicationBatchAdmission {
+    pub total: usize,
+    pub queued: usize,
+    pub missed: usize,
+}
+
+impl ReplicationBatchAdmission {
+    pub fn record(&mut self, admission: ReplicationQueueAdmission) {
+        self.total += 1;
+        match admission {
+            ReplicationQueueAdmission::Queued => self.queued += 1,
+            ReplicationQueueAdmission::Missed | ReplicationQueueAdmission::Skipped => self.missed += 1,
+        }
+    }
+
+    pub fn outcome(self) -> &'static str {
+        match (self.queued, self.missed) {
+            (0, 0) => "empty",
+            (_, 0) => "all_queued",
+            (0, _) => "all_missed",
+            _ => "partial",
+        }
+    }
+}
+
 pub fn mrf_save_admission(saved: bool) -> ReplicationQueueAdmission {
     if saved {
         ReplicationQueueAdmission::Queued
@@ -267,7 +293,8 @@ mod tests {
     use uuid::Uuid;
 
     use super::{
-        ReplicationHealQueueAction, ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue,
+        ReplicationBatchAdmission, ReplicationHealQueueAction, ReplicationOperation, ReplicationPriority,
+        ReplicationQueueAdmission, ReplicationWorkerQueue,
     };
     use crate::storage_api::DeletedObject;
     use crate::{
@@ -297,6 +324,25 @@ mod tests {
         assert_eq!(mrf_save_admission(false), ReplicationQueueAdmission::Missed);
     }
 
+    #[test]
+    fn replication_batch_admission_classifies_all_partial_and_empty_batches() {
+        let mut all = ReplicationBatchAdmission::default();
+        all.record(ReplicationQueueAdmission::Queued);
+        assert_eq!(all.outcome(), "all_queued");
+
+        let mut partial = ReplicationBatchAdmission::default();
+        partial.record(ReplicationQueueAdmission::Queued);
+        partial.record(ReplicationQueueAdmission::Missed);
+        assert_eq!(partial.outcome(), "partial");
+
+        let mut all_missed = ReplicationBatchAdmission::default();
+        all_missed.record(ReplicationQueueAdmission::Missed);
+        assert_eq!(all_missed.outcome(), "all_missed");
+
+        let empty = ReplicationBatchAdmission::default();
+        assert_eq!(empty.outcome(), "empty");
+    }
+
     #[test]
     fn worker_queue_routes_heal_and_existing_objects_to_mrf() {
         assert_eq!(worker_queue_for_replication_type(&ReplicationType::Heal), ReplicationWorkerQueue::Mrf);

rustfs/src/app/object_usecase.rs

@@ -45,7 +45,8 @@ use super::storage_api::object_usecase::bucket::{
         DeleteReplicationConfigSnapshot, REPLICATE_INCOMING_DELETE, ReplicationStatusType, delete_replication_state_from_config,
         delete_replication_version_id, deleted_object_has_pending_replication_delete, has_active_delete_rule,
         load_delete_config_snapshot, must_replicate_object, schedule_object_replication, schedule_replication_delete,
-        set_deleted_object_replication_state, should_schedule_delete_replication, should_use_existing_delete_replication_info,
+        schedule_replication_deletes, set_deleted_object_replication_state, should_schedule_delete_replication,
+        should_use_existing_delete_replication_info,
     },
     tagging::decode_tags,
     validate_restore_request,
@@ -7046,14 +7047,26 @@ impl DefaultObjectUsecase {
             ..Default::default()
         };
 
-        for result in &delete_results {
+        let replication_deletes = if replicate_deletes {
-            if let Some(dobj) = &result.delete_object
+            delete_results
-                && replicate_deletes
+                .iter()
-                && deleted_object_has_pending_replication_delete(dobj)
+                .filter_map(|result| result.delete_object.as_ref())
-            {
+                .filter(|dobj| deleted_object_has_pending_replication_delete(dobj))
+                .cloned()
+                .collect::<Vec<_>>()
+        } else {
+            Vec::new()
+        };
+        if !replication_deletes.is_empty() {
+            let bucket_for_replication = bucket.clone();
+            let replication_task = tokio::spawn(async move {
                 let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Replication);
-                schedule_replication_delete(dobj.clone(), bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
+                schedule_replication_deletes(replication_deletes, bucket_for_replication, REPLICATE_INCOMING_DELETE.to_string())
-            }
+                    .await;
+            });
+            // The spawned task owns every locally committed delete. Dropping the
+            // join handle on request cancellation therefore cannot lose the tail.
+            let _ = replication_task.await;
         }
 
         let req_headers = req.headers.clone();

rustfs/src/app/storage_api.rs

@@ -718,6 +718,19 @@ pub(crate) mod bucket {
             ReplicationObjectBridge::schedule_storage_delete(delete_object, bucket, event_type).await;
         }
 
+        pub(crate) async fn schedule_replication_deletes(
+            delete_objects: Vec<crate::storage::storage_api::StorageDeletedObject>,
+            bucket: String,
+            event_type: String,
+        ) {
+            #[cfg(test)]
+            SCHEDULED_REPLICATION_DELETES
+                .lock()
+                .expect("scheduled replication delete lock should not be poisoned")
+                .extend(delete_objects.iter().cloned());
+            ReplicationObjectBridge::schedule_storage_deletes(delete_objects, bucket, event_type).await;
+        }
+
         pub(crate) fn set_deleted_object_replication_state(
             delete_object: &mut crate::storage::storage_api::StorageDeletedObject,
             state: &replication_contracts::ReplicationState,