diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index d54a168ad..699cecc83 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -22,7 +22,8 @@ use super::replication_filemeta_boundary::{ use super::replication_lock_boundary::ReplicationLockTiming; use super::replication_logging::{EVENT_REPLICATION_CONFIG_LOOKUP_SKIPPED, LOG_COMPONENT_ECSTORE, LOG_SUBSYSTEM_REPLICATION}; use super::replication_metadata_boundary::ReplicationMetadataStore; -use super::replication_object_config::{ReplicationConfig, check_replicate_delete}; +use super::replication_object_config::{ReplicationConfig, check_replicate_delete, must_replicate}; +use super::replication_object_decision_boundary::MustReplicateOptions; use super::replication_queue_boundary::{ DeletedObjectReplicationInfo, LARGE_WORKER_COUNT, ReplicationBackpressureRecommendation, ReplicationBackpressureState, ReplicationHealQueueAction, ReplicationHealQueueResult, ReplicationHealResyncDeletes, ReplicationOperation, @@ -1021,6 +1022,28 @@ impl ReplicationPool { queue_replication_heal(&entry.bucket, oi, entry.retry_count as u32).await; queued_count += 1; } + MrfOpKind::Metadata => { + let opts = ObjectOptions { + version_id: entry.version_id.map(|u| u.to_string()), + ..Default::default() + }; + let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await { + Ok(oi) => oi, + Err(e) => { + debug!( + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + bucket = %entry.bucket, + object = %entry.object, + error = %e, + "MRF metadata recovery: object not found, skipping" + ); + continue; + } + }; + queue_replication_metadata(&entry.bucket, oi, entry.retry_count as u32).await; + queued_count += 1; + } } } @@ -1942,6 +1965,26 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u queue_replication_heal_internal(bucket, oi, rcfg_wrapper, retry_count).await; } +pub async fn queue_replication_metadata(bucket: &str, oi: ObjectInfo, retry_count: u32) { + let dsc = must_replicate( + bucket, + &oi.name, + MustReplicateOptions::new(&oi.user_defined, (*oi.user_tags).clone(), ReplicationType::Metadata, false) + .with_replication_status(oi.replication_status.clone()), + ) + .await; + + if !dsc.replicate_any() { + return; + } + + let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata); + roi.retry_count = retry_count; + if let Some(pool) = runtime_sources::replication_pool() { + let _ = pool.queue_replica_task(roi).await; + } +} + /// queue_replication_heal_internal enqueues objects that failed replication OR eligible for resyncing through /// an ongoing resync operation or via existing objects replication configuration setting. pub(crate) async fn queue_replication_heal_internal( diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index f28c18cb6..462865cb4 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -96,7 +96,6 @@ const EVENT_REPLICATION_FORCE_DELETE_SKIPPED: &str = "replication_force_delete_s 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", @@ -2860,7 +2859,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { // The target already holds a matching object (reached here only via // the version-id fallback ETag match above) — there is nothing to // copy. Record it as synced and return, instead of falling into the - // metadata-unsupported failure branch below, which previously left + // metadata propagation path below, which previously left // AWS-style targets permanently FAILED and never converging // (backlog#860 / #799 B11). if self.op_type == ReplicationType::ExistingObject && !tgt_client.reset_id.is_empty() { @@ -2877,10 +2876,71 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { return rinfo; } - // action == Metadata: metadata-only replication is not implemented. - if replication_action != ReplicationAction::All { + // The target client has no metadata-only operation. Reuse the existing + // object transport so metadata changes carry tags and object-lock state + // atomically with the source version. + let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) { + Ok((put_opts, is_mp)) => (put_opts, is_mp), + Err(e) => { + rinfo.error = Some(e.to_string()); + warn!( + event = EVENT_RESYNC_TARGET_OPERATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, + bucket = %bucket, + arn = %tgt_client.arn, + operation = "build_put_options", + error = %e, + "Replication target operation failed" + ); + send_local_event(EventArgs { + event_name: EventName::ObjectReplicationNotTracked.to_string(), + bucket_name: bucket.clone(), + object: object_info, + user_agent: "Internal: [Replication]".to_string(), + ..Default::default() + }); + + rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + return rinfo; + } + }; + + let has_tagging_replication = !put_opts.user_tags.is_empty(); + if let Some(err) = if is_multipart { + drop(gr); + let result = replicate_object_with_multipart(MultipartReplicationContext { + storage: storage.clone(), + cli: tgt_client.clone(), + src_bucket: &bucket, + dst_bucket: &tgt_client.bucket, + object: &object, + object_info: &object_info, + obj_opts: &obj_opts, + arn: &rinfo.arn, + put_opts, + }) + .await; + record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } + result.err() + } else { + gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); + let byte_stream = async_read_to_bytestream(gr.stream); + let result = tgt_client + .put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts) + .await + .map_err(|e| std::io::Error::other(e.to_string())); + record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } + result.err() + } { rinfo.replication_status = ReplicationStatusType::Failed; - rinfo.error = Some(ERR_REPLICATION_METADATA_COPY_UNSUPPORTED.to_string()); + rinfo.error = Some(err.to_string()); warn!( event = EVENT_RESYNC_TARGET_OPERATION_FAILED, component = LOG_COMPONENT_ECSTORE, @@ -2888,98 +2948,14 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { bucket = %bucket, arn = %tgt_client.arn, object = %object, - operation = "copy_object_metadata", - error = ERR_REPLICATION_METADATA_COPY_UNSUPPORTED, + operation = "put_object", + error = ?err, "Replication target operation failed" ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: object_info, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); + + mark_replication_target_offline_if_needed(&tgt_client, &err).await; return rinfo; - } else { - let (put_opts, is_multipart) = match replication_put_object_options(&tgt_client.storage_class, &object_info) { - Ok((put_opts, is_mp)) => (put_opts, is_mp), - Err(e) => { - rinfo.error = Some(e.to_string()); - warn!( - event = EVENT_RESYNC_TARGET_OPERATION_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - arn = %tgt_client.arn, - operation = "build_put_options", - error = %e, - "Replication target operation failed" - ); - send_local_event(EventArgs { - event_name: EventName::ObjectReplicationNotTracked.to_string(), - bucket_name: bucket.clone(), - object: object_info, - user_agent: "Internal: [Replication]".to_string(), - ..Default::default() - }); - - rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); - return rinfo; - } - }; - - let has_tagging_replication = !put_opts.user_tags.is_empty(); - if let Some(err) = if is_multipart { - drop(gr); - let result = replicate_object_with_multipart(MultipartReplicationContext { - storage: storage.clone(), - cli: tgt_client.clone(), - src_bucket: &bucket, - dst_bucket: &tgt_client.bucket, - object: &object, - object_info: &object_info, - obj_opts: &obj_opts, - arn: &rinfo.arn, - put_opts, - }) - .await; - record_proxy_request(&bucket, "PutObject", result.is_err()).await; - if has_tagging_replication { - record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; - } - result.err() - } else { - gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); - let byte_stream = async_read_to_bytestream(gr.stream); - let result = tgt_client - .put_object(&tgt_client.bucket, &object, size, byte_stream, &put_opts) - .await - .map_err(|e| std::io::Error::other(e.to_string())); - record_proxy_request(&bucket, "PutObject", result.is_err()).await; - if has_tagging_replication { - record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; - } - result.err() - } { - rinfo.replication_status = ReplicationStatusType::Failed; - rinfo.error = Some(err.to_string()); - warn!( - event = EVENT_RESYNC_TARGET_OPERATION_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_REPLICATION_RESYNC, - bucket = %bucket, - arn = %tgt_client.arn, - object = %object, - operation = "put_object", - error = ?err, - "Replication target operation failed" - ); - rinfo.duration = (OffsetDateTime::now_utc() - start_time).unsigned_abs(); - - mark_replication_target_offline_if_needed(&tgt_client, &err).await; - return rinfo; - } } rinfo diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 45a71516c..133a0bc60 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -2768,6 +2768,11 @@ impl SetDisks { let (mut fi, _, disks) = self.get_object_fileinfo_gated(bucket, object, opts, false, false).await?; fi.metadata.insert(AMZ_OBJECT_TAGGING.to_owned(), tags.to_owned()); + if let Some(eval_metadata) = &opts.eval_metadata { + for (key, value) in eval_metadata { + fi.metadata.insert(key.clone(), value.clone()); + } + } #[cfg(test)] pause_object_tagging_commit(bucket, object).await; diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index 667e9a3d1..172c364e3 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -550,6 +550,8 @@ pub enum MrfOpKind { #[default] #[serde(rename = "object")] Object, + #[serde(rename = "metadata")] + Metadata, #[serde(rename = "delete")] Delete, } @@ -810,7 +812,11 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo { version_id: self.version_id, retry_count: retry_count_to_mrf(self.retry_count), size: self.size, - op: MrfOpKind::Object, + op: if self.op_type == ReplicationType::Metadata { + MrfOpKind::Metadata + } else { + MrfOpKind::Object + }, delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, @@ -865,7 +871,11 @@ impl ReplicateObjectInfo { version_id: self.version_id, retry_count: retry_count_to_mrf(self.retry_count), size: self.size, - op: MrfOpKind::Object, + op: if self.op_type == ReplicationType::Metadata { + MrfOpKind::Metadata + } else { + MrfOpKind::Object + }, delete_marker_version_id: None, delete_marker: false, delete_marker_mtime: None, @@ -1067,6 +1077,18 @@ mod tests { assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]); } + #[test] + fn metadata_replication_mrf_entry_preserves_operation_kind() { + let info = ReplicateObjectInfo { + bucket: "bucket".to_string(), + name: "object".to_string(), + op_type: ReplicationType::Metadata, + ..Default::default() + }; + + assert_eq!(info.to_mrf_entry().op, MrfOpKind::Metadata); + } + #[test] fn target_state_reads_resync_timestamp_from_target_reset_header_key() { let arn = "arn:rustfs:replication:us-east-1:target:bucket"; diff --git a/crates/replication/src/mrf.rs b/crates/replication/src/mrf.rs index 2b9fff269..45faaf301 100644 --- a/crates/replication/src/mrf.rs +++ b/crates/replication/src/mrf.rs @@ -57,10 +57,22 @@ mod tests { use uuid::Uuid; #[test] - fn mrf_file_round_trips_object_and_delete_entries() { + fn mrf_file_round_trips_object_metadata_and_delete_entries() { let obj_vid = Uuid::new_v4(); let del_vid = Uuid::new_v4(); let entries = vec![ + MrfReplicateEntry { + bucket: "bucket-a".to_string(), + object: "metadata-a".to_string(), + version_id: Some(obj_vid), + retry_count: 1, + size: 1024, + op: MrfOpKind::Metadata, + delete_marker_version_id: None, + delete_marker: false, + delete_marker_mtime: None, + target_arns: vec!["arn:target-a".to_string()], + }, MrfReplicateEntry { bucket: "bucket-a".to_string(), object: "object-a".to_string(), @@ -90,17 +102,20 @@ mod tests { let encoded = encode_mrf_file(&entries).expect("mrf file should encode"); let decoded = decode_mrf_file(&encoded).expect("mrf file should decode"); - assert_eq!(decoded.len(), 2); + assert_eq!(decoded.len(), 3); assert_eq!(decoded[0].version_id, Some(obj_vid)); - assert_eq!(decoded[0].op, MrfOpKind::Object); - assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string(), "arn:target-b".to_string()]); + assert_eq!(decoded[0].op, MrfOpKind::Metadata); + assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string()]); assert_eq!(decoded[0].delete_marker_mtime, None); - assert_eq!(decoded[1].delete_marker_version_id, Some(del_vid)); - assert_eq!(decoded[1].op, MrfOpKind::Delete); - assert_eq!(decoded[1].target_arns, vec!["arn:target-a".to_string()]); - assert!(decoded[1].delete_marker); + assert_eq!(decoded[1].op, MrfOpKind::Object); + assert_eq!(decoded[1].target_arns, vec!["arn:target-a".to_string(), "arn:target-b".to_string()]); + assert_eq!(decoded[1].delete_marker_mtime, None); + assert_eq!(decoded[2].delete_marker_version_id, Some(del_vid)); + assert_eq!(decoded[2].op, MrfOpKind::Delete); + assert_eq!(decoded[2].target_arns, vec!["arn:target-a".to_string()]); + assert!(decoded[2].delete_marker); assert_eq!( - decoded[1].delete_marker_mtime, + decoded[2].delete_marker_mtime, Some(1_705_312_200_123_456_789), "delete-marker mtime must survive the MRF disk round-trip" ); diff --git a/crates/replication/src/object.rs b/crates/replication/src/object.rs index 572d97d97..afbb2ef46 100644 --- a/crates/replication/src/object.rs +++ b/crates/replication/src/object.rs @@ -164,6 +164,7 @@ mod tests { replication_etags_match, target_is_newer_than_source_null_version, }; use crate::filemeta::{ReplicationAction, ReplicationType}; + use crate::http::AMZ_OBJECT_LOCK_MODE; use std::collections::HashMap; use time::{Duration, OffsetDateTime}; @@ -267,4 +268,27 @@ mod tests { ReplicationAction::Metadata ); } + + #[test] + fn replication_action_detects_tags_and_object_lock_metadata_differences() { + let mut source_metadata = HashMap::new(); + source_metadata.insert(AMZ_OBJECT_LOCK_MODE.to_string(), "GOVERNANCE".to_string()); + let source = ReplicationSourceObject { + user_tags: "a=1&b=2", + user_defined: &source_metadata, + ..source_object(&source_metadata) + }; + + let target_metadata = HashMap::new(); + let target = ReplicationTargetObject { + tag_count: 1, + metadata: Some(&target_metadata), + ..target_object(&target_metadata) + }; + + assert_eq!( + replication_action_for_target(&source, &target, ReplicationType::Metadata), + ReplicationAction::Metadata + ); + } } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index c6df2bd75..feec504d8 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -685,15 +685,50 @@ pub(crate) mod bucket { status: ReplicationStatusType, opts: crate::storage::storage_api::StorageObjectOptions, ) -> ReplicateDecision { - #[cfg(test)] - MUST_REPLICATE_OBJECT_CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let mopts = ReplicationObjectBridge::must_replicate_options( + must_replicate_with_type( + bucket, + object, user_defined, user_tags, status, - replication_contracts::ReplicationType::Object, opts, - ); + replication_contracts::ReplicationType::Object, + ) + .await + } + + pub(crate) async fn must_replicate_metadata( + bucket: &str, + object: &str, + user_defined: &HashMap, + user_tags: String, + status: ReplicationStatusType, + opts: crate::storage::storage_api::StorageObjectOptions, + ) -> ReplicateDecision { + must_replicate_with_type( + bucket, + object, + user_defined, + user_tags, + status, + opts, + replication_contracts::ReplicationType::Metadata, + ) + .await + } + + async fn must_replicate_with_type( + bucket: &str, + object: &str, + user_defined: &HashMap, + user_tags: String, + status: ReplicationStatusType, + opts: crate::storage::storage_api::StorageObjectOptions, + op_type: replication_contracts::ReplicationType, + ) -> ReplicateDecision { + #[cfg(test)] + MUST_REPLICATE_OBJECT_CALLS.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + let mopts = ReplicationObjectBridge::must_replicate_options(user_defined, user_tags, status, op_type, opts); ReplicationObjectBridge::must_replicate(bucket, object, mopts).await } @@ -705,6 +740,14 @@ pub(crate) mod bucket { ReplicationObjectBridge::schedule_object(oi, store, dsc, replication_contracts::ReplicationType::Object).await; } + pub(crate) async fn schedule_metadata_replication( + oi: crate::storage::storage_api::StorageObjectInfo, + store: Arc, + dsc: ReplicateDecision, + ) { + ReplicationObjectBridge::schedule_object(oi, store, dsc, replication_contracts::ReplicationType::Metadata).await; + } + pub(crate) async fn schedule_replication_delete( delete_object: crate::storage::storage_api::StorageDeletedObject, bucket: String, diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 0745e3070..1d139d591 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -47,6 +47,7 @@ use rustfs_utils::http::headers::{ }; use rustfs_utils::http::{SUFFIX_REPLICATION_STATUS, SUFFIX_REPLICATION_TIMESTAMP, insert_str}; use s3s::{S3, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, dto::*, s3_error}; +use std::collections::HashMap; use std::fmt::Debug; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; use tracing::{debug, error, instrument, warn}; @@ -58,7 +59,7 @@ const LOG_SUBSYSTEM_OBJECT_LOCK: &str = "object_lock"; const LOG_SUBSYSTEM_TAGGING: &str = "tagging"; use crate::app::storage_api::object_usecase::bucket::replication::{ - ReplicateDecision, must_replicate_object, schedule_object_replication, + ReplicateDecision, must_replicate_metadata, schedule_metadata_replication, }; use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions; @@ -431,15 +432,29 @@ impl S3 for FS { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); }; - let version_id_for_parse = version_id.clone(); - let opts = ObjectOptions { - version_id: parse_object_version_id(version_id_for_parse)?.map(Into::into), - ..Default::default() - }; + let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers) + .await + .map_err(ApiError::from)?; + let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?; + let dsc = must_replicate_metadata( + &bucket, + &object, + &existing_object_info.user_defined, + String::new(), + existing_object_info.replication_status.clone(), + opts.clone(), + ) + .await; + if dsc.replicate_any() { + let mut eval_metadata = HashMap::new(); + insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); + insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + opts.eval_metadata = Some(eval_metadata); + } let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await; Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await; - delete_tags_result.map_err(|e| { + let object_info = delete_tags_result.map_err(|e| { error!( component = LOG_COMPONENT_STORAGE, subsystem = LOG_SUBSYSTEM_TAGGING, @@ -452,19 +467,10 @@ impl S3 for FS { ApiError::from(e) })?; - let event_object_info = match store.get_object_info(&bucket, &object, &opts).await { - Ok(info) => Some(info), - Err(err) => { - warn!( - bucket = %bucket, - object = %object, - version_id = ?version_id, - error = %err, - "failed to load object info for delete-object-tagging notification; falling back to request context" - ); - None - } - }; + let event_object_info = Some(object_info.clone()); + if dsc.replicate_any() { + schedule_metadata_replication(object_info, store.clone(), dsc).await; + } counter!("rustfs_delete_object_tagging_success").increment(1); @@ -1302,7 +1308,7 @@ impl S3 for FS { // to re-drive. Scheduling without that marker would lose the task silently. let dsc = match store.get_object_info(&bucket, &key, &opts).await.ok() { Some(info) => { - must_replicate_object( + must_replicate_metadata( &bucket, &key, &info.user_defined, @@ -1327,10 +1333,9 @@ impl S3 for FS { s3_error!(InternalError, "{}", e.to_string()) })?; - // This replicates the whole object, not just the changed metadata: metadata-only - // replication is not implemented yet, so the lock change triggers a full re-upload. + // The current target transport carries the updated metadata in a full-object PUT. if dsc.replicate_any() { - schedule_object_replication(info.clone(), store.clone(), dsc).await; + schedule_metadata_replication(info.clone(), store.clone(), dsc).await; } let output = PutObjectLegalHoldOutput { @@ -1508,7 +1513,7 @@ impl S3 for FS { // to re-drive. Scheduling without that marker would lose the task silently. let dsc = match existing_obj_info.as_ref() { Some(info) => { - must_replicate_object( + must_replicate_metadata( &bucket, &key, &info.user_defined, @@ -1533,10 +1538,9 @@ impl S3 for FS { S3Error::from(ApiError::from(e)) })?; - // This replicates the whole object, not just the changed metadata: metadata-only - // replication is not implemented yet, so the lock change triggers a full re-upload. + // The current target transport carries the updated metadata in a full-object PUT. if dsc.replicate_any() { - schedule_object_replication(object_info.clone(), store.clone(), dsc).await; + schedule_metadata_replication(object_info.clone(), store.clone(), dsc).await; } let output = PutObjectRetentionOutput { @@ -1576,32 +1580,38 @@ impl S3 for FS { debug!("Encoded tags: {}", tags); let version_id = req.input.version_id.clone(); - let opts = ObjectOptions { - version_id: parse_object_version_id(version_id)?.map(Into::into), - ..Default::default() - }; + let mut opts = get_opts(&bucket, &object, version_id.clone(), None, &req.headers) + .await + .map_err(ApiError::from)?; + let existing_object_info = store.get_object_info(&bucket, &object, &opts).await.map_err(ApiError::from)?; + let dsc = must_replicate_metadata( + &bucket, + &object, + &existing_object_info.user_defined, + tags.clone(), + existing_object_info.replication_status.clone(), + opts.clone(), + ) + .await; + if dsc.replicate_any() { + let mut eval_metadata = HashMap::new(); + insert_str(&mut eval_metadata, SUFFIX_REPLICATION_TIMESTAMP, jiff::Zoned::now().to_string()); + insert_str(&mut eval_metadata, SUFFIX_REPLICATION_STATUS, dsc.pending_status().unwrap_or_default()); + opts.eval_metadata = Some(eval_metadata); + } let put_tags_result = store.put_object_tags(&bucket, &object, &tags, &opts).await; Self::record_replication_tagging_metric(&bucket, &object, "PutObjectTagging", put_tags_result.is_err()).await; - put_tags_result.map_err(|e| { + let object_info = put_tags_result.map_err(|e| { error!("Failed to put object tags: {}", e); counter!("rustfs_put_object_tagging_failure").increment(1); ApiError::from(e) })?; - let event_object_info = match store.get_object_info(&bucket, &object, &opts).await { - Ok(info) => Some(info), - Err(err) => { - warn!( - bucket = %bucket, - object = %object, - version_id = ?req.input.version_id, - error = %err, - "failed to load object info for put-object-tagging notification; falling back to request context" - ); - None - } - }; + let event_object_info = Some(object_info.clone()); + if dsc.replicate_any() { + schedule_metadata_replication(object_info, store.clone(), dsc).await; + } counter!("rustfs_put_object_tagging_success").increment(1);