diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 42934318a..9bc1951b5 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -39,7 +39,7 @@ use super::replication_resync_boundary::{ }; use super::replication_resyncer::{ ReplicationResyncer, get_heal_replicate_object_info, replicate_delete, replicate_delete_with_outcome, replicate_object, - replicate_object_with_outcome, save_resync_status, + replicate_object_with_outcome, update_resync_status_cas, }; use super::replication_state::ReplicationStats; use super::replication_storage_boundary::{ @@ -1898,7 +1898,7 @@ impl ReplicationPool { } }; - let mut bucket_status = load_bucket_resync_metadata(&opts.bucket, self.storage.clone()).await?; + let 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 @@ -1924,26 +1924,43 @@ impl ReplicationPool { } 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, - }, - ); + let admitted = 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?; + // The admission lock serializes competing admissions, but status + // writers (mark_status, the periodic saver) do not take it — write + // through the CAS so their concurrent updates to other targets are + // never lost, re-checking the conflict gate on each retry. + let (bucket_status, _) = update_resync_status_cas(&opts.bucket, self.storage.clone(), |persisted| { + if let Some(active) = persisted.targets_map.get(&opts.arn) { + if active.resync_id == opts.resync_id { + return Ok(false); + } + if should_auto_resume_resync(active.resync_status) { + return Err(EcstoreError::other(ResyncActiveConflictError { + bucket: opts.bucket.clone(), + arn: opts.arn.clone(), + active_resync_id: active.resync_id.clone(), + })); + } + } + persisted.last_update = Some(now); + persisted.targets_map.insert(opts.arn.clone(), admitted.clone()); + Ok(true) + }) + .await?; self.resyncer .status_map .write() @@ -1953,6 +1970,83 @@ impl ReplicationPool { Ok(true) } + /// Cancel the pending/started resync intent recorded for `arn`, if any, + /// because its remote target is being removed. Returns the canceled run. + /// + /// Runs under the bucket admission lock and reloads `resync.bin` from disk + /// before writing, like admission does: this node's `status_map` entry may + /// be stale relative to intents admitted by other nodes, and persisting it + /// would silently drop their durable restart intents. + pub async fn cancel_bucket_resync_for_removed_target( + self: Arc, + bucket: &str, + arn: &str, + ) -> Result, EcstoreError> { + let bucket = bucket.to_string(); + let arn = arn.to_string(); + tokio::spawn(async move { self.cancel_bucket_resync_for_removed_target_transaction(bucket, arn).await }) + .await + .map_err(|error| EcstoreError::other(format!("replication resync cancellation task failed: {error}")))? + } + + async fn cancel_bucket_resync_for_removed_target_transaction( + self: Arc, + bucket: String, + arn: String, + ) -> Result, EcstoreError> { + let admission_lock_key = ReplicationMetadataStore::resync_admission_lock_key(&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 = admission_lock + .get_write_lock(ReplicationLockTiming::acquire_timeout()) + .await + .map_err(EcstoreError::from)?; + + let mut canceled: Option = None; + let (final_map, _) = update_resync_status_cas(&bucket, self.storage.clone(), |persisted| { + canceled = None; + let Some(intent) = persisted.targets_map.get_mut(&arn) else { + return Ok(false); + }; + if !should_auto_resume_resync(intent.resync_status) { + return Ok(false); + } + let now = OffsetDateTime::now_utc(); + canceled = Some(ResyncOpts { + bucket: bucket.clone(), + arn: arn.clone(), + resync_id: intent.resync_id.clone(), + resync_before: intent.resync_before_date, + }); + intent.resync_status = ResyncStatusType::ResyncCanceled; + intent.last_update = Some(now); + persisted.last_update = Some(now); + Ok(true) + }) + .await?; + + // Converge only the removed target's cached entry: cached progress + // counters for this node's other running targets stay authoritative. + { + let mut status_map = self.resyncer.status_map.write().await; + let cached = status_map + .entry(bucket.clone()) + .or_insert_with(BucketReplicationResyncStatus::new); + if let Some(final_target) = final_map.targets_map.get(&arn) { + cached.targets_map.insert(arn.clone(), final_target.clone()); + cached.last_update = final_map.last_update.or(cached.last_update); + } + } + + if let Some(opts) = &canceled { + self.resyncer.cancel(opts).await; + } + Ok(canceled) + } + 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 { @@ -2710,6 +2804,11 @@ 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 cancel_bucket_resync_for_removed_target( + self: Arc, + bucket: &str, + arn: &str, + ) -> 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>; @@ -2763,6 +2862,14 @@ impl ReplicationPoolTrait for ReplicationPool { self.cancel_bucket_resync(opts).await } + async fn cancel_bucket_resync_for_removed_target( + self: Arc, + bucket: &str, + arn: &str, + ) -> Result, EcstoreError> { + ReplicationPool::::cancel_bucket_resync_for_removed_target(self, bucket, arn).await + } + async fn admit_bucket_resync(self: Arc, opts: ResyncOpts) -> Result { self.admit_bucket_resync(opts).await } @@ -3241,11 +3348,17 @@ mod tests { async fn put_object( &self, - _bucket: &str, + bucket: &str, object: &str, data: &mut Self::PutObjectReader, opts: &Self::ObjectOptions, ) -> Result { + let _lock_guard = if opts.no_lock { + None + } else { + let lock = self.new_ns_lock(bucket, object).await?; + Some(lock.get_write_lock(Duration::from_secs(10)).await?) + }; if opts.http_preconditions.is_some() && let Some(replacement) = self .shared @@ -3891,6 +4004,157 @@ mod tests { assert_eq!(pool.resyncer.cancel_tokens.read().await.len(), 1); } + /// Removing a target on node A must cancel only A's intent. Node A's cached + /// status map predates node B's admission, so a cancel that persisted the + /// cache would erase B's durable restart intent for `arn:second`. + #[tokio::test] + async fn removed_target_cancel_preserves_intents_admitted_on_other_nodes() { + let shared = empty_resync_shared_state(); + let node_a = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let node_b = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await; + let bucket = "removed-target-cancel"; + + assert!( + node_a + .clone() + .admit_bucket_resync(test_resync_opts(bucket, "arn:first", "run-a")) + .await + .expect("node A admission should persist") + ); + assert!( + node_b + .clone() + .admit_bucket_resync(test_resync_opts(bucket, "arn:second", "run-b")) + .await + .expect("node B admission should persist") + ); + assert!( + !node_a.resyncer.status_map.read().await[bucket] + .targets_map + .contains_key("arn:second"), + "precondition: node A's cache must be stale relative to node B's admission" + ); + + let canceled = node_a + .clone() + .cancel_bucket_resync_for_removed_target(bucket, "arn:first") + .await + .expect("cancel should succeed"); + assert_eq!(canceled.map(|opts| opts.resync_id), Some("run-a".to_string())); + + let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("persisted status should decode"); + assert_eq!(persisted.targets_map["arn:first"].resync_status, ResyncStatusType::ResyncCanceled); + assert_eq!(persisted.targets_map["arn:second"].resync_status, ResyncStatusType::ResyncPending); + assert_eq!(persisted.targets_map["arn:second"].resync_id, "run-b"); + // Cache convergence is per-target: only the removed ARN is written + // back (a running target's cached progress counters stay + // authoritative), so node A's cache reflects the cancel while the + // persisted document remains the authority for `arn:second`. + assert_eq!( + node_a.resyncer.status_map.read().await[bucket].targets_map["arn:first"].resync_status, + ResyncStatusType::ResyncCanceled + ); + + let untouched = node_a + .clone() + .cancel_bucket_resync_for_removed_target(bucket, "arn:first") + .await + .expect("cancel of a terminal intent should be a no-op"); + assert!(untouched.is_none()); + assert!( + node_a + .clone() + .cancel_bucket_resync_for_removed_target(bucket, "arn:missing") + .await + .expect("cancel of an unknown arn should be a no-op") + .is_none() + ); + } + + /// The reviewer's resurrect scenario: after node A cancels `arn:first`, a + /// status write from node B — whose cache still holds the pre-cancel map — + /// must not flip `arn:first` back to `Pending` on disk. `mark_status` now + /// persists through the CAS with per-target guards instead of blind-saving + /// its cached whole-bucket map. + #[tokio::test] + async fn stale_peer_status_write_cannot_resurrect_canceled_intent() { + let shared = empty_resync_shared_state(); + let node_a = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + let node_b = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-b", shared.clone()))).await; + let bucket = "stale-peer-write"; + + assert!( + node_a + .clone() + .admit_bucket_resync(test_resync_opts(bucket, "arn:first", "run-a")) + .await + .expect("node A admission should persist") + ); + assert!( + node_b + .clone() + .admit_bucket_resync(test_resync_opts(bucket, "arn:second", "run-b")) + .await + .expect("node B admission should persist") + ); + // Seed node B's stale cache: it saw the map before A's cancel. + let pre_cancel = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("pre-cancel status should decode"); + node_b + .resyncer + .status_map + .write() + .await + .insert(bucket.to_string(), pre_cancel); + + node_a + .clone() + .cancel_bucket_resync_for_removed_target(bucket, "arn:first") + .await + .expect("cancel should succeed") + .expect("cancel should report the canceled run"); + + node_b + .resyncer + .mark_status( + ResyncStatusType::ResyncStarted, + test_resync_opts(bucket, "arn:second", "run-b"), + node_b.storage.clone(), + ) + .await + .expect("peer status write should succeed"); + + let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("persisted status should decode"); + assert_eq!( + persisted.targets_map["arn:first"].resync_status, + ResyncStatusType::ResyncCanceled, + "peer's stale cache must not resurrect the canceled intent" + ); + assert_eq!(persisted.targets_map["arn:second"].resync_status, ResyncStatusType::ResyncStarted); + + // And the reverse guard: a stale write for the canceled target itself + // is refused outright. + node_b + .resyncer + .mark_status( + ResyncStatusType::ResyncStarted, + test_resync_opts(bucket, "arn:first", "run-a"), + node_b.storage.clone(), + ) + .await + .expect("guarded status write should be skipped, not fail"); + let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("persisted status should decode"); + assert_eq!(persisted.targets_map["arn:first"].resync_status, ResyncStatusType::ResyncCanceled); + assert_eq!( + node_b.resyncer.status_map.read().await[bucket].targets_map["arn:first"].resync_status, + ResyncStatusType::ResyncCanceled, + "the refused writer's cache must converge to the persisted terminal state" + ); + } + #[tokio::test] async fn admitted_resync_waits_for_target_metadata_commit_before_activation() { let shared = empty_resync_shared_state(); @@ -4030,6 +4294,77 @@ mod tests { ); } + #[tokio::test] + async fn concurrent_resync_status_cas_preserves_both_mutations() { + let shared = empty_resync_shared_state(); + let mut seeded = BucketReplicationResyncStatus::new(); + for arn in ["arn:a", "arn:b"] { + seeded.targets_map.insert( + arn.to_string(), + TargetReplicationResyncStatus { + bucket: "cas-race".to_string(), + resync_id: format!("run-{arn}"), + resync_status: ResyncStatusType::ResyncPending, + ..Default::default() + }, + ); + } + *shared.data.lock().expect("test data lock should not be poisoned") = + encode_resync_file(&seeded).expect("seeded resync status should encode"); + shared.empty_object_exists.store(true, Ordering::SeqCst); + shared.etag_revision.store(1, Ordering::SeqCst); + shared.block_next_write.store(true, Ordering::SeqCst); + let node_a = Arc::new(LoadResyncNodeStore::new("cas-node-a", shared.clone())); + let node_b = Arc::new(LoadResyncNodeStore::new("cas-node-b", shared.clone())); + + let writer_a = tokio::spawn(async move { + update_resync_status_cas("cas-race", node_a, |status| { + status + .targets_map + .get_mut("arn:a") + .expect("seeded target A should exist") + .resync_status = ResyncStatusType::ResyncCanceled; + Ok(true) + }) + .await + }); + + tokio::time::timeout(Duration::from_secs(10), shared.write_started.notified()) + .await + .expect("writer A should pause after its precondition check"); + + let mut writer_b = tokio::spawn(async move { + update_resync_status_cas("cas-race", node_b, |status| { + status + .targets_map + .get_mut("arn:b") + .expect("seeded target B should exist") + .resync_status = ResyncStatusType::ResyncCompleted; + Ok(true) + }) + .await + }); + let writer_b_before_release = tokio::time::timeout(Duration::from_millis(250), &mut writer_b).await.ok(); + + shared.allow_write.notify_one(); + writer_a + .await + .expect("writer A task should finish") + .expect("writer A should report a successful conditional save"); + match writer_b_before_release { + Some(result) => result, + None => writer_b.await, + } + .expect("writer B task should finish") + .expect("writer B should retry and save its mutation"); + + let persisted = decode_resync_file(&shared.data.lock().expect("test data lock should not be poisoned")) + .expect("persisted resync status should decode"); + assert_eq!(persisted.targets_map["arn:a"].resync_status, ResyncStatusType::ResyncCanceled); + assert_eq!(persisted.targets_map["arn:b"].resync_status, ResyncStatusType::ResyncCompleted); + assert!(!shared.last_put_no_lock.load(Ordering::SeqCst)); + } + #[test] fn replication_queue_admission_combines_target_results() { let mut admission = ReplicationQueueAdmission::Skipped; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 587be7ae3..2fb6145d3 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -40,16 +40,17 @@ use super::replication_resync_boundary::ResyncStatusType; #[cfg(test)] use super::replication_resync_boundary::should_count_head_proxy_failure; use super::replication_resync_boundary::{ - BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, encode_resync_file, is_version_id_mismatch, - resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail, + BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus, decode_resync_file, encode_resync_file, + is_version_id_mismatch, resync_state_accepts_update, resync_status_duration, sanitize_resync_error_detail, + should_auto_resume_resync, }; #[cfg(test)] -use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file}; +use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX}; #[cfg(test)] use super::replication_storage_boundary::ReplicationDeletedObject; use super::replication_storage_boundary::{ - AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, - ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, + AdvancedGetOptions, EcstoreObjectOperations, GetObjectReader, HTTPPreconditions, HTTPRangeSpec, ObjectInfo, ObjectOptions, + ObjectToDelete, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, StorageObjectInfoOrErr, WalkOptions, }; use super::replication_target_boundary::{ ERR_REPLICATION_SSEC_PASSTHROUGH_UNSUPPORTED, HeadObjectSdkError, PutObjectOptions, PutObjectPartOptions, @@ -428,7 +429,7 @@ impl ReplicationResyncer { where S: ReplicationObjectIO, { - let (bucket_status, status_duration) = { + let (updated_target, status_duration) = { let mut status_map = self.status_map.write().await; let now = OffsetDateTime::now_utc(); @@ -499,28 +500,62 @@ impl ReplicationResyncer { bucket_status.last_update = Some(now); - (bucket_status.clone(), status_duration) + (state.clone(), status_duration) }; - save_resync_status(&opts.bucket, &bucket_status, obj_layer.clone()).await?; - if status != ResyncStatusType::ResyncCanceled { - let canceled_status = self - .status_map - .read() - .await - .get(&opts.bucket) - .filter(|current| { - current.targets_map.get(&opts.arn).is_some_and(|target| { - target.resync_id == opts.resync_id && target.resync_status == ResyncStatusType::ResyncCanceled - }) - }) - .cloned(); - if let Some(canceled_status) = canceled_status { - save_resync_status(&opts.bucket, &canceled_status, obj_layer).await?; - return Ok(()); + // Persist through the CAS so a stale cached map can never clobber + // states other nodes finalized for other targets; re-run the staleness + // and canceled-is-terminal guards against the freshest persisted entry. + let updated_last_update = updated_target.last_update; + let (final_map, saved) = update_resync_status_cas(&opts.bucket, obj_layer, |persisted| { + if let Some(current) = persisted.targets_map.get(&opts.arn) { + if !resync_state_accepts_update(current, &opts) { + debug!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + incoming_resync_id = %opts.resync_id, + current_resync_id = %current.resync_id, + reason = "stale_status_update", + "Skipped persisting stale resync status update" + ); + return Ok(false); + } + if current.resync_status == ResyncStatusType::ResyncCanceled && status != ResyncStatusType::ResyncCanceled { + debug!( + event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %opts.bucket, + arn = %opts.arn, + incoming_status = %status, + reason = "canceled_status_is_terminal", + "Skipped resync status update after cancellation" + ); + return Ok(false); + } + } + persisted.targets_map.insert(opts.arn.clone(), updated_target.clone()); + persisted.last_update = updated_last_update; + Ok(true) + }) + .await?; + + // Converge this target's cached entry with what the persisted document + // decided (our update, or the newer/terminal state that outranked it). + { + let mut status_map = self.status_map.write().await; + if let Some(cached) = status_map.get_mut(&opts.bucket) + && let Some(final_target) = final_map.targets_map.get(&opts.arn) + { + cached.targets_map.insert(opts.arn.clone(), final_target.clone()); + cached.last_update = final_map.last_update.or(cached.last_update); } } - if let Some(stats) = runtime_sources::replication_stats() { + + if saved && let Some(stats) = runtime_sources::replication_stats() { stats.record_resync_status(&opts.bucket, status, status_duration).await; } @@ -612,10 +647,16 @@ impl ReplicationResyncer { } _ = interval.tick() => { - let status_map = self.status_map.read().await; + let snapshot: Vec<(String, BucketReplicationResyncStatus)> = self + .status_map + .read() + .await + .iter() + .map(|(bucket, status)| (bucket.clone(), status.clone())) + .collect(); let mut update = false; - for (bucket, status) in status_map.iter() { + for (bucket, status) in &snapshot { for target in status.targets_map.values() { if target.last_update.is_none() { update = true; @@ -631,7 +672,14 @@ impl ReplicationResyncer { } if update { - if let Err(err) = save_resync_status(bucket, status, api.clone()).await { + // CAS-merge instead of a blind whole-map save: this + // cache may lag other nodes' admissions and + // cancellations, which must not be overwritten. + let result = update_resync_status_cas(bucket, api.clone(), |persisted| { + Ok(merge_local_resync_into_persisted(persisted, status)) + }) + .await; + if let Err(err) = result { error!( event = EVENT_RESYNC_STATUS_UPDATE_SKIPPED, component = LOG_COMPONENT_ECSTORE, @@ -641,8 +689,8 @@ impl ReplicationResyncer { error = %err, "Failed to persist resync status" ); - } else { - last_update_times.insert(bucket.clone(), status.last_update.expect("last_update should be set")); + } else if let Some(last_update) = status.last_update { + last_update_times.insert(bucket.clone(), last_update); } } } @@ -1451,17 +1499,109 @@ pub async fn get_heal_replicate_object_info(oi: &ObjectInfo, rcfg: &ReplicationC }) } -pub(crate) async fn save_resync_status( +/// Upper bound on optimistic retries for a `resync.bin` compare-and-swap +/// update before giving up; contention on one bucket's status is a handful of +/// writers (status transitions, the periodic saver, admissions), not a crowd. +const RESYNC_STATUS_CAS_MAX_ATTEMPTS: usize = 32; + +/// Read-merge-write `resync.bin` under an ETag compare-and-swap. +/// +/// Every writer used to persist its node's cached whole-bucket map, so one +/// node's stale cache could silently resurrect a state another node had +/// already finalized (e.g. flip a just-canceled intent back to `Pending`). +/// `apply` receives the freshest persisted map and mutates it in place, +/// returning `Ok(false)` to skip the write. On a concurrent write the load + +/// apply + save cycle is retried against the new document. Returns the final +/// map and whether this call wrote it. +pub(crate) async fn update_resync_status_cas( bucket: &str, - status: &BucketReplicationResyncStatus, api: Arc, -) -> Result<()> { - let data = encode_resync_file(status)?; - + mut apply: F, +) -> Result<(BucketReplicationResyncStatus, bool)> +where + S: ReplicationObjectIO, + F: FnMut(&mut BucketReplicationResyncStatus) -> Result, +{ let config_file = ReplicationMetadataStore::bucket_resync_file_path(bucket); - ReplicationConfigStore::save(api, &config_file, data).await?; + for _ in 0..RESYNC_STATUS_CAS_MAX_ATTEMPTS { + let (mut status, preconditions) = + match ReplicationConfigStore::read_no_lock_with_metadata(api.clone(), &config_file).await { + Ok((data, object_info)) => { + let etag = object_info + .etag + .filter(|etag| !etag.trim().is_empty()) + .ok_or_else(|| Error::other("replication resync status has no ETag for conditional update"))?; + let status = if data.is_empty() { + BucketReplicationResyncStatus::new() + } else { + decode_resync_file(&data)? + }; + ( + status, + HTTPPreconditions { + if_match: Some(etag), + ..Default::default() + }, + ) + } + Err(Error::ConfigNotFound) => ( + BucketReplicationResyncStatus::new(), + HTTPPreconditions { + if_none_match: Some("*".to_string()), + ..Default::default() + }, + ), + Err(err) => return Err(err), + }; + if !apply(&mut status)? { + return Ok((status, false)); + } + match ReplicationConfigStore::save_conditional(api.clone(), &config_file, encode_resync_file(&status)?, preconditions) + .await + { + Ok(()) => return Ok((status, true)), + Err(Error::PreconditionFailed) => continue, + Err(err) => return Err(err), + } + } + Err(Error::other("replication resync status conditional update did not converge")) +} - Ok(()) +/// Merge this node's cached bucket resync map into the persisted map for the +/// periodic saver. Per target: same run id overlays the fresher local state +/// unless the persisted state is already terminal and the local one is not +/// (a cancel/completion recorded by another node must stick); a different +/// persisted run id means a newer admission elsewhere and is kept; targets +/// unknown to disk are added. Returns whether `persisted` changed. +pub(crate) fn merge_local_resync_into_persisted( + persisted: &mut BucketReplicationResyncStatus, + local: &BucketReplicationResyncStatus, +) -> bool { + let mut changed = false; + for (arn, local_state) in &local.targets_map { + match persisted.targets_map.get(arn) { + Some(current) if current.resync_id == local_state.resync_id => { + let persisted_terminal = !should_auto_resume_resync(current.resync_status); + let local_terminal = !should_auto_resume_resync(local_state.resync_status); + if persisted_terminal && !local_terminal { + continue; + } + if current != local_state { + persisted.targets_map.insert(arn.clone(), local_state.clone()); + changed = true; + } + } + Some(_) => {} + None => { + persisted.targets_map.insert(arn.clone(), local_state.clone()); + changed = true; + } + } + } + if changed && local.last_update.is_some() { + persisted.last_update = local.last_update; + } + changed } pub async fn replicate_delete(dobj: DeletedObjectReplicationInfo, storage: Arc) { @@ -3913,6 +4053,87 @@ async fn replicate_object_with_multipart(ctx: MultipartR #[cfg(test)] mod tests { use super::super::replication_filemeta_boundary::ReplicateTargetDecision; + fn resync_target_state(resync_id: &str, status: ResyncStatusType, replicated_count: i64) -> TargetReplicationResyncStatus { + TargetReplicationResyncStatus { + resync_id: resync_id.to_string(), + resync_status: status, + replicated_count, + ..Default::default() + } + } + + /// Periodic-saver merge: fresher local progress overlays the same run, + /// but a terminal state persisted by another node must stick, a newer + /// admission elsewhere is kept, and locally-known targets are added. + #[test] + fn merge_local_resync_keeps_peer_terminal_and_newer_states() { + let mut persisted = BucketReplicationResyncStatus::new(); + persisted.targets_map.insert( + "arn:same-run".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncStarted, 1), + ); + persisted.targets_map.insert( + "arn:canceled".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncCanceled, 0), + ); + persisted.targets_map.insert( + "arn:new-run".to_string(), + resync_target_state("run-2", ResyncStatusType::ResyncPending, 0), + ); + + let mut local = BucketReplicationResyncStatus::new(); + local.targets_map.insert( + "arn:same-run".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncStarted, 9), + ); + local.targets_map.insert( + "arn:canceled".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncPending, 0), + ); + local.targets_map.insert( + "arn:new-run".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncStarted, 3), + ); + local.targets_map.insert( + "arn:local-only".to_string(), + resync_target_state("run-1", ResyncStatusType::ResyncPending, 0), + ); + local.last_update = Some(OffsetDateTime::now_utc()); + + assert!(merge_local_resync_into_persisted(&mut persisted, &local)); + assert_eq!(persisted.targets_map["arn:same-run"].replicated_count, 9, "fresher local progress wins"); + assert_eq!( + persisted.targets_map["arn:canceled"].resync_status, + ResyncStatusType::ResyncCanceled, + "peer terminal state must stick" + ); + assert_eq!( + persisted.targets_map["arn:new-run"].resync_id, "run-2", + "newer admission elsewhere is kept" + ); + assert!(persisted.targets_map.contains_key("arn:local-only")); + assert_eq!(persisted.last_update, local.last_update); + } + + /// A terminal local state for the same run (completion/failure recorded by + /// this node) still overlays a non-terminal persisted state. + #[test] + fn merge_local_resync_reports_no_change_when_maps_agree() { + let mut persisted = BucketReplicationResyncStatus::new(); + persisted + .targets_map + .insert("arn:same".to_string(), resync_target_state("run-1", ResyncStatusType::ResyncStarted, 5)); + let local = persisted.clone(); + assert!(!merge_local_resync_into_persisted(&mut persisted, &local)); + + let mut local = local.clone(); + local + .targets_map + .insert("arn:same".to_string(), resync_target_state("run-1", ResyncStatusType::ResyncCompleted, 5)); + assert!(merge_local_resync_into_persisted(&mut persisted, &local)); + assert_eq!(persisted.targets_map["arn:same"].resync_status, ResyncStatusType::ResyncCompleted); + } + use super::super::replication_target_boundary::{BucketTarget, BucketTargets}; use super::*; use s3s::dto::{ diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs index 144181dd2..96d0e6b3f 100644 --- a/crates/replication/src/resync.rs +++ b/crates/replication/src/resync.rs @@ -153,7 +153,7 @@ pub struct ResyncOpts { pub resync_before: Option, } -#[derive(Debug, Clone, Serialize, Deserialize, Default)] +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq)] pub struct TargetReplicationResyncStatus { pub start_time: Option, pub last_update: Option, diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index ea7c990b4..a198329f5 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -16,8 +16,8 @@ use crate::admin::auth::authorize_admin_request; use crate::admin::handlers::site_replication::site_replication_peer_deployment_id_for_endpoint; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::admin::runtime_sources::{ - AppContext, app_context_from_req, current_notification_system_for_context, current_replication_stats_handle_for_context, - current_runtime_port, object_store_from_req, + AppContext, app_context_from_req, current_notification_system_for_context, current_replication_pool_handle, + current_replication_stats_handle_for_context, current_runtime_port, object_store_from_req, }; use crate::admin::storage_api::bucket::metadata::BUCKET_TARGETS_FILE; use crate::admin::storage_api::bucket::metadata_sys; @@ -861,6 +861,8 @@ impl Operation for RemoveRemoteTargetHandler { let targets = sys.remove_target(bucket, arn_str).await.map_err(map_bucket_target_error)?; + cancel_active_resync_intent(bucket, arn_str).await?; + let json_targets = serde_json::to_vec(&targets).map_err(|e| { error!("Serialization error: {}", e); S3Error::with_message(S3ErrorCode::InternalError, "Failed to serialize targets".to_string()) @@ -878,6 +880,35 @@ impl Operation for RemoveRemoteTargetHandler { } } +/// Cancel a pending/started resync intent recorded for `arn` before the target +/// is removed. Without this the intent outlives its target in `resync.bin`, and +/// every later startup reconcile finds an accepted intent with no target to +/// bind it to. The pool reloads and rewrites the status under the bucket +/// admission lock so other nodes' intents are never clobbered. Buckets with no +/// resync history are a no-op. +async fn cancel_active_resync_intent(bucket: &str, arn: &str) -> S3Result<()> { + let Some(pool) = current_replication_pool_handle() else { + return Ok(()); + }; + let canceled = pool + .cancel_bucket_resync_for_removed_target(bucket, arn) + .await + .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to cancel resync: {e}")))?; + if let Some(opts) = canceled { + info!( + event = "replication_resync_intent_canceled", + component = "admin", + subsystem = "replication", + result = "canceled", + bucket, + arn, + resync_id = %opts.resync_id, + "canceled active resync intent for removed remote target" + ); + } + Ok(()) +} + /// Upper bound on the number of object versions scanned per `POST /// /v3/replication/diff` request. RustFS has no persisted per-object /// replication-diff index, so the diff is computed by scanning object versions diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index ba70d152b..cd0e45460 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -748,6 +748,14 @@ impl StorageReplicationPoolHandle { self.inner.clone().cancel_bucket_resync(opts).await } + pub(crate) async fn cancel_bucket_resync_for_removed_target( + &self, + bucket: &str, + arn: &str, + ) -> Result> { + self.inner.clone().cancel_bucket_resync_for_removed_target(bucket, arn).await + } + pub(crate) async fn admit_bucket_resync(&self, opts: ecstore_bucket::replication::ResyncOpts) -> Result { self.inner.clone().admit_bucket_resync(opts).await } @@ -872,10 +880,20 @@ pub(crate) async fn init_background_replication(store: Arc) { ecstore_bucket::replication::init_background_replication(store).await; } +/// Reconcile accepted (pending/started) resync intents into the bucket's +/// target metadata. Returns whether `targets` changed. +/// +/// An intent whose target ARN is no longer configured is an orphan: the +/// remote target was removed after the resync was admitted, or the record +/// predates the atomic-admission contract. Nothing can be reconciled for it, +/// so it is skipped here and left to the resync routine, which marks it +/// `ResyncFailed` through `resolve_resync_target`. Failing startup on it +/// would keep the whole server down over one stale replication record. fn apply_active_resync_intents( + bucket: &str, targets: &mut ecstore_bucket::target::BucketTargets, status: &ecstore_bucket::replication::BucketReplicationResyncStatus, -) -> Result { +) -> bool { let mut changed = false; for (arn, intent) in &status.targets_map { if !matches!( @@ -885,18 +903,26 @@ fn apply_active_resync_intents( ) { 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")))?; + let Some(target) = targets.targets.iter_mut().find(|target| target.arn == *arn) else { + tracing::warn!( + event = "replication_resync_intent_orphaned", + component = "storage", + subsystem = "replication", + result = "skipped", + bucket, + arn = %arn, + resync_status = ?intent.resync_status, + "accepted replication resync target is no longer configured; skipping startup reconcile" + ); + continue; + }; 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) + changed } pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String]) -> Result<()> { @@ -916,7 +942,7 @@ pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String]) - } else { serde_json::from_slice(&metadata.bucket_targets_config_json).map_err(Error::other)? }; - if !apply_active_resync_intents(&mut targets, &status)? { + if !apply_active_resync_intents(bucket, &mut targets, &status) { continue; } let encoded = serde_json::to_vec(&targets).map_err(Error::other)?; @@ -1967,8 +1993,56 @@ mod tests { }, ); - assert!(apply_active_resync_intents(&mut targets, &status).expect("accepted intent should reconcile")); + assert!(apply_active_resync_intents("bucket-a", &mut targets, &status)); assert_eq!(targets.targets[0].reset_id, "durable-id"); assert_eq!(targets.targets[1].reset_id, "concurrent-id"); } + + /// A pending/started intent whose target was removed (or predates the + /// atomic-admission contract) must not abort startup; it is skipped and + /// the remaining intents still reconcile. + #[test] + fn restart_reconcile_skips_orphaned_intent_without_failing_startup() { + let mut targets = ecstore_bucket::target::BucketTargets { + targets: vec![ecstore_bucket::target::BucketTarget { + arn: "arn:minio:replication::depl-1:configured".to_string(), + ..Default::default() + }], + }; + let mut status = ecstore_bucket::replication::BucketReplicationResyncStatus::new(); + for (arn, resync_status) in [ + ( + "arn:rustfs:replication::2ae1d6316a2f17d8:removed", + ecstore_bucket::replication::ResyncStatusType::ResyncStarted, + ), + ( + "arn:minio:replication::depl-1:configured", + ecstore_bucket::replication::ResyncStatusType::ResyncPending, + ), + ] { + status.targets_map.insert( + arn.to_string(), + ecstore_bucket::replication::TargetReplicationResyncStatus { + resync_id: "durable-id".to_string(), + resync_status, + ..Default::default() + }, + ); + } + + assert!(apply_active_resync_intents("bucket-a", &mut targets, &status)); + assert_eq!(targets.targets.len(), 1, "orphaned intent must not materialize a target"); + assert_eq!(targets.targets[0].reset_id, "durable-id"); + + let mut only_orphan = ecstore_bucket::replication::BucketReplicationResyncStatus::new(); + only_orphan.targets_map.insert( + "arn:rustfs:replication::2ae1d6316a2f17d8:removed".to_string(), + ecstore_bucket::replication::TargetReplicationResyncStatus { + resync_id: "durable-id".to_string(), + resync_status: ecstore_bucket::replication::ResyncStatusType::ResyncStarted, + ..Default::default() + }, + ); + assert!(!apply_active_resync_intents("bucket-a", &mut targets, &only_orphan)); + } }