diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 97c5bee7e..e885be5b0 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -107,12 +107,13 @@ pub mod bucket { pub mod metadata_sys { pub use crate::bucket::metadata_sys::{ - BucketMetadataSys, delete, get, get_accelerate_config, get_bucket_policy, get_bucket_policy_raw, - get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config, + BucketMetadataSys, acquire_bucket_targets_transaction_lock, delete, get, get_accelerate_config, get_bucket_policy, + get_bucket_policy_raw, get_bucket_targets_config, get_config_from_disk, get_cors_config, get_durability_config, get_global_bucket_metadata_sys, get_lifecycle_config, get_logging_config, get_notification_config, get_object_lock_config, get_public_access_block_config, get_quota_config, get_replication_config, get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config, init_bucket_metadata_sys, list_bucket_targets, remove_bucket_metadata, set_bucket_metadata, update, + update_bucket_targets_under_transaction_lock, }; } @@ -158,8 +159,8 @@ pub mod bucket { ReplicationType, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType, delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool, get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta, - replication_status_to_filemeta, replication_statuses_map, replication_target_arns, should_remove_replication_target, - should_schedule_delete_replication, should_use_existing_delete_replication_info, + 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, validate_replication_config_target_arns, version_purge_status_to_filemeta, }; diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 96b984824..b4610d415 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -12,15 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::metadata::{BucketMetadata, load_bucket_metadata}; +use super::metadata::{BUCKET_TARGETS_FILE, BucketMetadata, load_bucket_metadata}; use super::quota::BucketQuota; use super::target::BucketTargets; use crate::bucket::bucket_target_sys::BucketTargetSys; use crate::bucket::metadata::load_bucket_metadata_parse; use crate::bucket::utils::is_meta_bucketname; +use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result, is_err_bucket_not_found}; use crate::runtime::sources as runtime_sources; use crate::storage_api_contracts::heal::HealOperations as _; +use crate::storage_api_contracts::namespace::NamespaceLocking as _; use crate::store::ECStore; use futures::future::join_all; use rustfs_common::heal_channel::HealOpts; @@ -221,11 +223,35 @@ pub(crate) async fn remove_bucket_metadata_in(ctx: &crate::runtime::instance::In pub async fn update(bucket: &str, config_file: &str, data: Vec) -> Result { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; + let _targets_guard = if config_file == BUCKET_TARGETS_FILE { + Some(acquire_bucket_targets_transaction_lock(bucket).await?) + } else { + None + }; let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; bucket_meta_sys.update(bucket, config_file, data).await } +pub async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result { + let bucket_meta_sys_lock = get_bucket_metadata_sys()?; + let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; + bucket_meta_sys.update(bucket, BUCKET_TARGETS_FILE, data).await +} + +pub async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { + let bucket_meta_sys_lock = get_bucket_metadata_sys()?; + let api = bucket_meta_sys_lock.read().await.object_store(); + let lock = api + .new_ns_lock(RUSTFS_META_BUCKET, &bucket_targets_transaction_lock_key(bucket)) + .await?; + Ok(lock.get_write_lock(crate::set_disk::get_lock_acquire_timeout()).await?) +} + +fn bucket_targets_transaction_lock_key(bucket: &str) -> String { + format!("bucket-targets/{bucket}/transaction.lock") +} + pub async fn delete(bucket: &str, config_file: &str) -> Result { let bucket_meta_sys_lock = get_bucket_metadata_sys()?; let mut bucket_meta_sys = bucket_meta_sys_lock.write().await; diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index eb04a752c..e81f383ce 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -70,7 +70,7 @@ pub use replication_object_decision_boundary::{ }; pub use replication_pool::{ DurableMrfBacklog, DynReplicationPool, ReplicationPoolTrait, get_global_replication_pool, get_global_replication_stats, - init_background_replication, read_durable_mrf_backlog, + init_background_replication, read_durable_mrf_backlog, resync_start_conflict_id, }; pub use replication_queue_boundary::{ DeletedObjectReplicationInfo, ReplicationHealQueueResult, ReplicationOperation, ReplicationPriority, diff --git a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs index 192082133..dde8f91c2 100644 --- a/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs @@ -53,6 +53,10 @@ impl ReplicationMetadataStore { format!("{REPLICATION_DIR}/{bucket}/{arn}") } + pub(crate) fn resync_admission_lock_key(bucket: &str) -> String { + format!("{REPLICATION_DIR}/{bucket}/admission.lock") + } + pub(crate) fn bucket_resync_dir_path(bucket: &str) -> String { path_join_buf(&[BUCKET_META_PREFIX, bucket, REPLICATION_DIR]) } @@ -73,6 +77,10 @@ mod tests { ReplicationMetadataStore::resync_lock_key("bucket-a", "arn-a"), ".replication/bucket-a/arn-a" ); + assert_eq!( + ReplicationMetadataStore::resync_admission_lock_key("bucket-a"), + ".replication/bucket-a/admission.lock" + ); assert_eq!( ReplicationMetadataStore::bucket_resync_dir_path("bucket-a"), "buckets/bucket-a/.replication" diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index c25e49d6e..f2972aae9 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -95,6 +95,24 @@ pub async fn read_durable_mrf_backlog(storage: Arc) - durable_mrf_backlog_from_read(ReplicationConfigStore::read(storage, ReplicationMetadataStore::MRF_REPLICATION_FILE).await) } +#[derive(Debug, thiserror::Error)] +#[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")] +struct ResyncActiveConflictError { + bucket: String, + arn: String, + active_resync_id: String, +} + +pub fn resync_start_conflict_id(error: &EcstoreError) -> Option<&str> { + match error { + EcstoreError::Io(io_error) => io_error + .get_ref()? + .downcast_ref::() + .map(|conflict| conflict.active_resync_id.as_str()), + _ => None, + } +} + /// Main replication pool structure #[derive(Debug)] pub struct ReplicationPool { @@ -975,47 +993,123 @@ impl ReplicationPool { } 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 - }); + let new_run = self.clone().admit_bucket_resync(opts.clone()).await?; + self.activate_bucket_resync(opts, !new_run).await + } - 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, - }, - ); + pub async fn admit_bucket_resync(self: Arc, opts: ResyncOpts) -> Result { + tokio::spawn(async move { self.admit_bucket_resync_transaction(opts).await }) + .await + .map_err(|error| EcstoreError::other(format!("replication resync admission task failed: {error}")))? + } - bucket_status.clone() + async fn admit_bucket_resync_transaction(self: Arc, opts: ResyncOpts) -> Result { + let admission_lock_key = ReplicationMetadataStore::resync_admission_lock_key(&opts.bucket); + let admission_lock = self + .storage + .new_ns_lock(ReplicationMetadataStore::rustfs_meta_bucket(), &admission_lock_key) + .await?; + // Lock order: bucket resync admission lock -> resync status config-object lock. + let _admission_guard = match admission_lock.get_write_lock(ReplicationLockTiming::acquire_timeout()).await { + Ok(guard) => guard, + Err(lock_error) => { + if let Ok(status) = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await { + self.resyncer.status_map.write().await.insert(opts.bucket.clone(), status); + } + return Err(EcstoreError::from(lock_error)); + } }; + let mut bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?; + if let Some(active) = bucket_status.targets_map.get(&opts.arn) { + if active.resync_id == opts.resync_id { + self.resyncer + .status_map + .write() + .await + .insert(opts.bucket.clone(), bucket_status); + return Ok(false); + } + if should_auto_resume_resync(active.resync_status) { + let active_resync_id = active.resync_id.clone(); + self.resyncer + .status_map + .write() + .await + .insert(opts.bucket.clone(), bucket_status); + return Err(EcstoreError::other(ResyncActiveConflictError { + bucket: opts.bucket.clone(), + arn: opts.arn.clone(), + active_resync_id, + })); + } + } + + let now = OffsetDateTime::now_utc(); + 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, + }, + ); + save_resync_status(&opts.bucket, &bucket_status, self.storage.clone()).await?; + self.resyncer + .status_map + .write() + .await + .insert(opts.bucket.clone(), bucket_status); + + Ok(true) + } + + pub async fn activate_bucket_resync(self: Arc, opts: ResyncOpts, recovering: bool) -> Result<(), EcstoreError> { + let bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?; + let Some(target_status) = bucket_status.targets_map.get(&opts.arn) else { + return Err(EcstoreError::other("replication resync admission is missing")); + }; + if target_status.resync_id != opts.resync_id { + return Err(EcstoreError::other(ResyncActiveConflictError { + bucket: opts.bucket.clone(), + arn: opts.arn.clone(), + active_resync_id: target_status.resync_id.clone(), + })); + } + if !should_auto_resume_resync(target_status.resync_status) { + return Ok(()); + } + self.resyncer + .status_map + .write() + .await + .insert(opts.bucket.clone(), bucket_status); 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; - }); + if resyncer.register_cancel_token(&opts, cancel_token.clone()).await { + tokio::spawn(async move { + Box::pin( + resyncer + .clone() + .resync_bucket(cancel_token, storage, recovering, opts.clone()), + ) + .await; + resyncer.clear_cancel_token(&opts).await; + }); + } Ok(()) } @@ -1168,9 +1262,10 @@ impl ReplicationPool { 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; + if 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; + } }); } @@ -1274,6 +1369,8 @@ pub trait ReplicationPoolTrait: std::fmt::Debug { 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 admit_bucket_resync(self: Arc, opts: ResyncOpts) -> Result; + async fn activate_bucket_resync(self: Arc, opts: ResyncOpts, recovering: bool) -> Result<(), EcstoreError>; async fn start_bucket_resync(self: Arc, opts: ResyncOpts) -> Result<(), EcstoreError>; async fn init_resync( self: Arc, @@ -1317,6 +1414,14 @@ impl ReplicationPoolTrait for ReplicationPool { self.cancel_bucket_resync(opts).await } + async fn admit_bucket_resync(self: Arc, opts: ResyncOpts) -> Result { + self.admit_bucket_resync(opts).await + } + + async fn activate_bucket_resync(self: Arc, opts: ResyncOpts, recovering: bool) -> Result<(), EcstoreError> { + self.activate_bucket_resync(opts, recovering).await + } + async fn start_bucket_resync(self: Arc, opts: ResyncOpts) -> Result<(), EcstoreError> { self.start_bucket_resync(opts).await } @@ -1580,7 +1685,9 @@ mod tests { use std::collections::HashMap; use std::fmt::{Debug, Formatter}; use std::io::Cursor; - use std::sync::atomic::AtomicUsize; + use std::sync::Mutex as StdMutex; + use std::sync::atomic::{AtomicBool, AtomicUsize}; + use tokio::io::AsyncReadExt; use tokio::sync::Notify; use uuid::Uuid; @@ -1589,10 +1696,16 @@ mod tests { type TestObjectInfoOrErr = StorageObjectInfoOrErr; struct LoadResyncSharedState { - data: Vec, + data: StdMutex>, lock_manager: Arc, first_read_started: Notify, + delay_first_read: AtomicBool, read_count: AtomicUsize, + write_count: AtomicUsize, + fail_next_write: AtomicBool, + block_next_write: AtomicBool, + write_started: Notify, + allow_write: Notify, } struct LoadResyncNodeStore { @@ -1633,22 +1746,31 @@ mod tests { _h: Self::HeaderMap, _opts: &Self::ObjectOptions, ) -> Result { - if object != ReplicationMetadataStore::bucket_resync_file_path("load-resync-lock") { + if !object.ends_with("/.replication/resync.bin") { return Err(EcstoreError::FileNotFound); } let read_index = self.shared.read_count.fetch_add(1, Ordering::SeqCst); - if read_index == 0 { + if read_index == 0 && self.shared.delay_first_read.load(Ordering::SeqCst) { self.shared.first_read_started.notify_waiters(); tokio::time::sleep(Duration::from_millis(1_500)).await; } - let data = self.shared.data.clone(); + let data = self + .shared + .data + .lock() + .expect("test data lock should not be poisoned") + .clone(); + if data.is_empty() { + return Err(EcstoreError::FileNotFound); + } + let size = i64::try_from(data.len()).expect("test metadata length should fit i64"); Ok(Self::GetObjectReader { - stream: Box::new(Cursor::new(data.clone())), + stream: Box::new(Cursor::new(data)), object_info: ObjectInfo { - size: data.len() as i64, - actual_size: data.len() as i64, + size, + actual_size: size, ..Default::default() }, buffered_body: None, @@ -1660,9 +1782,20 @@ mod tests { &self, _bucket: &str, _object: &str, - _data: &mut Self::PutObjectReader, + data: &mut Self::PutObjectReader, _opts: &Self::ObjectOptions, ) -> Result { + if self.shared.fail_next_write.swap(false, Ordering::SeqCst) { + return Err(EcstoreError::Unexpected); + } + if self.shared.block_next_write.swap(false, Ordering::SeqCst) { + self.shared.write_started.notify_one(); + self.shared.allow_write.notified().await; + } + let mut encoded = Vec::new(); + data.stream.read_to_end(&mut encoded).await.map_err(EcstoreError::from)?; + *self.shared.data.lock().expect("test data lock should not be poisoned") = encoded; + self.shared.write_count.fetch_add(1, Ordering::SeqCst); Ok(ObjectInfo::default()) } } @@ -1896,6 +2029,267 @@ mod tests { encode_resync_file(&status).expect("test resync metadata should encode") } + fn empty_resync_shared_state() -> Arc { + Arc::new(LoadResyncSharedState { + data: StdMutex::new(Vec::new()), + lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), + first_read_started: Notify::new(), + delay_first_read: AtomicBool::new(false), + read_count: AtomicUsize::new(0), + write_count: AtomicUsize::new(0), + fail_next_write: AtomicBool::new(false), + block_next_write: AtomicBool::new(false), + write_started: Notify::new(), + allow_write: Notify::new(), + }) + } + + async fn hold_resync_runtime_lock( + shared: &Arc, + bucket: &str, + arn: &str, + ) -> rustfs_lock::NamespaceLockGuard { + let lock = + rustfs_lock::NamespaceLock::with_local_manager("resync-start-blocker".to_string(), shared.lock_manager.clone()); + let lock = rustfs_lock::NamespaceLockWrapper::new( + lock, + rustfs_lock::ObjectKey::new( + ReplicationMetadataStore::rustfs_meta_bucket().to_string(), + ReplicationMetadataStore::resync_lock_key(bucket, arn), + ), + "blocker".to_string(), + ); + lock.get_write_lock(Duration::from_secs(1)) + .await + .expect("test should hold the runtime resync lock") + } + + fn test_resync_opts(bucket: &str, arn: &str, id: &str) -> ResyncOpts { + ResyncOpts { + bucket: bucket.to_string(), + arn: arn.to_string(), + resync_id: id.to_string(), + resync_before: Some(OffsetDateTime::UNIX_EPOCH), + } + } + + #[tokio::test] + async fn concurrent_resync_starts_accept_one_id_and_reject_the_other() { + let shared = empty_resync_shared_state(); + let first_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let second_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await; + let _runtime_guard = hold_resync_runtime_lock(&shared, "atomic-start", "arn:test").await; + + let first = first_pool + .clone() + .start_bucket_resync(test_resync_opts("atomic-start", "arn:test", "run-a")); + let second = second_pool + .clone() + .start_bucket_resync(test_resync_opts("atomic-start", "arn:test", "run-b")); + let (first, second) = tokio::join!(first, second); + + let (accepted_id, conflict) = match (first, second) { + (Ok(()), Err(conflict)) => ("run-a", conflict), + (Err(conflict), Ok(())) => ("run-b", conflict), + outcome => panic!("exactly one concurrent start should be accepted: {outcome:?}"), + }; + assert_eq!(resync_start_conflict_id(&conflict), Some(accepted_id)); + + let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("accepted status should be persisted"); + assert_eq!(persisted.targets_map["arn:test"].resync_id, accepted_id); + assert_eq!(persisted.targets_map["arn:test"].resync_status, ResyncStatusType::ResyncPending); + assert_eq!( + first_pool.resyncer.status_map.read().await["atomic-start"].targets_map["arn:test"].resync_id, + accepted_id + ); + assert_eq!( + second_pool.resyncer.status_map.read().await["atomic-start"].targets_map["arn:test"].resync_id, + accepted_id + ); + } + + #[tokio::test] + async fn same_resync_id_retry_is_idempotent_without_rewriting_status() { + let shared = empty_resync_shared_state(); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let _runtime_guard = hold_resync_runtime_lock(&shared, "same-id", "arn:test").await; + let opts = test_resync_opts("same-id", "arn:test", "run-a"); + + pool.clone() + .start_bucket_resync(opts.clone()) + .await + .expect("first start should be accepted"); + let first_status = pool + .resyncer + .status_map + .read() + .await + .get("same-id") + .expect("accepted status should be published") + .targets_map["arn:test"] + .clone(); + + pool.clone() + .start_bucket_resync(opts) + .await + .expect("same ID retry should be accepted idempotently"); + let retried_status = pool + .resyncer + .status_map + .read() + .await + .get("same-id") + .expect("retried status should remain published") + .targets_map["arn:test"] + .clone(); + + assert_eq!(shared.write_count.load(Ordering::SeqCst), 1); + assert_eq!(retried_status.resync_id, first_status.resync_id); + assert_eq!(retried_status.start_time, first_status.start_time); + assert_eq!(retried_status.resync_status, ResyncStatusType::ResyncPending); + assert_eq!(pool.resyncer.cancel_tokens.read().await.len(), 1); + } + + #[tokio::test] + async fn admitted_resync_waits_for_target_metadata_commit_before_activation() { + let shared = empty_resync_shared_state(); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let _runtime_guard = hold_resync_runtime_lock(&shared, "two-phase-start", "arn:test").await; + let opts = test_resync_opts("two-phase-start", "arn:test", "run-a"); + + let new_run = pool + .clone() + .admit_bucket_resync(opts.clone()) + .await + .expect("admission should persist the intent"); + assert!(new_run); + assert!(pool.resyncer.cancel_tokens.read().await.is_empty()); + assert_eq!(shared.write_count.load(Ordering::SeqCst), 1); + + pool.clone() + .activate_bucket_resync(opts, false) + .await + .expect("activation should start the admitted run"); + assert_eq!(pool.resyncer.cancel_tokens.read().await.len(), 1); + } + + #[tokio::test] + async fn same_id_retry_after_restart_recreates_missing_runtime_task() { + let shared = empty_resync_shared_state(); + let mut persisted = BucketReplicationResyncStatus::new(); + persisted.targets_map.insert( + "arn:test".to_string(), + TargetReplicationResyncStatus { + bucket: "restart-retry".to_string(), + resync_id: "run-a".to_string(), + resync_status: ResyncStatusType::ResyncPending, + ..Default::default() + }, + ); + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_resync_file(&persisted).expect("restart status should encode"); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let _runtime_guard = hold_resync_runtime_lock(&shared, "restart-retry", "arn:test").await; + + pool.clone() + .start_bucket_resync(test_resync_opts("restart-retry", "arn:test", "run-a")) + .await + .expect("same ID retry should recover an accepted run"); + + assert_eq!(shared.write_count.load(Ordering::SeqCst), 0); + assert_eq!(pool.resyncer.cancel_tokens.read().await.len(), 1); + assert_eq!( + pool.resyncer.status_map.read().await["restart-retry"].targets_map["arn:test"].resync_id, + "run-a" + ); + } + + #[tokio::test] + async fn same_completed_resync_id_retry_does_not_restart_work() { + let shared = empty_resync_shared_state(); + let mut persisted = BucketReplicationResyncStatus::new(); + persisted.targets_map.insert( + "arn:test".to_string(), + TargetReplicationResyncStatus { + bucket: "completed-retry".to_string(), + resync_id: "run-a".to_string(), + resync_status: ResyncStatusType::ResyncCompleted, + ..Default::default() + }, + ); + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_resync_file(&persisted).expect("completed status should encode"); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + + pool.clone() + .start_bucket_resync(test_resync_opts("completed-retry", "arn:test", "run-a")) + .await + .expect("completed same ID retry should remain idempotent"); + + assert_eq!(shared.write_count.load(Ordering::SeqCst), 0); + assert!(pool.resyncer.cancel_tokens.read().await.is_empty()); + assert_eq!( + pool.resyncer.status_map.read().await["completed-retry"].targets_map["arn:test"].resync_status, + ResyncStatusType::ResyncCompleted + ); + } + + #[tokio::test] + async fn start_failure_does_not_publish_or_persist_requested_id() { + let shared = empty_resync_shared_state(); + shared.fail_next_write.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + + let error = pool + .clone() + .start_bucket_resync(test_resync_opts("failed-start", "arn:test", "run-a")) + .await + .expect_err("metadata save failure should reject the start"); + + assert!(matches!(error, EcstoreError::Unexpected)); + assert!(shared.data.lock().expect("test data lock should not be poisoned").is_empty()); + assert!(!pool.resyncer.status_map.read().await.contains_key("failed-start")); + assert_eq!(shared.write_count.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn canceled_start_request_finishes_accepted_transaction() { + let shared = empty_resync_shared_state(); + shared.block_next_write.store(true, Ordering::SeqCst); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let _runtime_guard = hold_resync_runtime_lock(&shared, "canceled-start", "arn:test").await; + + let start_pool = pool.clone(); + let start = tokio::spawn(async move { + start_pool + .start_bucket_resync(test_resync_opts("canceled-start", "arn:test", "run-a")) + .await + }); + tokio::time::timeout(Duration::from_secs(10), shared.write_started.notified()) + .await + .expect("start transaction should reach the durable write"); + start.abort(); + assert!(start.await.expect_err("caller task should be canceled").is_cancelled()); + shared.allow_write.notify_one(); + + tokio::time::timeout(Duration::from_secs(10), async { + loop { + if pool.resyncer.status_map.read().await.contains_key("canceled-start") { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("detached admission transaction should finish after caller cancellation"); + assert_eq!(shared.write_count.load(Ordering::SeqCst), 1); + assert_eq!( + pool.resyncer.status_map.read().await["canceled-start"].targets_map["arn:test"].resync_id, + "run-a" + ); + } + #[test] fn replication_queue_admission_combines_target_results() { let mut admission = ReplicationQueueAdmission::Skipped; @@ -1985,10 +2379,16 @@ mod tests { async fn load_resync_leader_lock_allows_only_one_startup_recovery() { temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async { let shared = Arc::new(LoadResyncSharedState { - data: load_resync_test_metadata(), + data: StdMutex::new(load_resync_test_metadata()), lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()), first_read_started: Notify::new(), + delay_first_read: AtomicBool::new(true), read_count: AtomicUsize::new(0), + write_count: AtomicUsize::new(0), + fail_next_write: AtomicBool::new(false), + block_next_write: AtomicBool::new(false), + write_started: Notify::new(), + allow_write: Notify::new(), }); let leader_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; let skipped_pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index f9ecfced0..fb2f85ce5 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -255,11 +255,13 @@ fn resync_status_duration( Some(std::time::Duration::from_millis(millis)) } +type ResyncCancelKey = (String, String, String); + #[derive(Debug)] pub struct ReplicationResyncer { pub status_map: Arc>>, pub worker_size: usize, - pub cancel_tokens: Arc>>, + pub(crate) cancel_tokens: Arc>>, } impl ReplicationResyncer { @@ -271,12 +273,19 @@ impl ReplicationResyncer { } } - fn cancel_key(opts: &ResyncOpts) -> String { - format!("{}:{}", opts.bucket, opts.arn) + fn cancel_key(opts: &ResyncOpts) -> ResyncCancelKey { + (opts.bucket.clone(), opts.arn.clone(), opts.resync_id.clone()) } - pub async fn register_cancel_token(&self, opts: &ResyncOpts, token: CancellationToken) { - self.cancel_tokens.write().await.insert(Self::cancel_key(opts), token); + pub async fn register_cancel_token(&self, opts: &ResyncOpts, token: CancellationToken) -> bool { + let mut cancel_tokens = self.cancel_tokens.write().await; + match cancel_tokens.entry(Self::cancel_key(opts)) { + std::collections::hash_map::Entry::Vacant(entry) => { + entry.insert(token); + true + } + std::collections::hash_map::Entry::Occupied(_) => false, + } } pub async fn clear_cancel_token(&self, opts: &ResyncOpts) { diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index f65d57ac6..04d08f891 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -277,7 +277,7 @@ pub fn sanitize_resync_error_detail(detail: &str) -> Option { } pub fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &ResyncOpts) -> bool { - state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id + state.resync_id.is_empty() || state.resync_id == opts.resync_id } pub fn should_count_head_proxy_failure(is_not_found: bool, code: Option<&str>, raw_status: Option) -> bool { @@ -650,6 +650,13 @@ mod tests { assert!(resync_state_accepts_update(&TargetReplicationResyncStatus::default(), &matching)); assert!(resync_state_accepts_update(¤t, &matching)); assert!(!resync_state_accepts_update(¤t, &stale)); + assert!(!resync_state_accepts_update( + ¤t, + &ResyncOpts { + resync_id: String::new(), + ..matching + } + )); } #[test] diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index ab251be94..134e84a26 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -5845,7 +5845,20 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st status: "running".to_string(), ..Default::default() }; - let targets_guard = lock_bucket_targets_metadata(bucket).await; + let Some(pool) = current_replication_pool_handle() else { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = "replication pool is not initialized".to_string(); + return bucket_status; + }; + let _targets_guard = lock_bucket_targets_metadata(bucket).await; + let _transaction_guard = match metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await { + Ok(guard) => guard, + Err(_) => { + bucket_status.status = "failed".to_string(); + bucket_status.err_detail = "replication target metadata transaction lock is unavailable".to_string(); + return bucket_status; + } + }; let (config, _) = match metadata_sys::get_replication_config(bucket).await { Ok(config) => config, @@ -5856,7 +5869,7 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st } }; - let mut targets = match metadata_sys::list_bucket_targets(bucket).await { + let targets = match metadata_sys::list_bucket_targets_from_disk(bucket).await { Ok(targets) => targets, Err(err) => { bucket_status.status = "failed".to_string(); @@ -5864,13 +5877,6 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st return bucket_status; } }; - - let Some(pool) = current_replication_pool_handle() else { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = "replication pool is not initialized".to_string(); - return bucket_status; - }; - let Some(target_index) = targets .targets .iter() @@ -5898,7 +5904,7 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st let reset_before = Some(OffsetDateTime::now_utc()); let target_arn = { - let target = &mut targets.targets[target_index]; + let target = &targets.targets[target_index]; let (has_arn, existing_object_enabled) = config.has_existing_object_replication(&target.arn); if !has_arn || !existing_object_enabled { @@ -5907,35 +5913,46 @@ async fn start_site_bucket_resync(bucket: &str, target_arn: &str, resync_id: &st return bucket_status; } - target.reset_id = resync_id.to_string(); - target.reset_before_date = reset_before; target.arn.clone() }; - let json_targets = match serde_json::to_vec(&targets) { - Ok(json_targets) => json_targets, + let opts = replication::resync_opts(bucket, target_arn.clone(), resync_id, reset_before); + let admission_pool = pool.clone(); + let activation_pool = pool.clone(); + let committed_targets = match replication::commit_resync_target( + targets, + opts, + move |opts| async move { admission_pool.admit_bucket_resync(opts).await }, + move |encoded| async move { + metadata_sys::update_bucket_targets_under_transaction_lock(bucket, encoded) + .await + .map(|_| ()) + .map_err(|_| { + StorageError::other( + "replication resync was accepted but target metadata commit failed; retry the same resync ID to reconcile", + ) + }) + }, + move |opts, recovering| async move { activation_pool.activate_bucket_resync(opts, recovering).await }, + ) + .await + { + Ok(targets) => targets, Err(err) => { bucket_status.status = "failed".to_string(); - bucket_status.err_detail = err.to_string(); + if let Some(active_resync_id) = replication::resync_start_conflict_id(&err) { + bucket_status.status = "conflict".to_string(); + bucket_status.err_detail = + format!("replication resync {active_resync_id} is already active for this target"); + } else { + bucket_status.err_detail = err.to_string(); + } return bucket_status; } }; - - if let Err(err) = metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets).await { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = err.to_string(); - return bucket_status; - } - BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await; - drop(targets_guard); - - if let Err(err) = pool - .start_bucket_resync(replication::resync_opts(bucket, target_arn, resync_id, reset_before)) - .await - { - bucket_status.status = "failed".to_string(); - bucket_status.err_detail = err.to_string(); - } + BucketTargetSys::get() + .update_all_targets(bucket, Some(&committed_targets)) + .await; bucket_status } diff --git a/rustfs/src/admin/router.rs b/rustfs/src/admin/router.rs index e60fbe4aa..dd4e537eb 100644 --- a/rustfs/src/admin/router.rs +++ b/rustfs/src/admin/router.rs @@ -13,7 +13,6 @@ // limitations under the License. use super::storage_api::bucket::bandwidth::monitor::BandwidthDetails; -use super::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use super::storage_api::bucket::metadata_sys; use super::storage_api::bucket::replication::{self, BucketReplicationResyncStatus, BucketStats, ReplicationStatusType}; use super::storage_api::bucket::target::{BucketTarget, BucketTargetType, BucketTargets}; @@ -1738,14 +1737,18 @@ fn build_replication_reset_response(targets: Vec) -> S3R Ok(resp) } -fn apply_replication_reset_to_targets(targets: &mut BucketTargets, reset: &ReplicationResetStartRequest) -> S3Result<()> { - let Some(target) = targets.targets.iter_mut().find(|target| target.arn == reset.arn) else { - return Err(s3_error!(InvalidRequest, "replication reset arn is not configured for this bucket")); - }; +fn map_replication_resync_start_error(error: StorageError) -> S3Error { + match replication::resync_start_conflict_id(&error) { + Some(active_resync_id) => replication_resync_active_conflict_error(active_resync_id), + None => s3_error!(InternalError, "{error}"), + } +} - target.reset_id = reset.reset_id.clone(); - target.reset_before_date = reset.reset_before; - Ok(()) +fn replication_resync_active_conflict_error(active_resync_id: &str) -> S3Error { + s3_error!( + OperationAborted, + "replication resync {active_resync_id} is already active for this target" + ) } fn parse_reset_status_target(uri: &Uri) -> ReplicationResetStatusRequest { @@ -2467,34 +2470,43 @@ async fn target_client_object_lock_enabled_with_client( } async fn start_replication_resync(bucket: &str, reset: &ReplicationResetStartRequest) -> S3Result { - let targets_guard = lock_bucket_targets_metadata(bucket).await; - let (config, _) = metadata_sys::get_replication_config(bucket).await.map_err(ApiError::from)?; - let resolved_arn = resolve_replication_reset_target_arn(&config, &reset.arn)?; - let mut resolved_reset = reset.clone(); - resolved_reset.arn = resolved_arn.clone(); - - let mut targets = metadata_sys::list_bucket_targets(bucket).await.map_err(ApiError::from)?; - apply_replication_reset_to_targets(&mut targets, &resolved_reset)?; - - let json_targets = serde_json::to_vec(&targets).map_err(|e| s3_error!(InternalError, "{e}"))?; - metadata_sys::update(bucket, BUCKET_TARGETS_FILE, json_targets) - .await - .map_err(ApiError::from)?; - BucketTargetSys::get().update_all_targets(bucket, Some(&targets)).await; - drop(targets_guard); - let Some(pool) = current_replication_pool_handle() else { return Err(s3_error!(InternalError, "replication pool is not initialized")); }; - pool.start_bucket_resync(replication::resync_opts( - bucket, - resolved_arn.clone(), - &reset.reset_id, - reset.reset_before, - )) + let _targets_guard = lock_bucket_targets_metadata(bucket).await; + let _transaction_guard = metadata_sys::acquire_bucket_targets_transaction_lock(bucket) + .await + .map_err(ApiError::from)?; + let (config, _) = metadata_sys::get_replication_config(bucket).await.map_err(ApiError::from)?; + let resolved_arn = resolve_replication_reset_target_arn(&config, &reset.arn)?; + let targets = metadata_sys::list_bucket_targets_from_disk(bucket) + .await + .map_err(ApiError::from)?; + let opts = replication::resync_opts(bucket, resolved_arn.clone(), &reset.reset_id, reset.reset_before); + let admission_pool = pool.clone(); + let activation_pool = pool.clone(); + let committed_targets = replication::commit_resync_target( + targets, + opts, + move |opts| async move { admission_pool.admit_bucket_resync(opts).await }, + move |encoded| async move { + metadata_sys::update_bucket_targets_under_transaction_lock(bucket, encoded) + .await + .map(|_| ()) + .map_err(|_| { + StorageError::other( + "replication resync was accepted but target metadata commit failed; retry the same reset ID to reconcile", + ) + }) + }, + move |opts, recovering| async move { activation_pool.activate_bucket_resync(opts, recovering).await }, + ) .await - .map_err(|e| s3_error!(InternalError, "{e}"))?; + .map_err(map_replication_resync_start_error)?; + BucketTargetSys::get() + .update_all_targets(bucket, Some(&committed_targets)) + .await; Ok(ReplicationResetTarget { arn: resolved_arn, @@ -3197,23 +3209,12 @@ mod tests { } #[test] - fn apply_replication_reset_to_targets_updates_matching_target() { - let mut targets = BucketTargets { - targets: vec![crate::admin::storage_api::bucket::target::BucketTarget { - arn: "arn:target".to_string(), - ..Default::default() - }], - }; - let reset = ReplicationResetStartRequest { - arn: "arn:target".to_string(), - reset_id: "rid-1".to_string(), - reset_before: Some(OffsetDateTime::now_utc()), - }; + fn active_replication_resync_conflict_maps_to_http_conflict() { + let error = replication_resync_active_conflict_error("run-active"); - apply_replication_reset_to_targets(&mut targets, &reset).expect("target update should succeed"); - - assert_eq!(targets.targets[0].reset_id, "rid-1"); - assert_eq!(targets.targets[0].reset_before_date, reset.reset_before); + assert_eq!(error.code(), &S3ErrorCode::OperationAborted); + assert_eq!(error.status_code(), Some(StatusCode::CONFLICT)); + assert!(error.message().unwrap_or_default().contains("run-active")); } #[test] diff --git a/rustfs/src/admin/storage_api.rs b/rustfs/src/admin/storage_api.rs index f875094e7..fa21b756e 100644 --- a/rustfs/src/admin/storage_api.rs +++ b/rustfs/src/admin/storage_api.rs @@ -264,6 +264,14 @@ pub(crate) mod metadata_sys { crate::storage::storage_api::update_bucket_metadata_config(bucket, config_file, data).await } + pub(crate) async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { + crate::storage::storage_api::acquire_bucket_targets_transaction_lock(bucket).await + } + + pub(crate) async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result { + crate::storage::storage_api::update_bucket_targets_under_transaction_lock(bucket, data).await + } + pub(crate) async fn delete(bucket: &str, config_file: &str) -> Result { crate::storage::storage_api::delete_bucket_metadata_config(bucket, config_file).await } @@ -280,6 +288,14 @@ pub(crate) mod metadata_sys { super::ecstore_bucket::metadata_sys::get_config_from_disk(bucket).await } + pub(crate) async fn list_bucket_targets_from_disk(bucket: &str) -> Result { + let metadata = get_config_from_disk(bucket).await?; + if metadata.bucket_targets_config_json.is_empty() { + return Ok(BucketTargets::default()); + } + serde_json::from_slice(&metadata.bucket_targets_config_json).map_err(super::Error::other) + } + pub(crate) async fn get_lifecycle_config(bucket: &str) -> Result<(BucketLifecycleConfiguration, OffsetDateTime)> { super::ecstore_bucket::metadata_sys::get_lifecycle_config(bucket).await } @@ -363,6 +379,174 @@ pub(crate) mod replication { resync_before, } } + + pub(crate) fn resync_start_conflict_id(error: &super::Error) -> Option<&str> { + super::ecstore_bucket::replication::resync_start_conflict_id(error) + } + + pub(crate) async fn commit_resync_target( + mut targets: super::target::BucketTargets, + opts: ResyncOpts, + admit: Admit, + persist: Persist, + activate: Activate, + ) -> super::Result + where + Admit: FnOnce(ResyncOpts) -> AdmitFuture, + AdmitFuture: std::future::Future>, + Persist: FnOnce(Vec) -> PersistFuture, + PersistFuture: std::future::Future>, + Activate: FnOnce(ResyncOpts, bool) -> ActivateFuture, + ActivateFuture: std::future::Future>, + { + let target = targets + .targets + .iter_mut() + .find(|target| target.arn == opts.arn) + .ok_or_else(|| super::Error::other("replication resync target is not configured"))?; + target.reset_id = opts.resync_id.clone(); + target.reset_before_date = opts.resync_before; + let encoded = serde_json::to_vec(&targets).map_err(super::Error::other)?; + + let new_run = admit(opts.clone()).await?; + persist(encoded).await?; + activate(opts, !new_run).await?; + Ok(targets) + } + + #[cfg(test)] + mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; + use tokio::sync::Mutex; + + fn targets() -> super::super::target::BucketTargets { + super::super::target::BucketTargets { + targets: vec![ + super::super::target::BucketTarget { + arn: "arn:primary".to_string(), + reset_id: "old-primary".to_string(), + ..Default::default() + }, + super::super::target::BucketTarget { + arn: "arn:other".to_string(), + reset_id: "other-node-value".to_string(), + ..Default::default() + }, + ], + } + } + + fn opts() -> ResyncOpts { + ResyncOpts { + bucket: "bucket".to_string(), + arn: "arn:primary".to_string(), + resync_id: "accepted-id".to_string(), + resync_before: Some(time::OffsetDateTime::UNIX_EPOCH), + } + } + + #[tokio::test] + async fn admission_failure_never_mutates_or_persists_target_metadata() { + let persist_calls = Arc::new(AtomicUsize::new(0)); + let activate_calls = Arc::new(AtomicUsize::new(0)); + let original = targets(); + let error = commit_resync_target( + original.clone(), + opts(), + |_| async { Err(super::super::Error::other("pool unavailable")) }, + { + let persist_calls = persist_calls.clone(); + move |_| async move { + persist_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + }, + { + let activate_calls = activate_calls.clone(); + move |_, _| async move { + activate_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + }, + ) + .await + .expect_err("admission failure must fail closed"); + + assert!(error.to_string().contains("pool unavailable")); + assert_eq!(persist_calls.load(Ordering::SeqCst), 0); + assert_eq!(activate_calls.load(Ordering::SeqCst), 0); + assert_eq!(original.targets[0].reset_id, "old-primary"); + } + + #[tokio::test] + async fn commit_failure_after_durable_admission_waits_for_recovery_without_rollback() { + let admitted = Arc::new(AtomicBool::new(false)); + let activate_calls = Arc::new(AtomicUsize::new(0)); + let result = commit_resync_target( + targets(), + opts(), + { + let admitted = admitted.clone(); + move |_| async move { + admitted.store(true, Ordering::SeqCst); + Ok(true) + } + }, + |_| async { Err(super::super::Error::other("injected target write failure")) }, + { + let activate_calls = activate_calls.clone(); + move |_, _| async move { + activate_calls.fetch_add(1, Ordering::SeqCst); + Ok(()) + } + }, + ) + .await; + + assert!(result.is_err()); + assert!(admitted.load(Ordering::SeqCst)); + assert_eq!(activate_calls.load(Ordering::SeqCst), 0); + } + + #[tokio::test] + async fn committed_resync_changes_only_its_target_and_activates_after_write() { + let persisted = Arc::new(Mutex::new(Vec::new())); + let write_finished = Arc::new(AtomicBool::new(false)); + let committed = commit_resync_target( + targets(), + opts(), + |_| async { Ok(true) }, + { + let persisted = persisted.clone(); + let write_finished = write_finished.clone(); + move |encoded| async move { + *persisted.lock().await = encoded; + write_finished.store(true, Ordering::SeqCst); + Ok(()) + } + }, + { + let write_finished = write_finished.clone(); + move |_, recovering| async move { + assert!(!recovering); + assert!(write_finished.load(Ordering::SeqCst)); + Ok(()) + } + }, + ) + .await + .expect("transaction should commit"); + + assert_eq!(committed.targets[0].reset_id, "accepted-id"); + assert_eq!(committed.targets[1].reset_id, "other-node-value"); + let persisted: super::super::target::BucketTargets = + serde_json::from_slice(&persisted.lock().await).expect("persisted targets should decode"); + assert_eq!(persisted.targets[0].reset_id, "accepted-id"); + assert_eq!(persisted.targets[1].reset_id, "other-node-value"); + } + } } pub(crate) mod target { diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index 4150e0288..0f5f0e239 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -15,7 +15,8 @@ use crate::startup_runtime_sources; use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::bucket_metadata::{ - ECStore, init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config, + ECStore, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, + try_migrate_iam_config, }; use std::{ io::{Error, Result}, @@ -37,6 +38,7 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc) - try_migrate_bucket_metadata(store.clone()).await; init_bucket_metadata_sys(store.clone(), buckets.clone()).await; try_migrate_iam_config(store).await; + reconcile_bucket_resync_target_intents(&buckets).await?; Ok(buckets) } @@ -54,12 +56,13 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance try_migrate_bucket_metadata(store.clone()).await; + try_migrate_iam_config(store.clone()).await; + init_bucket_metadata_sys(store, buckets.clone()).await; + reconcile_bucket_resync_target_intents(&buckets).await?; + if let Some(pool) = startup_runtime_sources::replication_pool_handle() { pool.init_resync(ctx, buckets.clone()).await?; } - try_migrate_iam_config(store.clone()).await; - init_bucket_metadata_sys(store, buckets.clone()).await; - Ok(buckets) } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 413412628..91fb0430e 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -669,8 +669,16 @@ impl StorageReplicationPoolHandle { self.inner.clone().cancel_bucket_resync(opts).await } - pub(crate) async fn start_bucket_resync(&self, opts: ecstore_bucket::replication::ResyncOpts) -> Result<()> { - self.inner.clone().start_bucket_resync(opts).await + pub(crate) async fn admit_bucket_resync(&self, opts: ecstore_bucket::replication::ResyncOpts) -> Result { + self.inner.clone().admit_bucket_resync(opts).await + } + + pub(crate) async fn activate_bucket_resync( + &self, + opts: ecstore_bucket::replication::ResyncOpts, + recovering: bool, + ) -> Result<()> { + self.inner.clone().activate_bucket_resync(opts, recovering).await } pub(crate) async fn init_resync(self: Arc, ctx: CancellationToken, buckets: Vec) -> Result<()> { @@ -777,6 +785,60 @@ pub(crate) async fn init_background_replication(store: Arc) { ecstore_bucket::replication::init_background_replication(store).await; } +fn apply_active_resync_intents( + targets: &mut ecstore_bucket::target::BucketTargets, + status: &ecstore_bucket::replication::BucketReplicationResyncStatus, +) -> Result { + let mut changed = false; + for (arn, intent) in &status.targets_map { + if !matches!( + intent.resync_status, + ecstore_bucket::replication::ResyncStatusType::ResyncPending + | ecstore_bucket::replication::ResyncStatusType::ResyncStarted + ) { + continue; + } + let target = targets + .targets + .iter_mut() + .find(|target| target.arn == *arn) + .ok_or_else(|| Error::other(format!("accepted replication resync target {arn} is not configured")))?; + if target.reset_id != intent.resync_id || target.reset_before_date != intent.resync_before_date { + target.reset_id = intent.resync_id.clone(); + target.reset_before_date = intent.resync_before_date; + changed = true; + } + } + Ok(changed) +} + +pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String]) -> Result<()> { + let Some(pool) = ecstore_bucket::replication::get_global_replication_pool() else { + return Err(Error::other("replication pool is not initialized")); + }; + + for bucket in buckets { + let _transaction_guard = ecstore_bucket::metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await?; + let status = pool.get_bucket_resync_status(bucket).await?; + if status.targets_map.is_empty() { + continue; + } + let metadata = ecstore_bucket::metadata_sys::get_config_from_disk(bucket).await?; + let mut targets = if metadata.bucket_targets_config_json.is_empty() { + ecstore_bucket::target::BucketTargets::default() + } else { + serde_json::from_slice(&metadata.bucket_targets_config_json).map_err(Error::other)? + }; + if !apply_active_resync_intents(&mut targets, &status)? { + continue; + } + let encoded = serde_json::to_vec(&targets).map_err(Error::other)?; + ecstore_bucket::metadata_sys::update_bucket_targets_under_transaction_lock(bucket, encoded).await?; + } + + Ok(()) +} + pub(crate) async fn all_local_disk() -> Vec { ecstore_storage::all_local_disk().await } @@ -1340,6 +1402,14 @@ pub(crate) async fn update_bucket_metadata_config( Ok(updated_at) } +pub(crate) async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result { + ecstore_bucket::metadata_sys::acquire_bucket_targets_transaction_lock(bucket).await +} + +pub(crate) async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Vec) -> Result { + ecstore_bucket::metadata_sys::update_bucket_targets_under_transaction_lock(bucket, data).await +} + fn record_scanner_maintenance_config_change(bucket: &str, config_file: &str) { if scanner_maintenance_config_file(config_file) { rustfs_scanner::record_scanner_maintenance_change(bucket); @@ -1593,7 +1663,8 @@ pub(crate) async fn init_compression_total_memory_from_backend(store: Arc