gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-11 07:36:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(replication): own filemeta wire contracts (#4254)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-04 06:39:54 +08:00
committed by GitHub
parent 09fcacbe2b
commit 3d4fb532e5
22 changed files with 1196 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/app/object_usecase.rs
+15 -15
View File
@@ -41,9 +41,10 @@ use super::storage_api::object_usecase::bucket::{
quota::QuotaOperation,
replication::{
REPLICATE_INCOMING_DELETE, ReplicationStatusType, VersionPurgeStatusType, check_replicate_delete,
delete_replication_state_from_config, delete_replication_version_id, must_replicate_object, schedule_object_replication,
schedule_replication_delete, should_schedule_delete_replication, should_use_existing_delete_replication_info,
should_use_existing_delete_replication_source,
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,
},
tagging::decode_tags,
validate_restore_request,
@@ -1528,7 +1529,7 @@ async fn enrich_delete_replication_state_if_needed(
version_id,
obj_info.replication_status == ReplicationStatusType::Replica,
) {
delete_object.replication_state = Some(local_state);
set_deleted_object_replication_state(delete_object, &local_state);
}
}
@@ -4701,7 +4702,7 @@ impl DefaultObjectUsecase {
.await;
if dsc.replicate_any() {
if object.version_id.is_some() {
object.version_purge_status = Some(VersionPurgeStatusType::Pending);
set_object_to_delete_version_purge_status(&mut object, VersionPurgeStatusType::Pending);
object.version_purge_statuses = dsc.pending_status();
} else {
object.delete_marker_replication_status = dsc.pending_status();
@@ -4843,8 +4844,7 @@ impl DefaultObjectUsecase {
for dobjs in &delete_results {
if let Some(dobj) = &dobjs.delete_object
&& replicate_deletes
&& (dobj.delete_marker_replication_status() == ReplicationStatusType::Pending
|| dobj.version_purge_status() == VersionPurgeStatusType::Pending)
&& deleted_object_has_pending_replication_delete(dobj)
{
let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Replication);
let mut dobj = dobj.clone();
@@ -5098,9 +5098,10 @@ impl DefaultObjectUsecase {
deleted_object_source.version_id
},
delete_marker_mtime: deleted_object_source.mod_time,
replication_state: Some(replication_state_source.replication_state()),
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;
schedule_replication_delete(deleted_object, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
}
@@ -8463,15 +8464,14 @@ mod tests {
#[test]
fn replica_delete_enrichment_must_not_reuse_upstream_targets() {
let delete_object = StorageDeletedObject {
replication_state: Some(ReplicationState {
replicate_decision_str: "arn:aws:s3:::upstream=true;false;arn:aws:s3:::upstream;".to_string(),
replication_status_internal: Some("arn:aws:s3:::upstream=COMPLETED;".to_string()),
targets: replication_statuses_map("arn:aws:s3:::upstream=COMPLETED;"),
..Default::default()
}),
let upstream_state = ReplicationState {
replicate_decision_str: "arn:aws:s3:::upstream=true;false;arn:aws:s3:::upstream;".to_string(),
replication_status_internal: Some("arn:aws:s3:::upstream=COMPLETED;".to_string()),
targets: replication_statuses_map("arn:aws:s3:::upstream=COMPLETED;"),
..Default::default()
};
let mut delete_object = StorageDeletedObject::default();
set_deleted_object_replication_state(&mut delete_object, &upstream_state);
let obj_info = ObjectInfo {
replication_status: ReplicationStatusType::Replica,
..Default::default()
rustfs/src/app/storage_api.rs
+23
View File
@@ -677,6 +677,29 @@ pub(crate) mod bucket {
ReplicationObjectBridge::schedule_storage_delete(delete_object, bucket, event_type).await;
}
pub(crate) fn set_deleted_object_replication_state(
delete_object: &mut crate::storage::storage_api::StorageDeletedObject,
state: &replication_contracts::ReplicationState,
) {
delete_object.replication_state = Some(replication_contracts::replication_state_to_filemeta(state));
}
pub(crate) fn set_object_to_delete_version_purge_status(
object: &mut crate::storage::storage_api::StorageObjectToDelete,
status: VersionPurgeStatusType,
) {
object.version_purge_status = Some(replication_contracts::version_purge_status_to_filemeta(status));
}
pub(crate) fn deleted_object_has_pending_replication_delete(
deleted_object: &crate::storage::storage_api::StorageDeletedObject,
) -> bool {
deleted_object.delete_marker_replication_status()
== replication_contracts::replication_status_to_filemeta(ReplicationStatusType::Pending)
|| deleted_object.version_purge_status()
== replication_contracts::version_purge_status_to_filemeta(VersionPurgeStatusType::Pending)
}
pub(crate) fn delete_replication_state_from_config(
config: &s3s::dto::ReplicationConfiguration,
obj_info: &crate::storage::storage_api::StorageObjectInfo,