mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 08:18:18 +00:00
fix(site-replication): replicate object-lock retention and legal hold (#5105)
* fix(site-replication): replicate object-lock retention and legal hold Object Lock did not survive site replication, so a WORM-protected object was either missing from the peer entirely or present there unprotected. Two independent defects caused this. 1. The site-replicator service account policy granted no object-lock actions. Any replicated object carrying x-amz-object-lock-mode / x-amz-object-lock-retain-until-date was rejected by the peer with AccessDenied, so an object created under a retention rule never replicated at all. Grant s3:GetObjectRetention, s3:PutObjectRetention, s3:GetObjectLegalHold and s3:PutObjectLegalHold. s3:BypassGovernanceRetention is deliberately NOT granted: replication must never be able to erase a retained version on the peer. 2. PutObjectRetention and PutObjectLegalHold only rewrote object metadata. The object PUT path is what normally computes a replication decision and schedules replication, so a lock applied after upload stayed local and the peer kept its previous, unprotected state. Schedule replication explicitly after both metadata writes. Verified live on a two-site cluster: an object created under a COMPLIANCE retention rule now replicates, the peer copy reports COMPLIANCE, a version-scoped delete on the peer is refused, and the content matches by MD5. Retention and legal hold applied after upload propagate to the peer within about 15s, and the replicated copy is likewise undeletable. No AccessDenied replication errors remain. Adds a regression test asserting the policy allows the four object-lock actions and still denies governance bypass. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> * fix(site-replication): persist pending marker before object-lock metadata commit Review follow-up on the object-lock replication fix. PutObjectRetention and PutObjectLegalHold scheduled replication but did not persist a pending replication marker first, unlike the canonical object PUT path. If the process restarted between the metadata commit and the replication worker write-back, the in-memory task was lost and the object carried no Pending/Failed marker, so the scanner heal skipped it and the peer silently stayed unprotected. Both handlers now mirror the PUT path ordering: compute the replication decision once BEFORE the commit, fold SUFFIX_REPLICATION_STATUS and SUFFIX_REPLICATION_TIMESTAMP into the eval_metadata passed to put_object_metadata, then schedule after the commit. eval_metadata is merged into the object metadata, so the lock fields are preserved. When the object info cannot be read the decision is empty and nothing is scheduled. Also aligns the two handlers to build their options the same way, and notes at both schedule sites that this triggers a full object re-upload because metadata-only replication is not implemented yet. Adds an integration test asserting both handlers compute a replication decision exactly once, so a regression that drops the scheduling is caught. Verified live on a two-site cluster: applying retention while the peer is down now leaves Replication Status: PENDING on disk at commit time, and once the peer returns the lock propagates in about 15s and the status becomes COMPLETED. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
abdullahnah92
@@ -187,7 +187,11 @@ fn site_replicator_service_account_policy() -> S3Result<Policy> {
|
||||
"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<Vec<String>> = 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");
|
||||
|
||||
@@ -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"
|
||||
);
|
||||
}
|
||||
|
||||
@@ -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)),
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user