fix(replication): prevent target state loss across buckets (#2704)

crates/e2e_test/src/replication_extension_test.rs

@@ -21,6 +21,7 @@ use aws_sdk_s3::types::{BucketVersioningStatus, VersioningConfiguration};
 use aws_sdk_s3::{Client, Config};
 use http::header::{CONTENT_TYPE, HOST};
 use reqwest::StatusCode;
+use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys;
 use rustfs_madmin::{
     AddServiceAccountReq, ListServiceAccountsResp, PeerInfo, PeerSite, ReplicateAddStatus, ReplicateEditStatus,
     ReplicateRemoveStatus, SRRemoveReq, SRResyncOpStatus, SRStatusInfo, SiteReplicationInfo, SyncStatus,
@@ -1648,6 +1649,90 @@ async fn test_single_bucket_replication_fans_out_to_multiple_targets() -> Result
     Ok(())
 }
 
+#[tokio::test]
+#[serial]
+async fn test_sequential_bucket_replication_succeeds_for_multiple_buckets() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env.start_rustfs_server(vec![]).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    for idx in 1..=5 {
+        let source_bucket = format!("replication-multi-src-{idx}");
+        let target_bucket = format!("replication-multi-dst-{idx}");
+        let object_key = format!("probe-{idx}.txt");
+        let body = format!("payload-{idx}");
+
+        source_client.create_bucket().bucket(&source_bucket).send().await?;
+        target_client.create_bucket().bucket(&target_bucket).send().await?;
+        enable_bucket_versioning(&source_env, &source_bucket).await?;
+        enable_bucket_versioning(&target_env, &target_bucket).await?;
+
+        let target_arn = set_replication_target(&source_env, &source_bucket, &target_env, &target_bucket).await?;
+        put_bucket_replication(&source_env, &source_bucket, &target_arn).await?;
+
+        source_client
+            .put_object()
+            .bucket(&source_bucket)
+            .key(&object_key)
+            .body(ByteStream::from(body.clone().into_bytes()))
+            .send()
+            .await?;
+
+        wait_for_replicated_object(&target_client, &target_bucket, &object_key, &body).await?;
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+#[serial]
+async fn test_replication_recovers_after_runtime_target_cache_is_cleared() -> Result<(), Box<dyn Error + Send + Sync>> {
+    init_logging();
+
+    let mut source_env = RustFSTestEnvironment::new().await?;
+    source_env.start_rustfs_server(vec![]).await?;
+
+    let mut target_env = RustFSTestEnvironment::new().await?;
+    target_env.start_rustfs_server_without_cleanup(vec![]).await?;
+
+    let source_bucket = "replication-refresh-src";
+    let target_bucket = "replication-refresh-dst";
+    let object_key = "probe-refresh.txt";
+    let body = "payload-refresh";
+
+    let source_client = source_env.create_s3_client();
+    let target_client = target_env.create_s3_client();
+
+    source_client.create_bucket().bucket(source_bucket).send().await?;
+    target_client.create_bucket().bucket(target_bucket).send().await?;
+    enable_bucket_versioning(&source_env, source_bucket).await?;
+    enable_bucket_versioning(&target_env, target_bucket).await?;
+
+    let target_arn = set_replication_target(&source_env, source_bucket, &target_env, target_bucket).await?;
+    put_bucket_replication(&source_env, source_bucket, &target_arn).await?;
+
+    BucketTargetSys::get().delete(source_bucket).await;
+
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(object_key)
+        .body(ByteStream::from(body.as_bytes().to_vec()))
+        .send()
+        .await?;
+
+    wait_for_replicated_object(&target_client, target_bucket, object_key, body).await?;
+
+    Ok(())
+}
+
 #[tokio::test]
 #[serial]
 async fn test_site_replication_resync_start_cancel_restart_real_dual_node() -> Result<(), Box<dyn Error + Send + Sync>> {

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -395,8 +395,27 @@ impl BucketTargetSys {
         }
     }
 
-    pub async fn set_target(&self, bucket: &str, target: &BucketTarget, update: bool) -> Result<(), BucketTargetError> {
-        if !target.target_type.is_valid() && !update {
+    pub async fn set_target(
+        &self,
+        bucket: &str,
+        target: &BucketTarget,
+        update: bool,
+    ) -> Result<BucketTargets, BucketTargetError> {
+        self.validate_target(bucket, target).await?;
+
+        let mut bucket_targets = match self.list_bucket_targets(bucket).await {
+            Ok(targets) => targets,
+            Err(BucketTargetError::BucketRemoteTargetNotFound { .. }) => BucketTargets::default(),
+            Err(err) => return Err(err),
+        };
+
+        Self::upsert_target_entry(&mut bucket_targets.targets, target, update)?;
+
+        Ok(bucket_targets)
+    }
+
+    pub async fn validate_target(&self, bucket: &str, target: &BucketTarget) -> Result<(), BucketTargetError> {
+        if !target.target_type.is_valid() {
             return Err(BucketTargetError::BucketRemoteArnTypeInvalid {
                 bucket: bucket.to_string(),
             });
@@ -450,52 +469,44 @@ impl BucketTargetSys {
             }
         }
 
-        {
-            let mut targets_map = self.targets_map.write().await;
-            let bucket_targets = targets_map.entry(bucket.to_string()).or_insert_with(Vec::new);
-            let mut found = false;
+        Ok(())
+    }
 
-            for (idx, existing_target) in bucket_targets.iter().enumerate() {
-                if existing_target.target_type.to_string() == target.target_type.to_string() {
-                    if existing_target.arn == target.arn {
-                        if !update {
-                            return Err(BucketTargetError::BucketRemoteAlreadyExists {
-                                bucket: existing_target.target_bucket.clone(),
-                            });
-                        }
-                        bucket_targets[idx] = target.clone();
-                        found = true;
-                        break;
-                    }
-                    if existing_target.endpoint == target.endpoint {
+    fn upsert_target_entry(
+        bucket_targets: &mut Vec<BucketTarget>,
+        target: &BucketTarget,
+        update: bool,
+    ) -> Result<(), BucketTargetError> {
+        let mut found = false;
+
+        for (idx, existing_target) in bucket_targets.iter().enumerate() {
+            if existing_target.target_type.to_string() == target.target_type.to_string() {
+                if existing_target.arn == target.arn {
+                    if !update {
                         return Err(BucketTargetError::BucketRemoteAlreadyExists {
                             bucket: existing_target.target_bucket.clone(),
                         });
                     }
+                    bucket_targets[idx] = target.clone();
+                    found = true;
+                    break;
+                }
+                if existing_target.endpoint == target.endpoint {
+                    return Err(BucketTargetError::BucketRemoteAlreadyExists {
+                        bucket: existing_target.target_bucket.clone(),
+                    });
                 }
             }
-
-            if !found && !update {
-                bucket_targets.push(target.clone());
-            }
         }
 
-        {
-            let mut arn_remotes_map = self.arn_remotes_map.write().await;
-            arn_remotes_map.insert(
-                target.arn.clone(),
-                ArnTarget {
-                    client: Some(Arc::new(target_client)),
-                    last_refresh: OffsetDateTime::now_utc(),
-                },
-            );
+        if !found && !update {
+            bucket_targets.push(target.clone());
         }
 
-        self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
         Ok(())
     }
 
-    pub async fn remove_target(&self, bucket: &str, arn_str: &str) -> Result<(), BucketTargetError> {
+    pub async fn remove_target(&self, bucket: &str, arn_str: &str) -> Result<BucketTargets, BucketTargetError> {
         if arn_str.is_empty() {
             return Err(BucketTargetError::BucketRemoteArnInvalid {
                 bucket: bucket.to_string(),
@@ -524,33 +535,16 @@ impl BucketTargetSys {
             }
         }
 
-        {
-            let mut targets_map = self.targets_map.write().await;
+        let targets = self.list_bucket_targets(bucket).await?;
+        let new_targets: Vec<BucketTarget> = targets.targets.iter().filter(|t| t.arn != arn_str).cloned().collect();
 
-            let Some(targets) = targets_map.get(bucket) else {
-                return Err(BucketTargetError::BucketRemoteTargetNotFound {
-                    bucket: bucket.to_string(),
-                });
-            };
-
-            let new_targets: Vec<BucketTarget> = targets.iter().filter(|t| t.arn != arn_str).cloned().collect();
-
-            if new_targets.len() == targets.len() {
-                return Err(BucketTargetError::BucketRemoteTargetNotFound {
-                    bucket: bucket.to_string(),
-                });
-            }
-
-            targets_map.insert(bucket.to_string(), new_targets);
+        if new_targets.len() == targets.targets.len() {
+            return Err(BucketTargetError::BucketRemoteTargetNotFound {
+                bucket: bucket.to_string(),
+            });
         }
 
-        {
-            self.arn_remotes_map.write().await.remove(arn_str);
-        }
-
-        self.update_bandwidth_limit(bucket, arn_str, 0);
-
-        Ok(())
+        Ok(BucketTargets { targets: new_targets })
     }
 
     pub async fn mark_refresh_in_progress(&self, bucket: &str, arn: &str) {
@@ -603,7 +597,7 @@ impl BucketTargetSys {
 
         if let Some(last_refresh) = last_refresh {
             let now = OffsetDateTime::now_utc();
-            if now - last_refresh > Duration::from_secs(60 * 5) {
+            if now - last_refresh < Duration::from_secs(60 * 5) {
                 return None;
             }
         }
@@ -619,6 +613,16 @@ impl BucketTargetSys {
             }
         };
 
+        let cli = self
+            .arn_remotes_map
+            .read()
+            .await
+            .get(arn)
+            .and_then(|target| target.client.clone());
+        if cli.is_some() {
+            return cli;
+        }
+
         self.inc_arn_errs(bucket, arn).await;
         None
     }

rustfs/src/admin/handlers/replication.rs

@@ -288,15 +288,10 @@ impl Operation for SetRemoteTargetHandler {
 
         let arn = remote_target.arn.clone();
 
-        bucket_target_sys
+        let targets = bucket_target_sys
             .set_target(bucket, &remote_target, update)
             .await
             .map_err(map_bucket_target_error)?;
-
-        let targets = bucket_target_sys.list_bucket_targets(bucket).await.map_err(|e| {
-            error!("Failed to list bucket targets: {}", e);
-            S3Error::with_message(S3ErrorCode::InternalError, "Failed to list bucket targets".to_string())
-        })?;
         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())
@@ -308,6 +303,7 @@ impl Operation for SetRemoteTargetHandler {
                 error!("Failed to update bucket targets: {}", e);
                 S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to update bucket targets: {e}"))
             })?;
+        bucket_target_sys.update_all_targets(bucket, Some(&targets)).await;
 
         let arn_str = serde_json::to_string(&arn).unwrap_or_default();
 
@@ -405,12 +401,7 @@ impl Operation for RemoveRemoteTargetHandler {
 
         let sys = BucketTargetSys::get();
 
-        sys.remove_target(bucket, arn_str).await.map_err(map_bucket_target_error)?;
-
-        let targets = sys.list_bucket_targets(bucket).await.map_err(|e| {
-            error!("Failed to list bucket targets: {}", e);
-            S3Error::with_message(S3ErrorCode::InternalError, "Failed to list bucket targets".to_string())
-        })?;
+        let targets = sys.remove_target(bucket, arn_str).await.map_err(map_bucket_target_error)?;
 
         let json_targets = serde_json::to_vec(&targets).map_err(|e| {
             error!("Serialization error: {}", e);
@@ -423,6 +414,7 @@ impl Operation for RemoveRemoteTargetHandler {
                 error!("Failed to update bucket targets: {}", e);
                 S3Error::with_message(S3ErrorCode::InternalError, format!("Failed to update bucket targets: {e}"))
             })?;
+        sys.update_all_targets(bucket, Some(&targets)).await;
 
         Ok(S3Response::new((StatusCode::NO_CONTENT, Body::from("".to_string()))))
     }

rustfs/src/app/bucket_usecase.rs

@@ -46,7 +46,7 @@ use rustfs_ecstore::bucket::{
     metadata_sys,
     object_lock::ObjectLockApi,
     policy_sys::PolicySys,
-    target::BucketTargetType,
+    target::{BucketTargetType, BucketTargets},
     utils::serialize,
     versioning::VersioningApi,
     versioning_sys::BucketVersioningSys,
@@ -119,6 +119,59 @@ fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet<String>
     arns
 }
 
+fn validate_replication_config_targets(targets: &BucketTargets, config: &ReplicationConfiguration) -> S3Result<()> {
+    let configured_arns = targets
+        .targets
+        .iter()
+        .filter(|target| target.target_type == BucketTargetType::ReplicationService)
+        .map(|target| target.arn.as_str())
+        .collect::<HashSet<_>>();
+
+    for rule in &config.rules {
+        if rule.status == ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED) {
+            continue;
+        }
+
+        let configured_arn = if config.role.trim().is_empty() {
+            rule.destination.bucket.trim()
+        } else {
+            config.role.trim()
+        };
+
+        if !configured_arn.is_empty() && configured_arns.contains(configured_arn) {
+            continue;
+        }
+
+        return Err(s3_error!(
+            InvalidRequest,
+            "replication config with rule ID {} has a stale target",
+            rule.id.clone().unwrap_or_default()
+        ));
+    }
+
+    Ok(())
+}
+
+async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> {
+    if !BucketVersioningSys::enabled(bucket).await {
+        return Err(s3_error!(
+            InvalidRequest,
+            "bucket versioning must be enabled before replication can be configured"
+        ));
+    }
+
+    let targets = metadata_sys::get_bucket_targets_config(bucket)
+        .await
+        .map_err(|err| match err {
+            StorageError::ConfigNotFound => {
+                S3Error::with_message(S3ErrorCode::InvalidRequest, "replication target configuration not found".to_string())
+            }
+            other => ApiError::from(other).into(),
+        })?;
+
+    validate_replication_config_targets(&targets, config)
+}
+
 async fn remove_replication_targets_for_config(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> {
     let target_arns = replication_target_arns(config);
     if target_arns.is_empty() {
@@ -1615,7 +1668,7 @@ impl DefaultBucketUsecase {
             .await
             .map_err(ApiError::from)?;
 
-        // TODO: check enable, versioning enable
+        validate_bucket_replication_update(&bucket, &replication_configuration).await?;
         let data = serialize_config(&replication_configuration)?;
         metadata_sys::update(&bucket, BUCKET_REPLICATION_CONFIG, data)
             .await
@@ -1979,6 +2032,70 @@ mod tests {
         assert!(arns.contains(destination));
     }
 
+    fn replication_targets_with_arn(arns: &[&str]) -> BucketTargets {
+        BucketTargets {
+            targets: arns
+                .iter()
+                .map(|arn| rustfs_ecstore::bucket::target::BucketTarget {
+                    arn: (*arn).to_string(),
+                    target_type: BucketTargetType::ReplicationService,
+                    ..Default::default()
+                })
+                .collect(),
+        }
+    }
+
+    #[test]
+    fn validate_replication_config_targets_accepts_matching_destination_arns() {
+        let arn = "arn:rustfs:replication:us-east-1:target:bucket";
+        let targets = replication_targets_with_arn(&[arn]);
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(arn)],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("matching target should pass validation");
+    }
+
+    #[test]
+    fn validate_replication_config_targets_rejects_stale_destination_arns() {
+        let targets = replication_targets_with_arn(&["arn:rustfs:replication:us-east-1:target:bucket-a"]);
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![replication_rule_for_target(
+                "arn:rustfs:replication:us-east-1:target:bucket-b",
+            )],
+        };
+
+        let err = validate_replication_config_targets(&targets, &config).expect_err("stale target should fail validation");
+        assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
+    }
+
+    #[test]
+    fn validate_replication_config_targets_accepts_matching_role_arn() {
+        let arn = "arn:rustfs:replication:us-east-1:role-target:bucket";
+        let targets = replication_targets_with_arn(&[arn]);
+        let config = ReplicationConfiguration {
+            role: arn.to_string(),
+            rules: vec![replication_rule_for_target("arn:rustfs:replication:us-east-1:ignored:bucket")],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("matching role ARN should pass validation");
+    }
+
+    #[test]
+    fn validate_replication_config_targets_ignores_disabled_rules() {
+        let targets = replication_targets_with_arn(&[]);
+        let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:stale:bucket");
+        rule.status = ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED);
+        let config = ReplicationConfiguration {
+            role: String::new(),
+            rules: vec![rule],
+        };
+
+        validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets");
+    }
+
     #[test]
     fn versioning_configuration_has_object_lock_incompatible_settings_rejects_suspended() {
         let config = VersioningConfiguration {