From 7eb9ad0f8c42348d660ce92a1cdf3902241c917d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E9=A9=AC=E7=99=BB=E5=B1=B1?= Date: Sun, 2 Aug 2026 19:05:55 +0800 Subject: [PATCH] feat(replication): batch DeleteObjects queue admission --- crates/ecstore/src/api/mod.rs | 25 ++--- crates/ecstore/src/bucket/replication/mod.rs | 4 +- .../replication/replication_object_bridge.rs | 40 ++++++- .../bucket/replication/replication_pool.rs | 100 +++++++++++++++++- .../replication/replication_queue_boundary.rs | 4 +- crates/replication/src/lib.rs | 6 +- crates/replication/src/queue.rs | 48 ++++++++- rustfs/src/app/object_usecase.rs | 29 +++-- rustfs/src/app/storage_api.rs | 13 +++ 9 files changed, 236 insertions(+), 33 deletions(-) diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index ae100d831..342c10b56 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, - ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, - ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, - ReplicationScannerBridge, ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, - ReplicationTargetValidationError, ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, - TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config, - delete_replication_version_id, get_global_replication_pool, get_global_replication_stats, - init_background_replication, invalid_replication_config_status_field, read_durable_mrf_backlog, - replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns, - resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication, - should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, - unsupported_replication_config_field, validate_replication_config_target_arns, version_purge_status_to_filemeta, + REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationBatchAdmission, ReplicationConfig, + ReplicationConfigurationExt, ReplicationDeleteScheduleInput, ReplicationDeleteStateSource, + ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO, ReplicationOperation, ReplicationPoolTrait, + ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge, ReplicationState, ReplicationStats, + ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError, ReplicationType, ResyncOpts, + ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus, VersionPurgeStatusType, + delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, + get_global_replication_stats, init_background_replication, invalid_replication_config_status_field, + read_durable_mrf_backlog, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, + replication_target_arns, resync_start_conflict_id, should_remove_replication_target, + should_schedule_delete_replication, should_use_existing_delete_replication_info, + should_use_existing_delete_replication_source, unsupported_replication_config_field, + validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 741f0a258..f47ace016 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -74,8 +74,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, - ReplicationQueueAdmission, + DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation, + ReplicationPriority, ReplicationQueueAdmission, }; pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; pub use replication_scanner_bridge::ReplicationScannerBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 1e677772d..0f00a1658 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/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, 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::>(); + Self::schedule_deletes(&delete_objects).await; + } } #[cfg(test)] diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index d54a168ad..ae228e2aa 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/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, - ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, WORKER_MAX_LIMIT, - initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count, replication_backpressure_recommendation, - replication_heal_queue_action, resized_worker_counts, should_queue_large_object, worker_queue_for_replication_type, + ReplicationBatchAdmission, ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, + ReplicationOperation, ReplicationPoolOpts, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, + WORKER_MAX_LIMIT, initial_worker_counts, large_worker_backpressure_resize, mrf_worker_size_to_count, + 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 ReplicationPool { 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; async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>; @@ -1748,6 +1789,10 @@ impl ReplicationPoolTrait for ReplicationPool { 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; } @@ -3028,6 +3073,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::>(); + 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); diff --git a/crates/ecstore/src/bucket/replication/replication_queue_boundary.rs b/crates/ecstore/src/bucket/replication/replication_queue_boundary.rs index 409680ab2..c0bff7741 100644 --- a/crates/ecstore/src/bucket/replication/replication_queue_boundary.rs +++ b/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, - ReplicationQueueAdmission, + DeletedObjectReplicationInfo, ReplicationBatchAdmission, ReplicationHealQueueResult, ReplicationOperation, + ReplicationPriority, ReplicationQueueAdmission, }; pub(crate) use rustfs_replication::{ LARGE_WORKER_COUNT, ReplicationBackpressureRecommendation, ReplicationBackpressureState, ReplicationHealQueueAction, diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index fda3da1fd..871059692 100644 --- a/crates/replication/src/lib.rs +++ b/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, - ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, mrf_save_admission, replication_heal_queue_action, - worker_queue_for_replication_type, + ReplicationBatchAdmission, ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, + ReplicationOperation, ReplicationPriority, ReplicationQueueAdmission, ReplicationWorkerQueue, mrf_save_admission, + replication_heal_queue_action, worker_queue_for_replication_type, }; pub use resync::{ BucketReplicationResyncStatus, Error, Result, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, diff --git a/crates/replication/src/queue.rs b/crates/replication/src/queue.rs index 1d804a185..569c0200c 100644 --- a/crates/replication/src/queue.rs +++ b/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); diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 265fe3afc..ba9c137b2 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/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, @@ -7031,14 +7032,26 @@ impl DefaultObjectUsecase { ..Default::default() }; - for result in &delete_results { - if let Some(dobj) = &result.delete_object - && replicate_deletes - && deleted_object_has_pending_replication_delete(dobj) - { + let replication_deletes = if replicate_deletes { + delete_results + .iter() + .filter_map(|result| result.delete_object.as_ref()) + .filter(|dobj| deleted_object_has_pending_replication_delete(dobj)) + .cloned() + .collect::>() + } 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(); diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c6df2bd75..cbdb141e4 100644 --- a/rustfs/src/app/storage_api.rs +++ b/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, + 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,