From ea49d5686b04a096ea3d45f70e5adab40d9b336a Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 12:08:22 +0800 Subject: [PATCH] fix(replication): mark failed targets offline (#4405) --- .../ecstore/src/bucket/bucket_target_sys.rs | 25 ++++- .../replication/replication_resyncer.rs | 94 ++++++++++++++++++- .../replication_target_boundary.rs | 4 + 3 files changed, 119 insertions(+), 4 deletions(-) diff --git a/crates/ecstore/src/bucket/bucket_target_sys.rs b/crates/ecstore/src/bucket/bucket_target_sys.rs index ddc082c05..b68f208d4 100644 --- a/crates/ecstore/src/bucket/bucket_target_sys.rs +++ b/crates/ecstore/src/bucket/bucket_target_sys.rs @@ -1759,7 +1759,30 @@ impl TargetClient { .await { Ok(_res) => Ok(()), - Err(e) => Err(e.into()), + Err(e) => match e { + SdkError::ServiceError(service_err) => { + let err = service_err.into_err(); + let meta = err.meta(); + let error = match (meta.code(), meta.message()) { + (Some(code), Some(message)) => format!("remove_object failed: {code}: {message}"), + (Some(code), None) => format!("remove_object failed: {code}"), + (None, Some(message)) => format!("remove_object failed: {message}"), + (None, None) => format!("remove_object failed: {err:?}"), + }; + Err(S3ClientError::with_metadata( + error, + None, + meta.code().map(ToOwned::to_owned), + meta.message().map(ToOwned::to_owned), + )) + } + SdkError::DispatchFailure(dispatch_err) => Err(S3ClientError::new(format!( + "remove_object dispatch failure for bucket:{bucket} object:{object}: {dispatch_err:?}" + ))), + other => Err(S3ClientError::new(format!( + "remove_object request failed for bucket:{bucket} object:{object}: {other:?}" + ))), + }, } } } diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 3039d639e..fba2fba4e 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -73,6 +73,7 @@ use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash}; #[cfg(test)] use s3s::dto::ReplicationConfiguration; use std::collections::HashMap; +use std::fmt::Display; use std::sync::Arc; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; @@ -94,6 +95,21 @@ const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; const ERR_REPLICATION_METADATA_COPY_UNSUPPORTED: &str = "metadata-only replication is not implemented"; +const REPLICATION_TARGET_OFFLINE_ERROR_MARKERS: &[&str] = &[ + "dispatch failure", + "timeouterror", + "timed out", + "connection refused", + "connection reset", + "connection closed", + "connection aborted", + "broken pipe", + "dns error", + "failed to lookup address", + "name or service not known", + "deadline has elapsed", + "tcp connect error", +]; const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60); @@ -171,6 +187,19 @@ fn is_version_id_format_mismatch(err: &SdkError) -> bool { is_version_id_mismatch(code, raw_status) } +fn is_replication_target_offline_error(err: &(impl Display + ?Sized)) -> bool { + let message = err.to_string().to_ascii_lowercase(); + REPLICATION_TARGET_OFFLINE_ERROR_MARKERS + .iter() + .any(|marker| message.contains(marker)) +} + +async fn mark_replication_target_offline_if_needed(target_client: &TargetClient, err: &(impl Display + ?Sized)) { + if is_replication_target_offline_error(err) { + ReplicationTargetStore::mark_target_offline(target_client).await; + } +} + async fn head_object_fallback( source_bucket: &str, tgt_client: &TargetClient, @@ -1845,7 +1874,7 @@ async fn replicate_delete_to_target(dobj: &DeletedObjectReplicationInfo, tgt_cli } else { rinfo.version_purge_status = VersionPurgeStatusType::Failed; } - // TODO: check offline + mark_replication_target_offline_if_needed(&tgt_client, &e).await; } } @@ -2406,7 +2435,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { "Replication target operation failed" ); - // TODO: check offline + mark_replication_target_offline_if_needed(&tgt_client, &err).await; return rinfo; } @@ -2835,7 +2864,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { ); rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); - // TODO: check offline + mark_replication_target_offline_if_needed(&tgt_client, &err).await; return rinfo; } } @@ -3018,6 +3047,65 @@ mod tests { use time::OffsetDateTime; use uuid::Uuid; + fn test_target_client(endpoint: String) -> TargetClient { + let config = aws_sdk_s3::Config::builder() + .endpoint_url(endpoint.clone()) + .region(aws_sdk_s3::config::Region::new("us-east-1")) + .credentials_provider(aws_sdk_s3::config::SharedCredentialsProvider::new( + aws_credential_types::Credentials::new("access", "secret", None, None, "test"), + )) + .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest()) + .build(); + + TargetClient { + endpoint, + credentials: None, + bucket: "target-bucket".to_string(), + storage_class: String::new(), + disable_proxy: false, + arn: format!("arn:rustfs:replication:us-east-1:target:{}", Uuid::new_v4()), + reset_id: String::new(), + secure: false, + health_check_duration: std::time::Duration::from_secs(5), + replicate_sync: false, + client: Arc::new(aws_sdk_s3::Client::from_conf(config)), + } + } + + #[test] + fn replication_target_offline_error_classifier_is_network_scoped() { + assert!(is_replication_target_offline_error(&"put_object dispatch failure: connector error")); + assert!(is_replication_target_offline_error(&"request TimeoutError after retry")); + assert!(is_replication_target_offline_error(&"tcp connect error: connection refused")); + assert!(!is_replication_target_offline_error(&"put_object failed: AccessDenied: denied")); + assert!(!is_replication_target_offline_error(&"put_object failed: NoSuchBucket")); + } + + #[tokio::test] + async fn replication_target_network_failure_marks_target_offline() { + let endpoint = format!("http://network-failure-{}.example:9000", Uuid::new_v4()); + let target_client = test_target_client(endpoint); + + assert!(!ReplicationTargetStore::target_is_offline(&target_client).await); + + let err = std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "connection refused"); + mark_replication_target_offline_if_needed(&target_client, &err).await; + + assert!(ReplicationTargetStore::target_is_offline(&target_client).await); + } + + #[tokio::test] + async fn replication_target_service_failure_keeps_target_online() { + let endpoint = format!("http://service-failure-{}.example:9000", Uuid::new_v4()); + let target_client = test_target_client(endpoint); + + assert!(!ReplicationTargetStore::target_is_offline(&target_client).await); + + mark_replication_target_offline_if_needed(&target_client, &"put_object failed: AccessDenied: denied").await; + + assert!(!ReplicationTargetStore::target_is_offline(&target_client).await); + } + #[test] fn test_unmarshal_resync_payload() { let start = OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("valid ts"); diff --git a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs index 6c7ced0b0..cb6885504 100644 --- a/crates/ecstore/src/bucket/replication/replication_target_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_target_boundary.rs @@ -92,6 +92,10 @@ impl ReplicationTargetStore { pub(crate) async fn target_is_offline(target_client: &TargetClient) -> bool { BucketTargetSys::get().is_offline(&target_client.to_url()).await } + + pub(crate) async fn mark_target_offline(target_client: &TargetClient) { + BucketTargetSys::get().mark_offline(&target_client.to_url()).await + } } pub(crate) fn replication_put_object_options(sc: &str, object_info: &ObjectInfo) -> Result<(PutObjectOptions, bool)> {