From 4d19b069c39b51b22975be416c8673d191a65b63 Mon Sep 17 00:00:00 2001 From: weisd Date: Wed, 4 Feb 2026 13:39:35 +0800 Subject: [PATCH] fix: replication delete (#1714) --- .../replication/replication_resyncer.rs | 63 +++++++++++++++- crates/ecstore/src/disk/local.rs | 7 +- crates/ecstore/src/set_disk.rs | 73 ++++++++++++++----- crates/filemeta/src/filemeta.rs | 2 +- crates/filemeta/src/replication.rs | 2 +- rustfs/src/storage/ecfs.rs | 2 +- 6 files changed, 123 insertions(+), 26 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index f3a68f6dc..4961f4351 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -28,6 +28,7 @@ use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_fo use crate::event::name::EventName; use crate::event_notification::{EventArgs, send_event}; use crate::global::GLOBAL_LocalNodeName; +use crate::set_disk::get_lock_acquire_timeout; use crate::store_api::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, WalkOptions}; use crate::{StorageAPI, new_object_layer_fn}; use aws_sdk_s3::error::SdkError; @@ -1273,7 +1274,58 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, } }; - //TODO: nslock + let ns_lock = match storage + .new_ns_lock(&bucket, format!("/[replicate]/{}", dobj.delete_object.object_name).as_str()) + .await + { + Ok(ns_lock) => ns_lock, + Err(e) => { + warn!( + "failed to get ns lock for bucket:{} object:{} error:{}", + bucket, dobj.delete_object.object_name, e + ); + send_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(), + bucket_name: bucket.clone(), + object: ObjectInfo { + bucket: bucket.clone(), + name: dobj.delete_object.object_name.clone(), + version_id, + delete_marker: dobj.delete_object.delete_marker, + ..Default::default() + }, + user_agent: "Internal: [Replication]".to_string(), + host: GLOBAL_LocalNodeName.to_string(), + ..Default::default() + }); + return; + } + }; + + let _lock_guard = match ns_lock.get_write_lock(get_lock_acquire_timeout()).await { + Ok(lock_guard) => lock_guard, + Err(e) => { + warn!( + "failed to get write lock for bucket:{} object:{} error:{}", + bucket, dobj.delete_object.object_name, e + ); + send_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.as_ref().to_string(), + bucket_name: bucket.clone(), + object: ObjectInfo { + bucket: bucket.clone(), + name: dobj.delete_object.object_name.clone(), + version_id, + delete_marker: dobj.delete_object.delete_marker, + ..Default::default() + }, + user_agent: "Internal: [Replication]".to_string(), + host: GLOBAL_LocalNodeName.to_string(), + ..Default::default() + }); + return; + } + }; // Initialize replicated infos let mut rinfos = ReplicatedInfos { @@ -1379,7 +1431,7 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, dobj.delete_object.version_id.map(|v| v.to_string()), ); if replication_status != prev_status { - drs.replica_timestamp = Some(OffsetDateTime::now_utc()); + drs.replication_timestamp = Some(OffsetDateTime::now_utc()); } let event_name = if replication_status == ReplicationStatusType::Completed { @@ -1521,6 +1573,13 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli } } + if rinfo.replication_status == ReplicationStatusType::Completed + && !tgt_client.reset_id.is_empty() + && dobj.op_type == ReplicationType::ExistingObject + { + rinfo.resync_timestamp = format!("{};{}", OffsetDateTime::now_utc().format(&Rfc3339).unwrap(), tgt_client.reset_id); + } + rinfo } diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index cce32993c..e0f3f899f 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -2421,11 +2421,12 @@ impl DiskAPI for LocalDisk { return self.write_metadata("", volume, path, fi).await; } - return if fi.version_id.is_some() { - Err(DiskError::FileVersionNotFound) + let ret_err = if fi.version_id.is_some() { + DiskError::FileVersionNotFound } else { - Err(DiskError::FileNotFound) + DiskError::FileNotFound }; + return Err(ret_err); } }; diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 76e570ba8..a35b35366 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -71,7 +71,7 @@ use rustfs_common::heal_channel::{DriveState, HealChannelPriority, HealItemType, use rustfs_config::MI_B; use rustfs_filemeta::{ FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo, - RawFileInfo, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions, + RawFileInfo, ReplicateDecision, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw, merge_file_meta_versions, }; use rustfs_lock::LockClient; use rustfs_lock::fast_lock::types::LockResult; @@ -122,7 +122,7 @@ const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750); /// Get lock acquire timeout from environment variable RUSTFS_LOCK_ACQUIRE_TIMEOUT (in seconds) /// Defaults to 30 seconds if not set or invalid -fn get_lock_acquire_timeout() -> Duration { +pub fn get_lock_acquire_timeout() -> Duration { Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5)) } @@ -2251,23 +2251,42 @@ impl SetDisks { Ok((fi, parts_metadata, op_online_disks)) } - async fn get_object_info_and_quorum(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<(ObjectInfo, usize)> { - let (fi, _, _) = self.get_object_fileinfo(bucket, object, opts, false).await?; + async fn get_object_info_and_quorum( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> (ObjectInfo, usize, Option) { + let fi = match self.get_object_fileinfo(bucket, object, opts, false).await { + Ok((fi, _, _)) => fi, + Err(e) => return (ObjectInfo::default(), 0, Some(e)), + }; let write_quorum = fi.write_quorum(self.default_write_quorum()); let oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); - // TODO: replicatio + + if !fi.version_purge_status().is_empty() && opts.version_id.is_some() { + return ( + oi, + write_quorum, + Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])), + ); + } if fi.deleted { return if opts.version_id.is_none() || opts.delete_marker { - Err(to_object_err(StorageError::FileNotFound, vec![bucket, object])) + (oi, write_quorum, Some(to_object_err(StorageError::FileNotFound, vec![bucket, object]))) } else { - Err(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])) + ( + oi, + write_quorum, + Some(to_object_err(StorageError::MethodNotAllowed, vec![bucket, object])), + ) }; } - Ok((oi, write_quorum)) + (oi, write_quorum, None) } #[allow(clippy::too_many_arguments)] @@ -4461,10 +4480,19 @@ impl StorageAPI for SetDisks { return Ok(ObjectInfo::default()); } - let (mut goi, write_quorum, gerr) = match self.get_object_info_and_quorum(bucket, object, &opts).await { - Ok((oi, wq)) => (oi, wq, None), - Err(e) => (ObjectInfo::default(), 0, Some(e)), - }; + // TODO: Lifecycle + + let mut version_found = true; + let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await; + if let Some(err) = &gerr + && goi.name.is_empty() + { + if opts.delete_marker { + version_found = false; + } else { + return Err(err.clone()); + } + } let otd = ObjectToDelete { object_name: object.to_string(), @@ -4475,9 +4503,16 @@ impl StorageAPI for SetDisks { ..Default::default() }; - let version_found = if opts.delete_marker { gerr.is_none() } else { true }; - - let dsc = check_replicate_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await; + let dsc = if opts + .delete_replication + .as_ref() + .map(|v| v.replica_status == ReplicationStatusType::Replica) + == Some(true) + { + ReplicateDecision::default() + } else { + check_replicate_delete(bucket, &otd, &goi, &opts, gerr.map(|e| e.to_string())).await + }; if dsc.replicate_any() { opts.set_delete_replication_state(dsc); @@ -4501,11 +4536,11 @@ impl StorageAPI for SetDisks { mark_delete = false; } - if opts.version_purge_status() != VersionPurgeStatusType::Complete { + if opts.version_purge_status() == VersionPurgeStatusType::Complete { mark_delete = false; } - if version_found && (goi.version_purge_status.is_empty() || !goi.delete_marker) { + if version_found && (!goi.version_purge_status.is_empty() || !goi.delete_marker) { delete_marker = false; } } @@ -4550,7 +4585,9 @@ impl StorageAPI for SetDisks { .await .map_err(|e| to_object_err(e, vec![bucket, object]))?; - return Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended)); + let mut oi = ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended); + oi.replication_decision = goi.replication_decision; + return Ok(oi); } let version_id = opts.version_id.as_ref().and_then(|v| Uuid::parse_str(v).ok()); diff --git a/crates/filemeta/src/filemeta.rs b/crates/filemeta/src/filemeta.rs index 796876377..cdfbeec4e 100644 --- a/crates/filemeta/src/filemeta.rs +++ b/crates/filemeta/src/filemeta.rs @@ -615,7 +615,7 @@ impl FileMeta { } // delete_version deletes version, returns data_dir - #[tracing::instrument(skip(self))] + #[tracing::instrument(level = "debug", skip(self))] pub fn delete_version(&mut self, fi: &FileInfo) -> Result> { let vid = Some(fi.version_id.unwrap_or(Uuid::nil())); diff --git a/crates/filemeta/src/replication.rs b/crates/filemeta/src/replication.rs index 15d4d0a4d..5675ca587 100644 --- a/crates/filemeta/src/replication.rs +++ b/crates/filemeta/src/replication.rs @@ -281,7 +281,7 @@ impl ReplicationState { return repl_status; } } - } else if self.replica_status != ReplicationStatusType::default() { + } else if !self.replica_status.is_empty() { return self.replica_status.clone(); } diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 3e25675f2..123ce57f3 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -1642,7 +1642,7 @@ impl S3 for FS { return Ok(S3Response::with_status(DeleteObjectOutput::default(), StatusCode::NO_CONTENT)); } - if obj_info.replication_status == ReplicationStatusType::Replica + if obj_info.replication_status == ReplicationStatusType::Pending || obj_info.version_purge_status == VersionPurgeStatusType::Pending { schedule_replication_delete(DeletedObjectReplicationInfo {