diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index 10db8c767..c95c3e76a 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -187,7 +187,11 @@ fn site_replicator_service_account_policy() -> S3Result { "s3:PutObjectTagging", "s3:PutObjectVersionTagging", "s3:DeleteObjectTagging", - "s3:DeleteObjectVersionTagging" + "s3:DeleteObjectVersionTagging", + "s3:GetObjectRetention", + "s3:PutObjectRetention", + "s3:GetObjectLegalHold", + "s3:PutObjectLegalHold" ], "Resource": ["arn:aws:s3:::*/*"] } @@ -8860,6 +8864,57 @@ mod tests { assert!(!policy.is_allowed(&put_policy_args).await); } + // The replication service account must be able to carry object-lock metadata to the peer. + // Without these actions the peer answers AccessDenied for any replicated object that has + // retention or a legal hold, so a WORM-protected object never reaches the replica at all, + // and a retention change made after upload never propagates. + #[tokio::test] + async fn test_site_replicator_policy_allows_object_lock_replication() { + let policy = site_replicator_service_account_policy().expect("site replicator policy should parse"); + let groups: Option> = None; + let claims = HashMap::new(); + let conditions = HashMap::new(); + + let base_args = rustfs_policy::policy::Args { + account: SITE_REPLICATOR_SERVICE_ACCOUNT, + groups: &groups, + action: Action::S3Action(S3Action::PutObjectRetentionAction), + conditions: &conditions, + is_owner: false, + claims: &claims, + deny_only: false, + bucket: "photos", + object: "image.jpg", + }; + + for action in [ + S3Action::PutObjectRetentionAction, + S3Action::GetObjectRetentionAction, + S3Action::PutObjectLegalHoldAction, + S3Action::GetObjectLegalHoldAction, + ] { + let args = rustfs_policy::policy::Args { + action: Action::S3Action(action), + ..base_args + }; + assert!( + policy.is_allowed(&args).await, + "site replicator must be allowed to replicate object-lock metadata: {action:?}" + ); + } + + // Governance bypass stays denied: replication must not be able to erase a retained + // version on the peer. + let bypass_args = rustfs_policy::policy::Args { + action: Action::S3Action(S3Action::BypassGovernanceRetentionAction), + ..base_args + }; + assert!( + !policy.is_allowed(&bypass_args).await, + "site replicator must not be granted governance bypass" + ); + } + #[test] fn test_sr_peer_edit_handler_uses_site_replication_operation_action() { let src = include_str!("site_replication.rs"); diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 1a4dc96e7..97dc82d23 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/rustfs/src/app/lifecycle_transition_api_test.rs @@ -2089,3 +2089,78 @@ async fn put_object_computes_replication_decision_exactly_once() { pre-commit + post-commit recompute makes this observe 2 (rustfs/backlog#1320)" ); } + +/// The object-lock handlers do not go through the object PUT path, so they must schedule +/// replication themselves. Without it a retention or legal hold applied after upload stays +/// local and the replica keeps its previous, unprotected state (a WORM object that is still +/// deletable on the peer). This observes the shared decision counter: each handler must +/// compute a replication decision exactly once, which is the step that drives both the +/// persisted pending marker and the schedule. +#[tokio::test(flavor = "multi_thread", worker_threads = 1)] +#[serial] +#[ignore = "global-state usecase integration test: runs serialized in the CI ILM Integration (serial) lane, see ci.yml test-ilm-integration-serial and rustfs/backlog#1148 (ilm-1)"] +async fn object_lock_handlers_schedule_replication() { + use super::storage_api::object_usecase::bucket::replication::MUST_REPLICATE_OBJECT_CALLS; + use s3s::S3; + use std::sync::atomic::Ordering; + + let (_disk_paths, ecstore) = setup_test_env().await; + let fs = FS::new(); + + let bucket = format!("test-lock-repl-{}", &Uuid::new_v4().simple().to_string()[..8]); + let object = "locked/object.txt"; + + (*ecstore) + .make_bucket( + bucket.as_str(), + &MakeBucketOptions { + lock_enabled: true, + versioning_enabled: true, + ..Default::default() + }, + ) + .await + .expect("Failed to create object-lock bucket"); + upload_test_object(&ecstore, bucket.as_str(), object, b"worm payload").await; + + // PutObjectRetention must compute a replication decision. + MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst); + let retention_input = PutObjectRetentionInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .retention(Some(ObjectLockRetention { + mode: Some(ObjectLockRetentionMode::from_static(ObjectLockRetentionMode::COMPLIANCE)), + retain_until_date: Some(time::macros::datetime!(2030-01-01 00:00:00 UTC).into()), + })) + .build() + .unwrap(); + fs.put_object_retention(build_request(retention_input, Method::PUT)) + .await + .expect("put_object_retention should succeed"); + assert_eq!( + MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst), + 1, + "PutObjectRetention must compute the replication decision exactly once; 0 means the \ + retention change never replicates and the peer copy stays unprotected" + ); + + // PutObjectLegalHold must do the same. + MUST_REPLICATE_OBJECT_CALLS.store(0, Ordering::SeqCst); + let legal_hold_input = PutObjectLegalHoldInput::builder() + .bucket(bucket.clone()) + .key(object.to_string()) + .legal_hold(Some(ObjectLockLegalHold { + status: Some(ObjectLockLegalHoldStatus::from_static(ObjectLockLegalHoldStatus::ON)), + })) + .build() + .unwrap(); + fs.put_object_legal_hold(build_request(legal_hold_input, Method::PUT)) + .await + .expect("put_object_legal_hold should succeed"); + assert_eq!( + MUST_REPLICATE_OBJECT_CALLS.load(Ordering::SeqCst), + 1, + "PutObjectLegalHold must compute the replication decision exactly once; 0 means the \ + legal hold never replicates and the peer copy stays deletable" + ); +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 7fdb63a5a..0745e3070 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -45,6 +45,7 @@ use rustfs_targets::EventName; use rustfs_utils::http::headers::{ AMZ_OBJECT_LOCK_LEGAL_HOLD_LOWER, AMZ_OBJECT_LOCK_MODE_LOWER, AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE_LOWER, }; +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::fmt::Debug; use time::{OffsetDateTime, format_description::well_known::Rfc3339}; @@ -56,6 +57,9 @@ const LOG_SUBSYSTEM_OBJECT: &str = "object"; 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, +}; use crate::storage::storage_api::ecfs_consumer::StorageObjectOptions as ObjectOptions; #[derive(Debug, Clone)] @@ -1282,20 +1286,53 @@ impl S3 for FS { .await .map_err(ApiError::from)?; - let eval_metadata = parse_object_lock_legal_hold(legal_hold)?; - - let popts = ObjectOptions { + let mut popts = ObjectOptions { mod_time: opts.mod_time, - version_id: opts.version_id, - eval_metadata: Some(eval_metadata), + version_id: opts.version_id.clone(), ..Default::default() }; + // PutObjectLegalHold only rewrites metadata, so replication is not scheduled by the + // object PUT path. Schedule it explicitly, otherwise the legal hold never reaches the + // replica and the peer copy remains deletable. + // + // Mirror the PUT path ordering (see object_usecase::put_object): compute the decision + // once BEFORE the commit and persist the pending marker with it, so a restart between + // the commit and the worker write-back leaves a Pending status on disk for the scanner + // 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( + &bucket, + &key, + &info.user_defined, + info.user_tags.as_ref().clone(), + popts.delete_marker_replication_status(), + popts.clone(), + ) + .await + } + None => ReplicateDecision::new(), + }; + + let mut eval_metadata = parse_object_lock_legal_hold(legal_hold)?; + if dsc.replicate_any() { + 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()); + } + popts.eval_metadata = Some(eval_metadata); + let info = store.put_object_metadata(&bucket, &key, &popts).await.map_err(|e| { error!("put_object_metadata failed, {}", e.to_string()); 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. + if dsc.replicate_any() { + schedule_object_replication(info.clone(), store.clone(), dsc).await; + } + let output = PutObjectLegalHoldOutput { request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), }; @@ -1438,7 +1475,9 @@ impl S3 for FS { .await .map_err(ApiError::from)?; - if let Ok(existing_obj_info) = store.get_object_info(&bucket, &key, &check_opts).await + let existing_obj_info = store.get_object_info(&bucket, &key, &check_opts).await.ok(); + + if let Some(existing_obj_info) = existing_obj_info.as_ref() && let Some(block_reason) = check_retention_for_modification( &existing_obj_info.user_defined, new_mode.as_deref(), @@ -1449,23 +1488,57 @@ impl S3 for FS { return Err(S3Error::with_message(S3ErrorCode::AccessDenied, block_reason.error_message())); } - let eval_metadata = parse_object_lock_retention(retention)?; - let mut opts: ObjectOptions = get_opts(&bucket, &key, version_id, None, &req.headers) .await .map_err(ApiError::from)?; - opts.eval_metadata = Some(eval_metadata); opts.object_lock_retention = Some(ObjectLockRetentionOptions { mode: new_mode, retain_until: new_retain_until, bypass_governance, }); + // PutObjectRetention only rewrites metadata, so the object PUT path that normally + // computes a replication decision and schedules replication never runs for it. Without + // scheduling here the peer keeps the previous, unprotected lock state and a + // WORM-protected object stays deletable on the replica. + // + // Mirror the PUT path ordering (see object_usecase::put_object): compute the decision + // once BEFORE the commit and persist the pending marker with it, so a restart between + // the commit and the worker write-back leaves a Pending status on disk for the scanner + // 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( + &bucket, + &key, + &info.user_defined, + info.user_tags.as_ref().clone(), + opts.delete_marker_replication_status(), + opts.clone(), + ) + .await + } + None => ReplicateDecision::new(), + }; + + let mut eval_metadata = parse_object_lock_retention(retention)?; + if dsc.replicate_any() { + 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 object_info = store.put_object_metadata(&bucket, &key, &opts).await.map_err(|e| { error!("put_object_metadata failed, {}", e.to_string()); 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. + if dsc.replicate_any() { + schedule_object_replication(object_info.clone(), store.clone(), dsc).await; + } + let output = PutObjectRetentionOutput { request_charged: Some(RequestCharged::from_static(RequestCharged::REQUESTER)), };