// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use crate::bucket::bucket_target_sys::BucketTargetSys; use crate::bucket::metadata_sys; use crate::bucket::replication::ResyncOpts; use crate::bucket::replication::ResyncStatusType; use crate::bucket::replication::replicate_delete; use crate::bucket::replication::replicate_object; use crate::bucket::replication::replication_resyncer::{ BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MRF_REPLICATION_FILE, REPLICATION_DIR, RESYNC_FILE_NAME, ReplicationConfig, ReplicationResyncer, ReplicationStorage, TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, save_resync_status, }; use crate::bucket::replication::replication_state::ReplicationStats; use crate::config::com::{read_config, save_config}; use crate::disk::BUCKET_META_PREFIX; use crate::error::Error as EcstoreError; use crate::object_api::{ObjectInfo, ObjectOptions}; use crate::storage_api_contracts::EcstoreObjectIO; use lazy_static::lazy_static; use rustfs_filemeta::MrfOpKind; use rustfs_filemeta::MrfReplicateEntry; use rustfs_filemeta::ReplicateDecision; use rustfs_filemeta::ReplicateObjectInfo; use rustfs_filemeta::ReplicatedTargetInfo; use rustfs_filemeta::ReplicationStatusType; use rustfs_filemeta::ReplicationType; use rustfs_filemeta::ReplicationWorkerOperation; use rustfs_filemeta::ResyncDecision; use rustfs_filemeta::VersionPurgeStatusType; use rustfs_filemeta::replication_statuses_map; use rustfs_filemeta::version_purge_statuses_map; use rustfs_filemeta::{REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE}; use rustfs_storage_api::DeletedObject; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::any::Any; use std::sync::Arc; use std::sync::atomic::AtomicI32; use std::sync::atomic::Ordering; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::sync::Mutex; use tokio::sync::RwLock; use tokio::sync::mpsc; use tokio::sync::mpsc::Receiver; use tokio::sync::mpsc::Sender; use tokio::task::JoinHandle; use tokio::time::Duration; use tokio_util::sync::CancellationToken; use tracing::{debug, info, instrument, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; const EVENT_REPLICATION_WORKER_RESIZE_SKIPPED: &str = "replication_worker_resize_skipped"; 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_RESYNC_RECOVERED: &str = "replication_resync_recovered"; const EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED: &str = "replication_config_lookup_skipped"; const EVENT_REPLICATION_MRF_QUEUE_OVERFLOW: &str = "replication_mrf_queue_overflow"; fn should_auto_resume_resync(status: ResyncStatusType) -> bool { matches!(status, ResyncStatusType::ResyncPending | ResyncStatusType::ResyncStarted) } // Worker limits pub const WORKER_MAX_LIMIT: usize = 500; pub const WORKER_MIN_LIMIT: usize = 50; pub const WORKER_AUTO_DEFAULT: usize = 100; pub const MRF_WORKER_MAX_LIMIT: usize = 8; pub const MRF_WORKER_MIN_LIMIT: usize = 2; pub const MRF_WORKER_AUTO_DEFAULT: usize = 4; pub const LARGE_WORKER_COUNT: usize = 10; pub const MIN_LARGE_OBJ_SIZE: i64 = 128 * 1024 * 1024; // 128MiB #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub enum ReplicationQueueAdmission { #[default] Skipped, Queued, Missed, } impl ReplicationQueueAdmission { fn merge(&mut self, other: Self) { *self = match (*self, other) { (Self::Missed, _) | (_, Self::Missed) => Self::Missed, (Self::Queued, _) | (_, Self::Queued) => Self::Queued, (Self::Skipped, Self::Skipped) => Self::Skipped, }; } } #[derive(Debug, Clone, Default)] pub struct ReplicationHealQueueResult { pub object_info: ReplicateObjectInfo, pub admission: ReplicationQueueAdmission, } /// Priority levels for replication #[derive(Debug, Clone, PartialEq)] pub enum ReplicationPriority { Fast, Slow, Auto, } impl std::str::FromStr for ReplicationPriority { type Err = (); fn from_str(s: &str) -> Result { match s { "fast" => Ok(ReplicationPriority::Fast), "slow" => Ok(ReplicationPriority::Slow), "auto" => Ok(ReplicationPriority::Auto), _ => Ok(ReplicationPriority::Auto), // Default to Auto for unknown values } } } impl ReplicationPriority { pub fn as_str(&self) -> &'static str { match self { ReplicationPriority::Fast => "fast", ReplicationPriority::Slow => "slow", ReplicationPriority::Auto => "auto", } } } /// Enum for different types of replication operations #[derive(Debug)] pub enum ReplicationOperation { Object(Box), Delete(Box), } impl ReplicationWorkerOperation for ReplicationOperation { fn as_any(&self) -> &dyn Any { self } fn to_mrf_entry(&self) -> MrfReplicateEntry { match self { ReplicationOperation::Object(obj) => obj.to_mrf_entry(), ReplicationOperation::Delete(del) => del.to_mrf_entry(), } } fn get_bucket(&self) -> &str { match self { ReplicationOperation::Object(obj) => obj.get_bucket(), ReplicationOperation::Delete(del) => del.get_bucket(), } } fn get_object(&self) -> &str { match self { ReplicationOperation::Object(obj) => obj.get_object(), ReplicationOperation::Delete(del) => del.get_object(), } } fn get_size(&self) -> i64 { match self { ReplicationOperation::Object(obj) => obj.get_size(), ReplicationOperation::Delete(del) => del.get_size(), } } fn is_delete_marker(&self) -> bool { match self { ReplicationOperation::Object(obj) => obj.is_delete_marker(), ReplicationOperation::Delete(del) => del.is_delete_marker(), } } fn get_op_type(&self) -> ReplicationType { match self { ReplicationOperation::Object(obj) => obj.get_op_type(), ReplicationOperation::Delete(del) => del.get_op_type(), } } } /// Replication pool options #[derive(Debug, Clone)] pub struct ReplicationPoolOpts { pub priority: ReplicationPriority, pub max_workers: Option, pub max_l_workers: Option, } impl Default for ReplicationPoolOpts { fn default() -> Self { Self { priority: ReplicationPriority::Auto, max_workers: None, max_l_workers: None, } } } /// Main replication pool structure #[derive(Debug)] pub struct ReplicationPool { // Atomic counters for active workers active_workers: Arc, active_lrg_workers: Arc, active_mrf_workers: Arc, storage: Arc, // Configuration priority: RwLock, max_workers: RwLock, max_l_workers: RwLock, // Statistics stats: Arc, // Worker channels workers: RwLock>>, lrg_workers: RwLock>>, // MRF (Most Recent Failures) channels mrf_replica_tx: Sender, // Shared among N MRF workers; Arc allows spawning more than one worker. mrf_replica_rx: Arc>>, mrf_save_tx: Sender, mrf_save_rx: Mutex>>, // Control channels mrf_worker_kill_tx: Sender<()>, mrf_stop_tx: Sender<()>, // Worker size tracking mrf_worker_size: AtomicI32, // Task handles for cleanup task_handles: Mutex>>, // Replication resyncer for handling bucket resync operations resyncer: Arc, } impl ReplicationPool { /// Creates a new replication pool with specified options pub async fn new(opts: ReplicationPoolOpts, stats: Arc, storage: Arc) -> Arc { let max_workers = opts.max_workers.unwrap_or(WORKER_MAX_LIMIT); let (workers, failed_workers) = match opts.priority { ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT), ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT), ReplicationPriority::Auto => (WORKER_AUTO_DEFAULT, MRF_WORKER_AUTO_DEFAULT), }; let workers = std::cmp::min(workers, max_workers); let failed_workers = std::cmp::min(failed_workers, max_workers); let max_l_workers = opts.max_l_workers.unwrap_or(LARGE_WORKER_COUNT); // Create MRF channels let (mrf_replica_tx, mrf_replica_rx) = mpsc::channel(100000); let (mrf_save_tx, mrf_save_rx) = mpsc::channel(100000); let (mrf_worker_kill_tx, _mrf_worker_kill_rx) = mpsc::channel(failed_workers); let (mrf_stop_tx, _mrf_stop_rx) = mpsc::channel(1); let pool = Arc::new(Self { active_workers: Arc::new(AtomicI32::new(0)), active_lrg_workers: Arc::new(AtomicI32::new(0)), active_mrf_workers: Arc::new(AtomicI32::new(0)), priority: RwLock::new(opts.priority), max_workers: RwLock::new(max_workers), max_l_workers: RwLock::new(max_l_workers), stats, storage, workers: RwLock::new(Vec::new()), lrg_workers: RwLock::new(Vec::new()), mrf_replica_tx, mrf_replica_rx: Arc::new(Mutex::new(mrf_replica_rx)), mrf_save_tx, mrf_save_rx: Mutex::new(Some(mrf_save_rx)), mrf_worker_kill_tx, mrf_stop_tx, mrf_worker_size: AtomicI32::new(0), task_handles: Mutex::new(Vec::new()), resyncer: Arc::new(ReplicationResyncer::new().await), }); // Initialize workers pool.resize_lrg_workers(max_l_workers, 0).await; pool.resize_workers(workers, 0).await; pool.resize_failed_workers(failed_workers as i32).await; // Start background tasks pool.start_mrf_processor().await; pool.start_mrf_persister().await; pool } /// Returns the number of active workers handling replication traffic pub fn active_workers(&self) -> i32 { self.active_workers.load(Ordering::SeqCst) } /// Returns the number of active workers handling replication failures pub fn active_mrf_workers(&self) -> i32 { self.active_mrf_workers.load(Ordering::SeqCst) } /// Returns the number of active workers handling traffic > 128MiB object size pub fn active_lrg_workers(&self) -> i32 { self.active_lrg_workers.load(Ordering::SeqCst) } /// Resizes the large workers pool pub async fn resize_lrg_workers(&self, n: usize, check_old: usize) { let mut lrg_workers = self.lrg_workers.write().await; if (check_old > 0 && lrg_workers.len() != check_old) || n == lrg_workers.len() || n < 1 { return; } // Add workers if needed while lrg_workers.len() < n { let (tx, rx) = mpsc::channel(100000); lrg_workers.push(tx); let active_counter = self.active_lrg_workers.clone(); let storage = self.storage.clone(); let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { replicate_object(*obj_info, storage.clone()).await; } ReplicationOperation::Delete(del_info) => { replicate_delete(*del_info, storage.clone()).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } }); self.task_handles.lock().await.push(handle); } // Remove workers if needed while lrg_workers.len() > n { if let Some(worker) = lrg_workers.pop() { drop(worker); // Closing the channel will terminate the worker } } } /// Resizes the regular workers pool pub async fn resize_workers(&self, n: usize, check_old: usize) { let mut workers = self.workers.write().await; if (check_old > 0 && workers.len() != check_old) || n == workers.len() || n < 1 { debug!( event = EVENT_REPLICATION_WORKER_RESIZE_SKIPPED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, check_old_mismatch = check_old > 0 && workers.len() != check_old, same_size = n == workers.len(), invalid_target_size = n < 1, current_workers = workers.len(), target_workers = n, "Skipped replication worker resize" ); return; } // Add workers if needed if workers.len() < n { info!( event = EVENT_REPLICATION_WORKER_RESIZED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, action = "increase", from_workers = workers.len(), to_workers = n, "Resized replication workers" ); } while workers.len() < n { let (tx, rx) = mpsc::channel(10000); workers.push(tx); let active_counter = self.active_workers.clone(); let stats = self.stats.clone(); let storage = self.storage.clone(); let handle = tokio::spawn(async move { let mut rx = rx; while let Some(operation) = rx.recv().await { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { stats .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; // Perform actual replication (placeholder) replicate_object(obj_info.as_ref().clone(), storage.clone()).await; stats .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; } ReplicationOperation::Delete(del_info) => { stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await; // Perform actual delete replication (placeholder) replicate_delete(del_info.as_ref().clone(), storage.clone()).await; stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } }); self.task_handles.lock().await.push(handle); } // Remove workers if needed if workers.len() > n { info!( event = EVENT_REPLICATION_WORKER_RESIZED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, action = "decrease", from_workers = workers.len(), to_workers = n, "Resized replication workers" ); } while workers.len() > n { if let Some(worker) = workers.pop() { drop(worker); // Closing the channel will terminate the worker } } } /// Resizes the failed workers pool pub async fn resize_failed_workers(&self, n: i32) { // Spawn workers up to n. Each worker shares the receiver via Arc>. // The mutex is held only while calling recv() — released before processing — so // all workers process entries concurrently (the dequeue step is serialised but // the replication I/O is not). while self.mrf_worker_size.load(Ordering::SeqCst) < n { self.mrf_worker_size.fetch_add(1, Ordering::SeqCst); let active_counter = self.active_mrf_workers.clone(); let stats = self.stats.clone(); let storage = self.storage.clone(); let mrf_rx = Arc::clone(&self.mrf_replica_rx); let handle = tokio::spawn(async move { loop { let operation = { mrf_rx.lock().await.recv().await }; let Some(operation) = operation else { break }; active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { stats .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; replicate_object(obj_info.as_ref().clone(), storage.clone()).await; stats .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; } ReplicationOperation::Delete(del_info) => { replicate_delete(*del_info, storage.clone()).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } }); self.task_handles.lock().await.push(handle); } // Remove workers if needed while self.mrf_worker_size.load(Ordering::SeqCst) > n { self.mrf_worker_size.fetch_sub(1, Ordering::SeqCst); let _ = self.mrf_worker_kill_tx.try_send(()); } } /// Resizes worker priority and counts pub async fn resize_worker_priority( &self, pri: ReplicationPriority, max_workers: Option, max_l_workers: Option, ) { let (workers, mrf_workers) = match pri { ReplicationPriority::Fast => (WORKER_MAX_LIMIT, MRF_WORKER_MAX_LIMIT), ReplicationPriority::Slow => (WORKER_MIN_LIMIT, MRF_WORKER_MIN_LIMIT), ReplicationPriority::Auto => { let mut workers = WORKER_AUTO_DEFAULT; let mut mrf_workers = MRF_WORKER_AUTO_DEFAULT; let current_workers = self.workers.read().await.len(); if current_workers < WORKER_AUTO_DEFAULT { workers = std::cmp::min(current_workers + 1, WORKER_AUTO_DEFAULT); } let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst) as usize; if current_mrf < MRF_WORKER_AUTO_DEFAULT { mrf_workers = std::cmp::min(current_mrf + 1, MRF_WORKER_AUTO_DEFAULT); } (workers, mrf_workers) } }; let (final_workers, final_mrf_workers) = if let Some(max_w) = max_workers { *self.max_workers.write().await = max_w; (std::cmp::min(workers, max_w), std::cmp::min(mrf_workers, max_w)) } else { (workers, mrf_workers) }; let max_l_workers_val = max_l_workers.unwrap_or(LARGE_WORKER_COUNT); *self.max_l_workers.write().await = max_l_workers_val; *self.priority.write().await = pri; self.resize_workers(final_workers, 0).await; self.resize_failed_workers(final_mrf_workers as i32).await; self.resize_lrg_workers(max_l_workers_val, 0).await; } /// Gets a worker channel deterministically based on bucket and object names async fn get_worker_ch(&self, bucket: &str, object: &str, _size: i64) -> Option> { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; let mut hasher = DefaultHasher::new(); format!("{bucket}{object}").hash(&mut hasher); let hash = hasher.finish(); let workers = self.workers.read().await; if workers.is_empty() { return None; } let index = (hash as usize) % workers.len(); workers.get(index).cloned() } /// Queues a replica task pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission { // If object is large, queue it to a static set of large workers if ri.size >= MIN_LARGE_OBJ_SIZE { use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; let mut hasher = DefaultHasher::new(); format!("{}{}", ri.bucket, ri.name).hash(&mut hasher); let hash = hasher.finish(); let lrg_workers = self.lrg_workers.read().await; if !lrg_workers.is_empty() { let index = (hash as usize) % lrg_workers.len(); if let Some(worker) = lrg_workers.get(index) && worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_err() { // Queue to MRF if worker is busy 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 }; // Try to add more workers if possible let max_l_workers = *self.max_l_workers.read().await; let existing = lrg_workers.len(); if self.active_lrg_workers() < std::cmp::min(max_l_workers, LARGE_WORKER_COUNT) as i32 { let workers = std::cmp::min(existing + 1, max_l_workers); drop(lrg_workers); self.resize_lrg_workers(workers, existing).await; } return admission; } return ReplicationQueueAdmission::Queued; } return ReplicationQueueAdmission::Missed; } // Handle regular sized objects let ch = match ri.op_type { ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()), _ => self.get_worker_ch(&ri.bucket, &ri.name, ri.size).await, }; let Some(channel) = ch else { return ReplicationQueueAdmission::Missed; }; if channel.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() { return ReplicationQueueAdmission::Queued; } // Queue to MRF if all workers are busy 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 }; // Try to scale up workers based on priority let priority = self.priority.read().await.clone(); let max_workers = *self.max_workers.read().await; match priority { ReplicationPriority::Fast => { debug!( event = EVENT_REPLICATION_BACKPRESSURE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, queue_type = "object", priority = "fast", recommendation = "none", "Replication queue is backpressured" ); } ReplicationPriority::Slow => { debug!( event = EVENT_REPLICATION_BACKPRESSURE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, queue_type = "object", priority = "slow", recommendation = "set_priority_auto", "Replication queue is backpressured" ); } ReplicationPriority::Auto => { let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); let active_workers = self.active_workers(); if active_workers < max_w as i32 { let workers = self.workers.read().await; let new_count = std::cmp::min(workers.len() + 1, max_w); let existing = workers.len(); drop(workers); self.resize_workers(new_count, existing).await; } let max_mrf_workers = std::cmp::min(max_workers, MRF_WORKER_MAX_LIMIT); let active_mrf = self.active_mrf_workers(); if active_mrf < max_mrf_workers as i32 { let current_mrf = self.mrf_worker_size.load(Ordering::SeqCst); let new_mrf = std::cmp::min(current_mrf + 1, max_mrf_workers as i32); self.resize_failed_workers(new_mrf).await; } } } admission } /// Queues a replica delete task pub async fn queue_replica_delete_task(&self, doi: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission { let ch = match doi.op_type { ReplicationType::Heal | ReplicationType::ExistingObject => Some(self.mrf_replica_tx.clone()), _ => self.get_worker_ch(&doi.bucket, &doi.delete_object.object_name, 0).await, }; let Some(channel) = ch else { return ReplicationQueueAdmission::Missed; }; if channel.try_send(ReplicationOperation::Delete(Box::new(doi.clone()))).is_ok() { return ReplicationQueueAdmission::Queued; } 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 }; let priority = self.priority.read().await.clone(); let max_workers = *self.max_workers.read().await; match priority { ReplicationPriority::Fast => { debug!( event = EVENT_REPLICATION_BACKPRESSURE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, queue_type = "delete", priority = "fast", recommendation = "none", "Replication delete queue is backpressured" ); } ReplicationPriority::Slow => { debug!( event = EVENT_REPLICATION_BACKPRESSURE, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, queue_type = "delete", priority = "slow", recommendation = "set_priority_auto", "Replication delete queue is backpressured" ); } ReplicationPriority::Auto => { let max_w = std::cmp::min(max_workers, WORKER_MAX_LIMIT); if self.active_workers() < max_w as i32 { let workers = self.workers.read().await; let new_count = std::cmp::min(workers.len() + 1, max_w); let existing = workers.len(); drop(workers); self.resize_workers(new_count, existing).await; } } } admission } /// Queues an MRF save operation async fn queue_mrf_save(&self, entry: MrfReplicateEntry) { 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 — one-shot at startup. /// /// Reads the on-disk MRF file, re-injects every entry into `mrf_replica_tx` as a /// Heal operation, then clears the file. The file is cleared AFTER all entries are /// successfully queued so a crash mid-replay results in at-most-twice delivery /// (safe — replication is idempotent) rather than entry loss. async fn start_mrf_processor(&self) { let storage = self.storage.clone(); let handle = tokio::spawn(async move { let data = match read_config(storage.clone(), MRF_REPLICATION_FILE).await { Ok(d) => d, Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start Err(e) => { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, error = %e, "Failed to load MRF recovery file" ); return; } }; let entries = match decode_mrf_file(&data) { Ok(v) => v, Err(e) => { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, error = %e, "Failed to decode MRF recovery file — discarding corrupt data" ); // Overwrite the corrupt file so we don't fail again on next restart. let _ = save_config(storage, MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default()).await; return; } }; let total = entries.len(); let mut queued_count = 0usize; for entry in entries.iter() { match entry.op { MrfOpKind::Delete => { // Reconstruct a heal delete and re-queue it. We do NOT call // get_object_info here because the delete-marker or version may // already be absent from the local store — that is expected. let dv = DeletedObjectReplicationInfo { delete_object: DeletedObject { object_name: entry.object.clone(), version_id: entry.version_id, delete_marker_version_id: entry.delete_marker_version_id, delete_marker: entry.delete_marker, ..Default::default() }, bucket: entry.bucket.clone(), op_type: ReplicationType::Heal, event_type: REPLICATE_HEAL_DELETE.to_string(), ..Default::default() }; schedule_replication_delete(dv).await; queued_count += 1; } MrfOpKind::Object => { let opts = ObjectOptions { version_id: entry.version_id.map(|u| u.to_string()), ..Default::default() }; let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await { Ok(oi) => oi, Err(e) => { debug!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, bucket = %entry.bucket, object = %entry.object, error = %e, "MRF recovery: object not found, skipping" ); continue; } }; // Route through queue_replication_heal so the replication decision (dsc) // is computed from the live config — required for replicate_object. queue_replication_heal(&entry.bucket, oi, entry.retry_count as u32).await; queued_count += 1; } } } // Clear AFTER all entries are processed so a crash mid-replay causes at-most-twice // delivery (idempotent) rather than entry loss. if let Err(e) = save_config(storage, MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default()).await { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, error = %e, "Failed to clear MRF recovery file after replay — entries may be replayed again on next restart" ); } if queued_count > 0 { info!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, recovered = queued_count, total, "Recovered MRF entries from disk and queued for retry" ); } }); self.task_handles.lock().await.push(handle); } /// Starts the MRF persister — ongoing background task. /// /// Drains `mrf_save_rx` (entries that overflowed the normal worker channels) and /// writes them to the on-disk MRF file every 10 seconds or when 1 000 entries /// accumulate. The file is overwritten (not appended) on each flush so it always /// reflects the current pending backlog. async fn start_mrf_persister(&self) { let Some(mut rx) = self.mrf_save_rx.lock().await.take() else { return; }; let storage = self.storage.clone(); let handle = tokio::spawn(async move { let mut pending: Vec = Vec::new(); let mut interval = tokio::time::interval(Duration::from_secs(10)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { tokio::select! { entry = rx.recv() => match entry { Some(e) => { pending.push(e); if pending.len() >= 1000 && flush_mrf_to_disk(&pending, &storage).await { pending.clear(); } } None => { // Channel closed (pool shutting down) — final flush. if !pending.is_empty() { flush_mrf_to_disk(&pending, &storage).await; } break; } }, _ = interval.tick() => { if !pending.is_empty() && flush_mrf_to_disk(&pending, &storage).await { pending.clear(); } } } } }); self.task_handles.lock().await.push(handle); } /// Worker function for handling regular replication operations async fn add_worker( &self, mut rx: Receiver, active_counter: Arc, stats: Arc, ) { while let Some(operation) = rx.recv().await { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { stats .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; // Perform actual replication (placeholder) replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await; stats .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; } ReplicationOperation::Delete(del_info) => { stats.inc_q(&del_info.bucket, 0, true, del_info.op_type).await; // Perform actual delete replication (placeholder) replicate_delete(del_info.as_ref().clone(), self.storage.clone()).await; stats.dec_q(&del_info.bucket, 0, true, del_info.op_type).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } } /// Worker function for handling large object replication operations async fn add_large_worker(&self, mut rx: Receiver, active_counter: Arc, storage: Arc) { while let Some(operation) = rx.recv().await { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { replicate_object(*obj_info, storage.clone()).await; } ReplicationOperation::Delete(del_info) => { replicate_delete(*del_info, storage.clone()).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } } /// Worker function for handling MRF (Most Recent Failures) operations async fn add_mrf_worker( &self, mut rx: Receiver, active_counter: Arc, stats: Arc, ) { while let Some(operation) = rx.recv().await { active_counter.fetch_add(1, Ordering::SeqCst); match operation { ReplicationOperation::Object(obj_info) => { stats .inc_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; replicate_object(obj_info.as_ref().clone(), self.storage.clone()).await; stats .dec_q(&obj_info.bucket, obj_info.size, obj_info.delete_marker, obj_info.op_type) .await; } ReplicationOperation::Delete(del_info) => { replicate_delete(*del_info, self.storage.clone()).await; } } active_counter.fetch_sub(1, Ordering::SeqCst); } } /// Delete resync metadata from replication resync state in memory pub async fn delete_resync_metadata(&self, bucket: &str) { let mut status_map = self.resyncer.status_map.write().await; status_map.remove(bucket); // Note: global site resync metrics deletion would be handled here // global_site_resync_metrics.delete_bucket(bucket); } /// Initialize bucket replication resync for all buckets pub async fn init_resync_internal( self: Arc, cancellation_token: CancellationToken, buckets: Vec, ) -> Result<(), EcstoreError> { // Load bucket metadata system in background let pool_clone = self; tokio::spawn(async move { pool_clone.start_resync_routine(buckets, cancellation_token).await; }); Ok(()) } pub async fn get_bucket_resync_status(&self, bucket: &str) -> Result { if let Some(status) = self.resyncer.status_map.read().await.get(bucket).cloned() { return Ok(status); } let status = load_bucket_resync_metadata(bucket, self.storage.clone()).await?; self.resyncer .status_map .write() .await .insert(bucket.to_string(), status.clone()); Ok(status) } pub async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError> { self.resyncer.cancel(&opts).await; self.resyncer .mark_status(ResyncStatusType::ResyncCanceled, opts, self.storage.clone()) .await?; Ok(()) } pub async fn start_bucket_resync(self: Arc, opts: ResyncOpts) -> Result<(), EcstoreError> { let now = OffsetDateTime::now_utc(); let bucket_status = { let mut status_map = self.resyncer.status_map.write().await; let bucket_status = status_map.entry(opts.bucket.clone()).or_insert_with(|| { let mut status = BucketReplicationResyncStatus::new(); status.id = 0; status }); bucket_status.last_update = Some(now); bucket_status.targets_map.insert( opts.arn.clone(), TargetReplicationResyncStatus { start_time: Some(now), last_update: Some(now), resync_id: opts.resync_id.clone(), resync_before_date: opts.resync_before, resync_status: ResyncStatusType::ResyncPending, failed_size: 0, failed_count: 0, replicated_size: 0, replicated_count: 0, bucket: opts.bucket.clone(), object: String::new(), error: None, }, ); bucket_status.clone() }; save_resync_status(&opts.bucket, &bucket_status, self.storage.clone()).await?; let resyncer = self.resyncer.clone(); let storage = self.storage.clone(); let cancel_token = CancellationToken::new(); resyncer.register_cancel_token(&opts, cancel_token.clone()).await; tokio::spawn(async move { Box::pin(resyncer.clone().resync_bucket(cancel_token, storage, false, opts.clone())).await; resyncer.clear_cancel_token(&opts).await; }); Ok(()) } /// Start the resync routine that runs in a loop async fn start_resync_routine(self: Arc, buckets: Vec, cancellation_token: CancellationToken) { // Run the replication resync in a loop loop { let self_clone = self.clone(); let ctx = cancellation_token.clone(); tokio::select! { _ = cancellation_token.cancelled() => { return; } result = self_clone.load_resync(&buckets, ctx) => { if result.is_ok() { return; } } } // Generate random duration between 0 and 1 minute use rand::RngExt; let duration_millis = rand::rng().random_range(0..60_000); let mut duration = Duration::from_millis(duration_millis); // Make sure to sleep at least a second to avoid high CPU ticks if duration < Duration::from_secs(1) { duration = Duration::from_secs(1); } tokio::time::sleep(duration).await; } } /// Load bucket replication resync statuses into memory #[instrument(skip(_cancellation_token))] async fn load_resync( self: Arc, buckets: &[String], _cancellation_token: CancellationToken, ) -> Result<(), EcstoreError> { // TODO: add leader_lock // Make sure only one node running resync on the cluster // Note: Leader lock implementation would be needed here // let _lock_guard = global_leader_lock.get_lock().await?; let mut recovered_statuses = Vec::new(); let mut restart_opts = Vec::new(); let mut recovered_bucket_count = 0usize; let mut skipped_failed_target_count = 0usize; for bucket in buckets { let meta = match load_bucket_resync_metadata(bucket, self.storage.clone()).await { Ok(meta) => meta, Err(err) => { if !matches!(err, EcstoreError::VolumeNotFound) { debug!( event = EVENT_REPLICATION_RESYNC_LOAD_SKIPPED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, bucket, error = ?err, reason = "metadata_load_failed", "Skipped replication resync metadata load" ); } continue; } }; if meta.targets_map.is_empty() { continue; } recovered_bucket_count += 1; for (arn, stats) in &meta.targets_map { if should_auto_resume_resync(stats.resync_status) { restart_opts.push(ResyncOpts { bucket: bucket.clone(), arn: arn.clone(), resync_id: stats.resync_id.clone(), resync_before: stats.resync_before_date, }); } else if stats.resync_status == ResyncStatusType::ResyncFailed { skipped_failed_target_count += 1; } } recovered_statuses.push((bucket.clone(), meta)); } if !recovered_statuses.is_empty() { let mut status_map = self.resyncer.status_map.write().await; status_map.extend(recovered_statuses); } if !restart_opts.is_empty() || skipped_failed_target_count > 0 { info!( event = EVENT_REPLICATION_RESYNC_RECOVERED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, recovered_buckets = recovered_bucket_count, resumed_targets = restart_opts.len(), skipped_failed_targets = skipped_failed_target_count, "Recovered replication resync state from persisted metadata; failed targets require manual resync restart" ); } for opts in restart_opts { let ctx = CancellationToken::new(); let resync = self.resyncer.clone(); let storage = self.storage.clone(); tokio::spawn(async move { resync.register_cancel_token(&opts, ctx.clone()).await; Box::pin(resync.clone().resync_bucket(ctx, storage, true, opts.clone())).await; resync.clear_cancel_token(&opts).await; }); } Ok(()) } } /// Encodes `entries` and overwrites the MRF persistence file. /// Returns `true` on success; on failure logs the error and returns `false`. /// Callers must NOT clear their in-memory buffer on `false` so the next tick /// can retry — otherwise a transient storage error permanently drops the batch. async fn flush_mrf_to_disk(entries: &[MrfReplicateEntry], storage: &Arc) -> bool { match encode_mrf_file(entries) { Ok(data) => { if let Err(e) = save_config(storage.clone(), MRF_REPLICATION_FILE, data).await { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, count = entries.len(), error = %e, "Failed to flush MRF entries to disk" ); return false; } true } Err(e) => { warn!( component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, count = entries.len(), error = %e, "Failed to encode MRF entries for disk flush" ); false } } } /// Load bucket resync metadata from disk async fn load_bucket_resync_metadata( bucket: &str, obj_api: Arc, ) -> Result { let mut brs = BucketReplicationResyncStatus::new(); let resync_dir_path = format!("{BUCKET_META_PREFIX}/{bucket}/{REPLICATION_DIR}"); let resync_file_path = format!("{resync_dir_path}/{RESYNC_FILE_NAME}"); let data = match read_config(obj_api, &resync_file_path).await { Ok(data) => data, Err(EcstoreError::ConfigNotFound) => return Ok(brs), Err(err) => return Err(err), }; if data.is_empty() { // Seems to be empty return Ok(brs); } brs = decode_resync_file(&data)?; Ok(brs) } // Define a trait object type for the replication pool pub type DynReplicationPool = dyn ReplicationPoolTrait + Send + Sync; /// Trait that abstracts the replication pool operations #[async_trait::async_trait] pub trait ReplicationPoolTrait: std::fmt::Debug { fn active_workers(&self) -> i32; fn active_mrf_workers(&self) -> i32; 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 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>; async fn start_bucket_resync(self: Arc, opts: ResyncOpts) -> Result<(), EcstoreError>; async fn init_resync( self: Arc, cancellation_token: CancellationToken, buckets: Vec, ) -> Result<(), EcstoreError>; } // Implement the trait for ReplicationPool #[async_trait::async_trait] impl ReplicationPoolTrait for ReplicationPool { fn active_workers(&self) -> i32 { ReplicationPool::::active_workers(self) } fn active_mrf_workers(&self) -> i32 { ReplicationPool::::active_mrf_workers(self) } fn active_lrg_workers(&self) -> i32 { ReplicationPool::::active_lrg_workers(self) } async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission { self.queue_replica_task(ri).await } async fn queue_replica_delete_task(&self, ri: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission { self.queue_replica_delete_task(ri).await } async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize) { self.resize(priority, max_workers, max_l_workers).await; } async fn get_bucket_resync_status(&self, bucket: &str) -> Result { self.get_bucket_resync_status(bucket).await } async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError> { self.cancel_bucket_resync(opts).await } async fn start_bucket_resync(self: Arc, opts: ResyncOpts) -> Result<(), EcstoreError> { self.start_bucket_resync(opts).await } async fn init_resync( self: Arc, cancellation_token: CancellationToken, buckets: Vec, ) -> Result<(), EcstoreError> { self.init_resync_internal(cancellation_token, buckets).await } } lazy_static! { pub static ref GLOBAL_REPLICATION_POOL: tokio::sync::OnceCell> = tokio::sync::OnceCell::new(); pub static ref GLOBAL_REPLICATION_STATS: tokio::sync::OnceCell> = tokio::sync::OnceCell::new(); } /// Initializes background replication with the given options pub async fn init_background_replication(storage: Arc) { let stats = GLOBAL_REPLICATION_STATS .get_or_init(|| async { let stats = Arc::new(ReplicationStats::new()); stats.start_background_tasks().await; stats }) .await; let _pool = GLOBAL_REPLICATION_POOL .get_or_init(|| async { let pool = ReplicationPool::new(ReplicationPoolOpts::default(), stats.clone(), storage).await; pool as Arc }) .await; assert!(GLOBAL_REPLICATION_STATS.get().is_some()); assert!(GLOBAL_REPLICATION_POOL.get().is_some()); } pub fn get_global_replication_pool() -> Option> { GLOBAL_REPLICATION_POOL.get().cloned() } pub async fn schedule_replication( oi: ObjectInfo, o: Arc, dsc: ReplicateDecision, op_type: ReplicationType, ) { let tgt_statuses = replication_statuses_map(&oi.replication_status_internal.clone().unwrap_or_default()); let purge_statuses = version_purge_statuses_map(&oi.version_purge_status_internal.clone().unwrap_or_default()); let tm = get_str(&oi.user_defined, SUFFIX_REPLICATION_TIMESTAMP) .map(|v| OffsetDateTime::parse(&v, &Rfc3339).unwrap_or(OffsetDateTime::UNIX_EPOCH)); let mut rstate = oi.replication_state(); rstate.replicate_decision_str = dsc.to_string(); let asz = oi.get_actual_size().unwrap_or_default(); let mut ri = ReplicateObjectInfo { name: oi.name, size: oi.size, actual_size: asz, bucket: oi.bucket, version_id: oi.version_id, etag: oi.etag, mod_time: oi.mod_time, replication_status: oi.replication_status, replication_status_internal: oi.replication_status_internal, delete_marker: oi.delete_marker, version_purge_status_internal: oi.version_purge_status_internal, version_purge_status: oi.version_purge_status, replication_state: Some(rstate), op_type, dsc: dsc.clone(), target_statuses: tgt_statuses, target_purge_statuses: purge_statuses, replication_timestamp: tm, user_tags: (*oi.user_tags).clone(), checksum: None, retry_count: 0, event_type: "".to_string(), existing_obj_resync: ResyncDecision::default(), ssec: false, }; if ri.ssec { ri.checksum = oi.checksum } if dsc.is_synchronous() { replicate_object(ri, o).await } else if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { let _ = pool.queue_replica_task(ri).await; } } pub async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) { if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { let _ = pool.queue_replica_delete_task(dv.clone()).await; } if let (Some(rs), Some(stats)) = (dv.delete_object.replication_state, GLOBAL_REPLICATION_STATS.get()) { for (k, _v) in rs.targets.iter() { let ri = ReplicatedTargetInfo { arn: k.clone(), size: 0, duration: Duration::default(), op_type: ReplicationType::Delete, ..Default::default() }; stats .update(&dv.bucket, &ri, ReplicationStatusType::Pending, ReplicationStatusType::Empty) .await; } } } /// QueueReplicationHeal is a wrapper for queue_replication_heal_internal pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u32) { // ignore modtime zero objects if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) { return; } let rcfg = match metadata_sys::get_replication_config(bucket).await { Ok((config, _)) => config, Err(err) => { debug!( event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, bucket, error = %err, reason = "config_lookup_failed", "Skipped replication heal queue due to missing replication config" ); return; } }; let tgts = match BucketTargetSys::get().list_bucket_targets(bucket).await { Ok(targets) => Some(targets), Err(err) => { debug!( event = EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_REPLICATION, bucket, error = %err, reason = "target_list_failed", "Skipped bucket target list during replication heal queue setup" ); None } }; let rcfg_wrapper = ReplicationConfig::new(Some(rcfg), tgts); queue_replication_heal_internal(bucket, oi, rcfg_wrapper, retry_count).await; } /// queue_replication_heal_internal enqueues objects that failed replication OR eligible for resyncing through /// an ongoing resync operation or via existing objects replication configuration setting. pub async fn queue_replication_heal_internal( _bucket: &str, oi: ObjectInfo, rcfg: ReplicationConfig, retry_count: u32, ) -> ReplicationHealQueueResult { let mut roi = ReplicateObjectInfo::default(); // ignore modtime zero objects if oi.mod_time.is_none() || oi.mod_time == Some(OffsetDateTime::UNIX_EPOCH) { return ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, }; } if rcfg.config.is_none() || rcfg.remotes.is_none() { return ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, }; } roi = get_heal_replicate_object_info(&oi, &rcfg).await; roi.retry_count = retry_count; if !roi.dsc.replicate_any() { return ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, }; } // early return if replication already done, otherwise we need to determine if this // version is an existing object that needs healing. if roi.replication_status == ReplicationStatusType::Completed && roi.version_purge_status.is_empty() && !roi.existing_obj_resync.must_resync() { return ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, }; } if roi.delete_marker || !roi.version_purge_status.is_empty() { let (version_id, dm_version_id) = if roi.version_purge_status.is_empty() { (None, roi.version_id) } else { (roi.version_id, None) }; let dv = DeletedObjectReplicationInfo { delete_object: DeletedObject { object_name: roi.name.clone(), delete_marker_version_id: dm_version_id, version_id, replication_state: roi.replication_state.clone(), delete_marker_mtime: roi.mod_time, delete_marker: roi.delete_marker, ..Default::default() }, bucket: roi.bucket.clone(), op_type: ReplicationType::Heal, event_type: REPLICATE_HEAL_DELETE.to_string(), ..Default::default() }; // heal delete marker replication failure or versioned delete replication failure if roi.replication_status == ReplicationStatusType::Pending || roi.replication_status == ReplicationStatusType::Failed || roi.version_purge_status == VersionPurgeStatusType::Failed || roi.version_purge_status == VersionPurgeStatusType::Pending { let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { pool.queue_replica_delete_task(dv).await } else { ReplicationQueueAdmission::Missed }; return ReplicationHealQueueResult { object_info: roi, admission, }; } // if replication status is Complete on DeleteMarker and existing object resync required let existing_obj_resync = roi.existing_obj_resync.clone(); if existing_obj_resync.must_resync() && (roi.replication_status == ReplicationStatusType::Completed || roi.replication_status.is_empty()) { let admission = queue_replicate_deletes_wrapper(dv, existing_obj_resync).await; return ReplicationHealQueueResult { object_info: roi, admission, }; } return ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, }; } if roi.existing_obj_resync.must_resync() { roi.op_type = ReplicationType::ExistingObject; } match roi.replication_status { ReplicationStatusType::Pending | ReplicationStatusType::Failed => { roi.event_type = REPLICATE_HEAL.to_string(); let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { pool.queue_replica_task(roi.clone()).await } else { ReplicationQueueAdmission::Missed }; return ReplicationHealQueueResult { object_info: roi, admission, }; } _ => {} } if roi.existing_obj_resync.must_resync() { roi.event_type = REPLICATE_EXISTING.to_string(); let admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { pool.queue_replica_task(roi.clone()).await } else { ReplicationQueueAdmission::Missed }; return ReplicationHealQueueResult { object_info: roi, admission, }; } ReplicationHealQueueResult { object_info: roi, admission: ReplicationQueueAdmission::Skipped, } } /// Wrapper function for queueing replicate deletes with resync decision async fn queue_replicate_deletes_wrapper( doi: DeletedObjectReplicationInfo, existing_obj_resync: ResyncDecision, ) -> ReplicationQueueAdmission { let mut admission = ReplicationQueueAdmission::Skipped; for (k, v) in existing_obj_resync.targets.iter() { if v.replicate { let mut dv = doi.clone(); dv.reset_id = v.reset_id.clone(); dv.target_arn = k.clone(); let target_admission = if let Some(pool) = GLOBAL_REPLICATION_POOL.get() { pool.queue_replica_delete_task(dv).await } else { ReplicationQueueAdmission::Missed }; admission.merge(target_admission); } } admission } #[cfg(test)] mod tests { use super::*; use crate::bucket::replication::replication_resyncer::{decode_mrf_file, encode_mrf_file}; use uuid::Uuid; #[test] fn replication_queue_admission_combines_target_results() { let mut admission = ReplicationQueueAdmission::Skipped; admission.merge(ReplicationQueueAdmission::Queued); assert_eq!(admission, ReplicationQueueAdmission::Queued); admission.merge(ReplicationQueueAdmission::Missed); assert_eq!(admission, ReplicationQueueAdmission::Missed); } #[test] fn auto_resume_resync_only_for_inflight_states() { assert!(should_auto_resume_resync(ResyncStatusType::ResyncPending)); assert!(should_auto_resume_resync(ResyncStatusType::ResyncStarted)); assert!(!should_auto_resume_resync(ResyncStatusType::NoResync)); assert!(!should_auto_resume_resync(ResyncStatusType::ResyncCanceled)); assert!(!should_auto_resume_resync(ResyncStatusType::ResyncCompleted)); assert!(!should_auto_resume_resync(ResyncStatusType::ResyncFailed)); } // ── MrfReplicateEntry encode/decode roundtrips ──────────────────────────── #[test] fn mrf_entry_object_roundtrip() { let vid = Uuid::new_v4(); let entry = MrfReplicateEntry { bucket: "my-bucket".to_string(), object: "path/to/obj".to_string(), version_id: Some(vid), retry_count: 3, size: 1024, op: MrfOpKind::Object, delete_marker_version_id: None, delete_marker: false, }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); let decoded = decode_mrf_file(&encoded).expect("decode"); assert_eq!(decoded.len(), 1); let got = &decoded[0]; assert_eq!(got.bucket, "my-bucket"); assert_eq!(got.object, "path/to/obj"); assert_eq!(got.version_id, Some(vid)); assert_eq!(got.retry_count, 3); assert_eq!(got.size, 1024); assert_eq!(got.op, MrfOpKind::Object); assert_eq!(got.delete_marker_version_id, None); assert!(!got.delete_marker); } #[test] fn mrf_entry_delete_marker_roundtrip() { let dm_vid = Uuid::new_v4(); let entry = MrfReplicateEntry { bucket: "del-bucket".to_string(), object: "key".to_string(), version_id: None, retry_count: 0, size: 0, op: MrfOpKind::Delete, delete_marker_version_id: Some(dm_vid), delete_marker: true, }; let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode"); let decoded = decode_mrf_file(&encoded).expect("decode"); assert_eq!(decoded.len(), 1); let got = &decoded[0]; assert_eq!(got.bucket, "del-bucket"); assert_eq!(got.object, "key"); assert_eq!(got.version_id, None); assert_eq!(got.op, MrfOpKind::Delete); assert_eq!(got.delete_marker_version_id, Some(dm_vid)); assert!(got.delete_marker); } #[test] fn mrf_entry_versioned_delete_roundtrip() { let vid = Uuid::new_v4(); let entry = MrfReplicateEntry { bucket: "ver-bucket".to_string(), object: "versioned-key".to_string(), version_id: Some(vid), retry_count: 0, size: 0, op: MrfOpKind::Delete, delete_marker_version_id: None, delete_marker: false, }; let encoded = encode_mrf_file(&[entry]).expect("encode"); let decoded = decode_mrf_file(&encoded).expect("decode"); assert_eq!(decoded.len(), 1); let got = &decoded[0]; assert_eq!(got.op, MrfOpKind::Delete); assert_eq!(got.version_id, Some(vid)); assert_eq!(got.delete_marker_version_id, None); assert!(!got.delete_marker); } #[test] fn mrf_entry_mixed_batch_roundtrip() { let obj_vid = Uuid::new_v4(); let del_dm_vid = Uuid::new_v4(); let entries = vec![ MrfReplicateEntry { bucket: "b".to_string(), object: "obj".to_string(), version_id: Some(obj_vid), retry_count: 1, size: 512, op: MrfOpKind::Object, delete_marker_version_id: None, delete_marker: false, }, MrfReplicateEntry { bucket: "b".to_string(), object: "del".to_string(), version_id: None, retry_count: 0, size: 0, op: MrfOpKind::Delete, delete_marker_version_id: Some(del_dm_vid), delete_marker: true, }, ]; let encoded = encode_mrf_file(&entries).expect("encode"); let decoded = decode_mrf_file(&encoded).expect("decode"); assert_eq!(decoded.len(), 2); assert_eq!(decoded[0].op, MrfOpKind::Object); assert_eq!(decoded[0].version_id, Some(obj_vid)); assert_eq!(decoded[1].op, MrfOpKind::Delete); assert_eq!(decoded[1].delete_marker_version_id, Some(del_dm_vid)); assert!(decoded[1].delete_marker); } // ── Recovery replay routing ─────────────────────────────────────────────── #[test] fn mrf_entry_op_routes_correctly() { // Object entries must have op=Object so the processor calls get_object_info + heal. let obj_entry = MrfReplicateEntry { bucket: "b".to_string(), object: "o".to_string(), version_id: None, retry_count: 0, size: 0, op: MrfOpKind::Object, delete_marker_version_id: None, delete_marker: false, }; assert_eq!(obj_entry.op, MrfOpKind::Object); // Delete entries must have op=Delete so the processor calls schedule_replication_delete. let del_entry = MrfReplicateEntry { bucket: "b".to_string(), object: "o".to_string(), version_id: None, retry_count: 0, size: 0, op: MrfOpKind::Delete, delete_marker_version_id: Some(Uuid::new_v4()), delete_marker: true, }; assert_eq!(del_entry.op, MrfOpKind::Delete); // Entries written by old code (before the op field existed) must deserialise as Object // so existing recovery behaviour is preserved. let legacy_entry = MrfReplicateEntry { bucket: "b".to_string(), object: "o".to_string(), version_id: None, retry_count: 0, size: 0, op: MrfOpKind::default(), delete_marker_version_id: None, delete_marker: false, }; assert_eq!(legacy_entry.op, MrfOpKind::Object, "legacy default must be Object"); } #[test] fn mrf_legacy_file_without_op_field_decoded_as_object() { // Hand-build the exact bytes a pre-MrfOpKind binary would have written to disk. // The old MrfReplicateEntry had only 4 persisted keys (versionID is omitted when // None due to skip_serializing_if): bucket, object, retryCount, size. // There is no "op", "deleteMarker", or "deleteMarkerVersionID" key. // // This proves that #[serde(default)] on the `op` field carries real weight: // if you remove that attribute, rmp_serde will return an error on this payload // and the test will fail. let mut msgpack = Vec::new(); // Outer: array of 1 (the Vec) rmp::encode::write_array_len(&mut msgpack, 1).unwrap(); // Inner: named map with the 4 original fields only — no "op", no "deleteMarker*" rmp::encode::write_map_len(&mut msgpack, 4).unwrap(); rmp::encode::write_str(&mut msgpack, "bucket").unwrap(); rmp::encode::write_str(&mut msgpack, "old-bucket").unwrap(); rmp::encode::write_str(&mut msgpack, "object").unwrap(); rmp::encode::write_str(&mut msgpack, "old-key").unwrap(); rmp::encode::write_str(&mut msgpack, "retryCount").unwrap(); rmp::encode::write_i32(&mut msgpack, 2).unwrap(); rmp::encode::write_str(&mut msgpack, "size").unwrap(); rmp::encode::write_i64(&mut msgpack, 100).unwrap(); // Prepend the MRF file header: format=1 (LE u16) || version=1 (LE u16) let mut data = Vec::with_capacity(4 + msgpack.len()); data.extend_from_slice(&1u16.to_le_bytes()); // MRF_META_FORMAT data.extend_from_slice(&1u16.to_le_bytes()); // MRF_META_VERSION data.extend_from_slice(&msgpack); let decoded = decode_mrf_file(&data).expect("legacy payload must decode without error"); assert_eq!(decoded.len(), 1); let entry = &decoded[0]; assert_eq!(entry.bucket, "old-bucket"); assert_eq!(entry.object, "old-key"); assert_eq!(entry.retry_count, 2); assert_eq!(entry.size, 100); assert_eq!(entry.version_id, None); // The "op" key was absent — #[serde(default)] must fill in MrfOpKind::Object. assert_eq!(entry.op, MrfOpKind::Object, "missing op key must default to Object"); assert!(!entry.delete_marker); assert_eq!(entry.delete_marker_version_id, None); } }