diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index 860e31fb0..df2b50143 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/crates/ecstore/src/api/mod.rs @@ -176,8 +176,8 @@ pub mod bucket { 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, 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, + should_use_existing_delete_replication_source, unsupported_replication_config_field, + validate_replication_config_target_arns, version_purge_status_to_filemeta, }; } diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index e81f383ce..eaa2e6ac4 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -46,7 +46,7 @@ mod runtime_boundary; pub use datatypes::ResyncStatusType; pub use replication_config_boundary::{ ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns, - should_remove_replication_target, validate_replication_config_target_arns, + should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; #[cfg(test)] pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision; diff --git a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs index da3e52a9d..e0cda9527 100644 --- a/crates/ecstore/src/bucket/replication/replication_config_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_config_boundary.rs @@ -14,5 +14,5 @@ pub use rustfs_replication::{ ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, replication_target_arns, - should_remove_replication_target, validate_replication_config_target_arns, + should_remove_replication_target, unsupported_replication_config_field, validate_replication_config_target_arns, }; diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 95ea82d6d..85c5f959b 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -14,8 +14,11 @@ use std::{collections::HashMap, sync::Arc}; +use super::replication_error_boundary::Result; use super::replication_filemeta_boundary::{ReplicateDecision, ReplicationStatusType, ReplicationType}; -use super::replication_object_config::{check_replicate_delete, get_must_replicate_options, must_replicate}; +use super::replication_object_config::{ + check_replicate_delete, check_replicate_delete_strict, get_must_replicate_options, must_replicate, +}; use super::replication_object_decision_boundary::MustReplicateOptions; use super::replication_pool::{schedule_replication, schedule_replication_delete}; use super::replication_queue_boundary::DeletedObjectReplicationInfo; @@ -50,6 +53,16 @@ impl ReplicationObjectBridge { check_replicate_delete(bucket, object, source, opts, get_error).await } + pub async fn check_delete_strict( + bucket: &str, + object: &ObjectToDelete, + source: &ObjectInfo, + opts: &ObjectOptions, + get_error: Option, + ) -> Result { + check_replicate_delete_strict(bucket, object, source, opts, get_error).await + } + pub async fn schedule_object( object: ObjectInfo, storage: Arc, diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs index 2d23466d3..ae69eac4e 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_config.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -169,9 +169,8 @@ pub(crate) async fn check_replicate_delete( del_opts: &ObjectOptions, gerr: Option, ) -> ReplicateDecision { - let rcfg = match get_replication_config(bucket).await { - Ok(Some(config)) => config, - Ok(None) => return ReplicateDecision::default(), + match check_replicate_delete_strict(bucket, dobj, oi, del_opts, gerr).await { + Ok(decision) => decision, Err(err) => { error!( event = EVENT_RESYNC_CONFIG_LOOKUP_SKIPPED, @@ -182,16 +181,30 @@ pub(crate) async fn check_replicate_delete( error = %err, "Failed to look up replication config for delete replication" ); - return ReplicateDecision::default(); + ReplicateDecision::default() } + } +} + +pub(crate) async fn check_replicate_delete_strict( + bucket: &str, + dobj: &ObjectToDelete, + oi: &ObjectInfo, + del_opts: &ObjectOptions, + gerr: Option, +) -> Result { + let rcfg = match get_replication_config(bucket).await { + Ok(Some(config)) => config, + Ok(None) => return Ok(ReplicateDecision::default()), + Err(err) => return Err(err), }; if del_opts.replication_request { - return ReplicateDecision::default(); + return Ok(ReplicateDecision::default()); } - if !del_opts.versioned { - return ReplicateDecision::default(); + if !del_opts.versioned && !del_opts.version_suspended { + return Ok(ReplicateDecision::default()); } let replication_delete = object_to_delete_for_replication(dobj); @@ -209,7 +222,7 @@ pub(crate) async fn check_replicate_delete( let mut dsc = ReplicateDecision::new(); if tgt_arns.is_empty() { - return dsc; + return Ok(dsc); } for tgt_arn in tgt_arns { @@ -239,7 +252,7 @@ pub(crate) async fn check_replicate_delete( dsc.set(tgt_dsc); } - dsc + Ok(dsc) } pub(crate) async fn must_replicate(bucket: &str, object: &str, mopts: MustReplicateOptions) -> ReplicateDecision { diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index e8a2f3472..2429cdc1e 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -1151,60 +1151,6 @@ pub async fn replicate_delete(dobj: DeletedObjectReplicat dobj.delete_object.version_id }; - let _rcfg = match get_replication_config(&bucket).await { - Ok(Some(config)) => config, - Ok(None) => { - debug!( - event = EVENT_REPLICATION_DELETE_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - reason = "replication_config_missing", - "Skipping replication delete because replication config is missing" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: ObjectInfo { - bucket: bucket.clone(), - name: dobj.delete_object.object_name.clone(), - version_id, - delete_marker: dobj.delete_object.delete_marker, - ..Default::default() - }, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - - return; - } - Err(err) => { - debug!( - event = EVENT_REPLICATION_DELETE_SKIPPED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - error = %err, - reason = "replication_config_lookup_failed", - "Skipping replication delete because replication config lookup failed" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: ObjectInfo { - bucket: bucket.clone(), - name: dobj.delete_object.object_name.clone(), - version_id, - delete_marker: dobj.delete_object.delete_marker, - ..Default::default() - }, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - return; - } - }; - if dobj.delete_object.delete_marker && let Some(delete_marker_version_id) = dobj.delete_object.delete_marker_version_id { diff --git a/crates/replication/src/config.rs b/crates/replication/src/config.rs index 9dea6d7eb..60fa9c235 100644 --- a/crates/replication/src/config.rs +++ b/crates/replication/src/config.rs @@ -51,6 +51,28 @@ pub enum ReplicationTargetValidationError { StaleTarget, } +pub fn unsupported_replication_config_field(config: &ReplicationConfiguration) -> Option<&'static str> { + for rule in &config.rules { + if rule + .source_selection_criteria + .as_ref() + .is_some_and(|criteria| criteria.sse_kms_encrypted_objects.is_some()) + { + return Some("SourceSelectionCriteria.SseKmsEncryptedObjects"); + } + if rule.destination.encryption_configuration.is_some() { + return Some("Destination.EncryptionConfiguration"); + } + if rule.destination.metrics.is_some() { + return Some("Destination.Metrics"); + } + if rule.destination.replication_time.is_some() { + return Some("Destination.ReplicationTime"); + } + } + None +} + pub fn active_replication_rule_destination_arns(config: &ReplicationConfiguration) -> HashSet { let mut arns = HashSet::new(); @@ -164,7 +186,17 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { if let Some(filter) = &rule.filter { let object_tags = ReplicationTagFilter::decode_tags_to_map(&obj.user_tags); - if filter.test_tags(&object_tags) { + let tags_match = if let Some(and) = &filter.and { + and.tags.as_ref().is_none_or(|tags| { + tags.iter() + .all(|tag| tag.key.as_ref().is_some_and(|key| object_tags.get(key) == tag.value.as_ref())) + }) + } else if let Some(tag) = &filter.tag { + tag.key.as_ref().is_some_and(|key| object_tags.get(key) == tag.value.as_ref()) + } else { + true + }; + if tags_match { rules.push(rule.clone()); } } else { @@ -217,11 +249,6 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { } if obj.version_id.is_some() { - if obj.delete_marker { - return rule.delete_marker_replication.clone().is_some_and(|d| { - d.status == Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)) - }); - } return rule .delete_replication .clone() @@ -305,7 +332,12 @@ impl ReplicationConfigurationExt for ReplicationConfiguration { #[cfg(test)] mod tests { use super::*; - use s3s::dto::{DeleteMarkerReplication, Destination, ExistingObjectReplication, ReplicationRule}; + use s3s::dto::{ + DeleteMarkerReplication, DeleteReplication, Destination, EncryptionConfiguration, ExistingObjectReplication, Metrics, + MetricsStatus, ReplicationRule, ReplicationRuleAndOperator, ReplicationRuleFilter, ReplicationTime, + ReplicationTimeStatus, ReplicationTimeValue, SourceSelectionCriteria, SseKmsEncryptedObjects, + SseKmsEncryptedObjectsStatus, Tag, + }; fn replication_rule(id: &str, arn: &str) -> ReplicationRule { ReplicationRule { @@ -605,4 +637,162 @@ mod tests { "highest-priority rule disables delete-marker replication, so the delete marker must not replicate" ); } + + #[test] + fn replication_filter_requires_every_and_tag_and_accepts_extra_tags() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let mut rule = replication_rule("and-tags", arn); + rule.filter = Some(ReplicationRuleFilter { + and: Some(ReplicationRuleAndOperator { + prefix: Some("logs/".to_string()), + tags: Some(vec![ + Tag { + key: Some("env".to_string()), + value: Some("prod".to_string()), + }, + Tag { + key: Some("team".to_string()), + value: Some("storage/core".to_string()), + }, + ]), + }), + ..Default::default() + }); + rule.prefix = None; + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![rule], + }; + + let matching = ObjectOpts { + name: "logs/app.log".to_string(), + user_tags: "env=prod&team=storage%2Fcore&extra=value".to_string(), + op_type: ReplicationType::Object, + ..Default::default() + }; + let partial = ObjectOpts { + user_tags: "env=prod".to_string(), + ..matching.clone() + }; + let wrong_prefix = ObjectOpts { + name: "archive/app.log".to_string(), + ..matching.clone() + }; + + for (op_type, existing_object) in [ + (ReplicationType::Object, false), + (ReplicationType::Delete, false), + (ReplicationType::ExistingObject, true), + (ReplicationType::Heal, false), + ] { + let mut matching = matching.clone(); + matching.op_type = op_type; + matching.existing_object = existing_object; + let mut partial = partial.clone(); + partial.op_type = op_type; + partial.existing_object = existing_object; + let mut wrong_prefix = wrong_prefix.clone(); + wrong_prefix.op_type = op_type; + wrong_prefix.existing_object = existing_object; + + assert_eq!(config.filter_actionable_rules(&matching).len(), 1); + assert!(config.filter_actionable_rules(&partial).is_empty()); + assert!(config.filter_actionable_rules(&wrong_prefix).is_empty()); + } + } + + #[test] + fn version_purge_uses_delete_replication_for_object_and_marker_versions() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let mut rule = replication_rule("delete", arn); + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::DISABLED)), + }); + rule.delete_replication = Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED), + }); + let mut config = ReplicationConfiguration { + role: String::new(), + rules: vec![rule], + }; + let version_id = Some(Uuid::new_v4()); + + for delete_marker in [false, true] { + assert!(config.replicate(&ObjectOpts { + name: "object".to_string(), + version_id, + delete_marker, + op_type: ReplicationType::Delete, + ..Default::default() + })); + } + assert!(!config.replicate(&ObjectOpts { + name: "object".to_string(), + delete_marker: true, + op_type: ReplicationType::Delete, + ..Default::default() + })); + + let rule = &mut config.rules[0]; + rule.delete_marker_replication = Some(DeleteMarkerReplication { + status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), + }); + rule.delete_replication = Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::DISABLED), + }); + + for delete_marker in [false, true] { + assert!(!config.replicate(&ObjectOpts { + name: "object".to_string(), + version_id, + delete_marker, + op_type: ReplicationType::Delete, + ..Default::default() + })); + } + assert!(config.replicate(&ObjectOpts { + name: "object".to_string(), + delete_marker: true, + op_type: ReplicationType::Delete, + ..Default::default() + })); + } + + #[test] + fn unsupported_replication_fields_are_reported_before_persistence() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let mut config = ReplicationConfiguration { + role: String::new(), + rules: vec![replication_rule("unsupported", arn)], + }; + + config.rules[0].source_selection_criteria = Some(SourceSelectionCriteria { + replica_modifications: None, + sse_kms_encrypted_objects: Some(SseKmsEncryptedObjects { + status: SseKmsEncryptedObjectsStatus::from_static(SseKmsEncryptedObjectsStatus::ENABLED), + }), + }); + assert_eq!( + unsupported_replication_config_field(&config), + Some("SourceSelectionCriteria.SseKmsEncryptedObjects") + ); + + config.rules[0].source_selection_criteria = None; + config.rules[0].destination.encryption_configuration = Some(EncryptionConfiguration::default()); + assert_eq!(unsupported_replication_config_field(&config), Some("Destination.EncryptionConfiguration")); + + config.rules[0].destination.encryption_configuration = None; + config.rules[0].destination.metrics = Some(Metrics { + event_threshold: None, + status: MetricsStatus::from_static(MetricsStatus::ENABLED), + }); + assert_eq!(unsupported_replication_config_field(&config), Some("Destination.Metrics")); + + config.rules[0].destination.metrics = None; + config.rules[0].destination.replication_time = Some(ReplicationTime { + status: ReplicationTimeStatus::from_static(ReplicationTimeStatus::ENABLED), + time: ReplicationTimeValue { minutes: Some(15) }, + }); + assert_eq!(unsupported_replication_config_field(&config), Some("Destination.ReplicationTime")); + } } diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index e6759cc49..14e0f9ab3 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -30,7 +30,8 @@ pub mod tagging; pub use config::{ ObjectOpts, ReplicationConfigurationExt, ReplicationTargetValidationError, active_replication_rule_destination_arns, - replication_target_arns, should_remove_replication_target, validate_replication_config_target_arns, + replication_target_arns, should_remove_replication_target, unsupported_replication_config_field, + validate_replication_config_target_arns, }; pub use delete::{ DeletedObjectReplicationInfo, is_retryable_delete_replication_head_error, is_version_delete_replication, diff --git a/crates/replication/src/operation.rs b/crates/replication/src/operation.rs index d587b1bee..f1ab3f069 100644 --- a/crates/replication/src/operation.rs +++ b/crates/replication/src/operation.rs @@ -322,9 +322,9 @@ mod tests { use crate::storage_api::ObjectToDelete; use crate::{ReplicationStatusType, ReplicationType, VersionPurgeStatusType, target_reset_header}; use s3s::dto::{ - DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication, - ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus, ReplicationConfiguration, - ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria, + DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination, + ExistingObjectReplication, ExistingObjectReplicationStatus, ReplicaModifications, ReplicaModificationsStatus, + ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, SourceSelectionCriteria, }; use std::collections::HashMap; use time::{Duration, OffsetDateTime}; @@ -433,7 +433,9 @@ mod tests { delete_marker_replication: Some(DeleteMarkerReplication { status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), }), - delete_replication: None, + delete_replication: Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED), + }), destination: Destination { bucket: arn.to_string(), ..Default::default() @@ -513,7 +515,7 @@ mod tests { }; let state = delete_replication_state_from_config(&config, &source) - .expect("delete-marker version purge should honor delete-marker replication rules"); + .expect("delete-marker version purge should honor delete replication rules"); let pending = format!("{arn}=PENDING;"); assert_eq!(state.version_purge_status_internal.as_deref(), Some(pending.as_str())); diff --git a/rustfs/src/admin/handlers/replication.rs b/rustfs/src/admin/handlers/replication.rs index bf1ff1f07..b2e6202e4 100644 --- a/rustfs/src/admin/handlers/replication.rs +++ b/rustfs/src/admin/handlers/replication.rs @@ -179,6 +179,20 @@ impl RemoteTargetRequest { return Err(s3_error!(InvalidRequest, "credentials.secretKey is required")); } + for (unsupported, configured) in [ + ("disableProxy", self.disable_proxy), + ("healthCheckDuration", self.health_check_duration != 0), + ("edge", self.edge), + ("edgeSyncBeforeExpiry", self.edge_sync_before_expiry), + ] { + if configured { + return Err(s3_error!( + InvalidRequest, + "remote target field {unsupported} is not supported by this RustFS version" + )); + } + } + Ok(BucketTarget { source_bucket: self.source_bucket, endpoint: self.endpoint, @@ -1204,6 +1218,27 @@ mod tests { assert!(err.to_string().contains("credentials.secretKey is required")); } + #[test] + fn remote_target_request_rejects_unimplemented_fields() { + for (field, value) in [ + ("disableProxy", serde_json::json!(true)), + ("healthCheckDuration", serde_json::json!(5)), + ("edge", serde_json::json!(true)), + ("edgeSyncBeforeExpiry", serde_json::json!(true)), + ] { + let mut request = valid_remote_target_request(); + request[field] = value; + let request: RemoteTargetRequest = + serde_json::from_value(request).expect("unsupported field should still deserialize"); + let err = request + .into_bucket_target() + .expect_err("unimplemented remote target fields must not be persisted"); + + assert!(err.to_string().contains(field)); + assert!(err.to_string().contains("not supported by this RustFS version")); + } + } + #[test] fn remote_target_request_converts_to_bucket_target() { let target = serde_json::from_value::(valid_remote_target_request()) diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 0fe1474a1..f425c663b 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -35,7 +35,7 @@ use super::storage_api::bucket_usecase::bucket::{ policy_sys::PolicySys, replication::{ ReplicationTargetValidationError, replication_target_arns, should_remove_replication_target, - validate_replication_config_target_arns, + unsupported_replication_config_field, validate_replication_config_target_arns, }, target::{BucketTargetType, BucketTargets}, utils::serialize, @@ -556,6 +556,16 @@ fn validate_replication_config_targets(targets: &BucketTargets, config: &Replica } } +fn validate_replication_config_capabilities(config: &ReplicationConfiguration) -> S3Result<()> { + if let Some(field) = unsupported_replication_config_field(config) { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + format!("replication field {field} is not supported by this RustFS version"), + )); + } + Ok(()) +} + async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationConfiguration) -> S3Result<()> { if !BucketVersioningSys::enabled(bucket).await { return Err(s3_error!( @@ -564,6 +574,8 @@ async fn validate_bucket_replication_update(bucket: &str, config: &ReplicationCo )); } + validate_replication_config_capabilities(config)?; + let targets = metadata_sys::get_bucket_targets_config(bucket) .await .map_err(|err| match err { @@ -3075,6 +3087,25 @@ mod tests { validate_replication_config_targets(&targets, &config).expect("disabled rules should not require live targets"); } + #[test] + fn validate_replication_config_capabilities_names_unsupported_field() { + let mut rule = replication_rule_for_target("arn:rustfs:replication:us-east-1:target:bucket"); + rule.destination.encryption_configuration = Some(s3s::dto::EncryptionConfiguration::default()); + let config = ReplicationConfiguration { + role: String::new(), + rules: vec![rule], + }; + + let err = validate_replication_config_capabilities(&config) + .expect_err("destination encryption must be rejected until the execution path supports it"); + + assert_eq!(err.code(), &S3ErrorCode::InvalidRequest); + assert!( + err.to_string() + .contains("Destination.EncryptionConfiguration is not supported") + ); + } + #[test] fn remove_replication_targets_from_config_targets_only_removes_referenced_replication_targets() { let removed_arn = "arn:rustfs:replication:us-east-1:removed:bucket"; diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 8d46feadd..328ad95f5 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -44,8 +44,8 @@ use super::storage_api::object_usecase::bucket::{ REPLICATE_INCOMING_DELETE, ReplicationStatusType, VersionPurgeStatusType, check_replicate_delete, delete_replication_state_from_config, delete_replication_version_id, deleted_object_has_pending_replication_delete, must_replicate_object, schedule_object_replication, schedule_replication_delete, set_deleted_object_replication_state, - set_object_to_delete_version_purge_status, should_schedule_delete_replication, - should_use_existing_delete_replication_info, should_use_existing_delete_replication_source, + set_object_to_delete_version_purge_status, should_use_existing_delete_replication_info, + should_use_existing_delete_replication_source, }, tagging::decode_tags, validate_restore_request, @@ -6704,7 +6704,7 @@ impl DefaultObjectUsecase { // the same early, advisory rejection as before. let store_ref = &store; let bucket_ref = bucket.as_str(); - let admitted_deletes: Vec = + let admitted_deletes: Vec> = futures::stream::iter(prepared_deletes.into_iter().map(|prepared| async move { let PreparedDelete { idx, @@ -6729,7 +6729,7 @@ impl DefaultObjectUsecase { && let Some(block_reason) = check_object_lock_for_deletion(bucket_ref, &goi, bypass_governance).await { let blocked_key = object.object_name.clone(); - return AdmittedDelete { + return Ok(AdmittedDelete { idx, object, size: 0, @@ -6740,15 +6740,11 @@ impl DefaultObjectUsecase { message: Some(block_reason.error_message()), version_id, }), - }; + }); } let size = goi.size; - if is_dir_object(&object.object_name) && object.version_id.is_none() { - object.version_id = Some(Uuid::nil()); - } - if replicate_deletes { let dsc = check_replicate_delete( bucket_ref, @@ -6761,7 +6757,8 @@ impl DefaultObjectUsecase { &opts, gerr.clone(), ) - .await; + .await + .map_err(ApiError::from)?; if dsc.replicate_any() { if object.version_id.is_some() { set_object_to_delete_version_purge_status(&mut object, VersionPurgeStatusType::Pending); @@ -6773,18 +6770,23 @@ impl DefaultObjectUsecase { } } + if is_dir_object(&object.object_name) && object.version_id.is_none() { + object.version_id = Some(Uuid::nil()); + } + let existing = (!skip_stat && gerr.is_none()).then_some(goi); - AdmittedDelete { + Ok(AdmittedDelete { idx, object, size, existing, blocked: None, - } + }) })) .buffered(DELETE_OBJECTS_PRE_STAT_CONCURRENCY) .collect() .await; + let admitted_deletes: Vec = admitted_deletes.into_iter().collect::>()?; // Phase 3 (serial): apply outcomes in the original request order so // per-key success/failure reporting is unchanged. @@ -7102,6 +7104,31 @@ impl DefaultObjectUsecase { } }; + let delete_replication_state = if !replica && !force_delete && (opts.versioned || opts.version_suspended) { + let fallback_source; + let source = if let Some(source) = existing_object_info.as_ref() { + source + } else { + fallback_source = ObjectInfo { + name: key.clone(), + ..Default::default() + }; + &fallback_source + }; + match metadata_sys::get_replication_config(&bucket).await { + Ok((config, _)) => delete_replication_state_from_config( + &config, + source, + version_id_clone.as_ref().and_then(|_| source.version_id), + false, + ), + Err(StorageError::ConfigNotFound) => None, + Err(err) => return Err(ApiError::from(err).into()), + } + } else { + None + }; + let cache_adapter = self.object_data_cache(); // A force (delete_prefix) delete removes every object under `key` as a // prefix, so invalidating only the exact key would strand every cached @@ -7197,7 +7224,7 @@ impl DefaultObjectUsecase { let schedule_delete_replication = if opts.replication_request && replica { should_schedule_replica_delete_replication(&bucket, replication_state_source, delete_replication_version_id).await } else { - should_schedule_delete_replication(&opts, deleted_object_source, deleted_delete_marker_version) + delete_replication_state.is_some() }; if schedule_delete_replication { @@ -7219,8 +7246,12 @@ impl DefaultObjectUsecase { replication_state: None, ..Default::default() }; - set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state()); - enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object, replication_state_source).await; + if let Some(state) = delete_replication_state.as_ref() { + set_deleted_object_replication_state(&mut deleted_object, state); + } else { + set_deleted_object_replication_state(&mut deleted_object, &replication_state_source.replication_state()); + enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object, replication_state_source).await; + } schedule_replication_delete(deleted_object, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await; } @@ -8358,9 +8389,10 @@ mod tests { use super::*; use http::{Extensions, HeaderMap, HeaderName, HeaderValue, Method, Uri}; use s3s::dto::{ - Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, Destination, ExistingObjectReplication, - ExistingObjectReplicationStatus, ObjectIdentifier, ReplicaModifications, ReplicaModificationsStatus, - ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, RestoreRequest, SourceSelectionCriteria, + Delete, DeleteMarkerReplication, DeleteMarkerReplicationStatus, DeleteReplication, DeleteReplicationStatus, Destination, + ExistingObjectReplication, ExistingObjectReplicationStatus, ObjectIdentifier, ReplicaModifications, + ReplicaModificationsStatus, ReplicationConfiguration, ReplicationRule, ReplicationRuleStatus, RestoreRequest, + SourceSelectionCriteria, }; use std::pin::Pin; use std::sync::Arc; @@ -13368,63 +13400,6 @@ mod tests { assert!(!can_skip_delete_objects_pre_stat(false, false, &delete_marker_creating_opts(), false)); } - #[test] - fn should_schedule_delete_replication_skips_replica_requests() { - let opts = ObjectOptions { - replication_request: true, - version_id: Some(Uuid::new_v4().to_string()), - ..Default::default() - }; - let replication_source = ObjectInfo { - delete_marker: true, - replication_status: ReplicationStatusType::Completed, - ..Default::default() - }; - - assert!( - !should_schedule_delete_replication(&opts, &replication_source, true), - "replica delete requests on target sites must not enqueue a second replication delete task" - ); - } - - #[test] - fn should_schedule_delete_replication_keeps_delete_marker_version_purge_from_source() { - let opts = ObjectOptions { - replication_request: false, - version_id: Some(Uuid::new_v4().to_string()), - ..Default::default() - }; - let replication_source = ObjectInfo { - delete_marker: true, - replication_status: ReplicationStatusType::Completed, - ..Default::default() - }; - - assert!( - should_schedule_delete_replication(&opts, &replication_source, true), - "source-side delete-marker version purge still needs replication scheduling" - ); - } - - #[test] - fn should_schedule_delete_replication_keeps_object_version_purge_from_completed_source() { - let opts = ObjectOptions { - replication_request: false, - version_id: Some(Uuid::new_v4().to_string()), - ..Default::default() - }; - let replication_source = ObjectInfo { - delete_marker: false, - replication_status: ReplicationStatusType::Completed, - ..Default::default() - }; - - assert!( - should_schedule_delete_replication(&opts, &replication_source, false), - "source-side object version purge must still enqueue delete replication after the original PUT completed" - ); - } - #[tokio::test] #[ignore = "requires isolated global object layer state"] async fn execute_get_object_attributes_returns_internal_error_when_store_uninitialized() { @@ -13689,7 +13664,9 @@ mod tests { delete_marker_replication: Some(DeleteMarkerReplication { status: Some(DeleteMarkerReplicationStatus::from_static(DeleteMarkerReplicationStatus::ENABLED)), }), - delete_replication: None, + delete_replication: Some(DeleteReplication { + status: DeleteReplicationStatus::from_static(DeleteReplicationStatus::ENABLED), + }), destination: Destination { bucket: arn.clone(), ..Default::default() diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 4b9f6daa7..5bfc615b9 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -649,8 +649,8 @@ pub(crate) mod bucket { oi: &crate::storage::storage_api::StorageObjectInfo, del_opts: &crate::storage::storage_api::StorageObjectOptions, gerr: Option, - ) -> ReplicateDecision { - ReplicationObjectBridge::check_delete(bucket, dobj, oi, del_opts, gerr).await + ) -> Result { + ReplicationObjectBridge::check_delete_strict(bucket, dobj, oi, del_opts, gerr).await } pub(crate) fn delete_replication_version_id( @@ -751,21 +751,6 @@ pub(crate) mod bucket { replication_contracts::should_remove_replication_target(target_arn, is_replication_service, target_arns) } - pub(crate) fn should_schedule_delete_replication( - opts: &crate::storage::storage_api::StorageObjectOptions, - replication_source: &crate::storage::storage_api::StorageObjectInfo, - deleted_delete_marker_version: bool, - ) -> bool { - replication_contracts::should_schedule_delete_replication(replication_contracts::ReplicationDeleteScheduleInput { - replication_request: opts.replication_request, - version_id_requested: opts.version_id.is_some(), - source_delete_marker: replication_source.delete_marker, - source_replication_status: &replication_source.replication_status, - source_version_purge_status: &replication_source.version_purge_status, - deleted_delete_marker_version, - }) - } - pub(crate) fn should_use_existing_delete_replication_info( opts: &crate::storage::storage_api::StorageObjectOptions, ) -> bool { @@ -790,6 +775,10 @@ pub(crate) mod bucket { ) -> Result<(), ReplicationTargetValidationError> { replication_contracts::validate_replication_config_target_arns(configured_arns, config) } + + pub(crate) fn unsupported_replication_config_field(config: &s3s::dto::ReplicationConfiguration) -> Option<&'static str> { + replication_contracts::unsupported_replication_config_field(config) + } } pub(crate) mod tagging {