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

refactor(replication): route filemeta paths through crate (#4161)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-02 09:08:47 +08:00
committed by GitHub
parent 7a075c91da
commit 18b79ccc2a
9 changed files with 53 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs
+9 -9
View File
@@ -2874,12 +2874,12 @@ fn should_reuse_lifecycle_delete_replication_state(oi: &ObjectInfo, version_dele
if version_delete {
oi.version_purge_status == VersionPurgeStatusType::Pending && !state.purge_targets.is_empty()
} else {
oi.replication_status == rustfs_filemeta::ReplicationStatusType::Pending && !state.targets.is_empty()
oi.replication_status == rustfs_replication::ReplicationStatusType::Pending && !state.targets.is_empty()
}
}
fn lifecycle_version_purge_state_from_completed_targets(oi: &ObjectInfo) -> Option<ReplicationState> {
if oi.replication_status != rustfs_filemeta::ReplicationStatusType::Completed {
if oi.replication_status != rustfs_replication::ReplicationStatusType::Completed {
return None;
}
@@ -2893,7 +2893,7 @@ fn lifecycle_version_purge_state_from_completed_targets(oi: &ObjectInfo) -> Opti
Some(ReplicationState {
replicate_decision_str: oi.replication_decision.clone(),
version_purge_status_internal: Some(pending_status.clone()),
purge_targets: rustfs_filemeta::version_purge_statuses_map(&pending_status),
purge_targets: rustfs_replication::version_purge_statuses_map(&pending_status),
..Default::default()
})
}
@@ -2939,10 +2939,10 @@ fn replication_state_for_delete(dsc: ReplicateDecision, version_delete: bool) ->
};
if version_delete {
state.version_purge_status_internal = pending_status.clone();
state.purge_targets = rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default());
state.purge_targets = rustfs_replication::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default());
} else {
state.replication_status_internal = pending_status.clone();
state.targets = rustfs_filemeta::replication_statuses_map(pending_status.as_deref().unwrap_or_default());
state.targets = rustfs_replication::replication_statuses_map(pending_status.as_deref().unwrap_or_default());
}
state
}
@@ -3856,7 +3856,7 @@ mod tests {
fn replication_state_for_delete_uses_replication_targets_for_current_delete() {
let arn = "arn:aws:s3:::target-bucket";
let mut dsc = ReplicateDecision::default();
dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false));
dsc.set(rustfs_replication::ReplicateTargetDecision::new(arn.to_string(), true, false));
let state = replication_state_for_delete(dsc, false);
@@ -3869,7 +3869,7 @@ mod tests {
fn replication_state_for_delete_uses_purge_targets_for_version_delete() {
let arn = "arn:aws:s3:::target-bucket";
let mut dsc = ReplicateDecision::default();
dsc.set(rustfs_filemeta::ReplicateTargetDecision::new(arn.to_string(), true, false));
dsc.set(rustfs_replication::ReplicateTargetDecision::new(arn.to_string(), true, false));
let state = replication_state_for_delete(dsc, true);
@@ -3894,7 +3894,7 @@ mod tests {
#[test]
fn lifecycle_delete_replication_state_does_not_reuse_put_replication_for_version_delete() {
let oi = ObjectInfo {
replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
replication_status: rustfs_replication::ReplicationStatusType::Completed,
replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()),
replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
..Default::default()
@@ -3909,7 +3909,7 @@ mod tests {
#[test]
fn lifecycle_version_purge_state_from_completed_targets_derives_pending_purge_targets() {
let oi = ObjectInfo {
replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
replication_status: rustfs_replication::ReplicationStatusType::Completed,
replication_status_internal: Some("arn:aws:s3:::target=COMPLETED;".to_string()),
replication_decision: "arn:aws:s3:::target=true;false;arn:aws:s3:::target;".to_string(),
..Default::default()
crates/ecstore/src/core/pools.rs
+3 -3
View File
@@ -4200,8 +4200,8 @@ mod tests {
let mod_time = OffsetDateTime::now_utc();
let version = rustfs_filemeta::FileInfo {
mod_time: Some(mod_time),
replication_state_internal: Some(rustfs_filemeta::ReplicationState {
replica_status: rustfs_filemeta::ReplicationStatusType::Replica,
replication_state_internal: Some(rustfs_replication::ReplicationState {
replica_status: rustfs_replication::ReplicationStatusType::Replica,
delete_marker: true,
replicate_decision_str: "existing".to_string(),
..Default::default()
@@ -4219,7 +4219,7 @@ mod tests {
assert_eq!(opts.src_pool_idx, 7);
assert_eq!(opts.version_id.as_deref(), Some("version-id"));
assert_eq!(opts.mod_time, Some(mod_time));
assert_eq!(replication.replica_status, rustfs_filemeta::ReplicationStatusType::Replica);
assert_eq!(replication.replica_status, rustfs_replication::ReplicationStatusType::Replica);
assert!(replication.delete_marker);
assert_eq!(replication.replicate_decision_str, "existing");
}
crates/ecstore/src/data_movement/mod.rs
+6 -6
View File
@@ -1348,9 +1348,9 @@ mod tests {
storage_class: Some("STANDARD_IA".to_string()),
checksum: Some(Bytes::from_static(b"object-checksum")),
replication_status_internal: Some("arn:minio:replication:target=COMPLETED;".to_string()),
replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
replication_status: rustfs_replication::ReplicationStatusType::Completed,
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
version_purge_status: rustfs_filemeta::VersionPurgeStatusType::Pending,
version_purge_status: rustfs_replication::VersionPurgeStatusType::Pending,
parts: Arc::new(vec![part]),
..Default::default()
};
@@ -1607,9 +1607,9 @@ mod tests {
expires: Some(OffsetDateTime::from_unix_timestamp(2_000).expect("valid expires timestamp")),
storage_class: Some("STANDARD_IA".to_string()),
replication_status_internal: Some("arn:minio:replication:target=COMPLETED;".to_string()),
replication_status: rustfs_filemeta::ReplicationStatusType::Completed,
replication_status: rustfs_replication::ReplicationStatusType::Completed,
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
version_purge_status: rustfs_filemeta::VersionPurgeStatusType::Pending,
version_purge_status: rustfs_replication::VersionPurgeStatusType::Pending,
parts: Arc::new(vec![part]),
..Default::default()
}
@@ -1682,7 +1682,7 @@ mod tests {
let source = overwrite_equivalence_source();
let target = ObjectInfo {
version_purge_status_internal: Some("arn:minio:replication:target=COMPLETE;".to_string()),
version_purge_status: rustfs_filemeta::VersionPurgeStatusType::Complete,
version_purge_status: rustfs_replication::VersionPurgeStatusType::Complete,
..source.clone()
};
@@ -1694,7 +1694,7 @@ mod tests {
let source = overwrite_equivalence_source();
let target = ObjectInfo {
replication_status_internal: Some("arn:minio:replication:target=FAILED;".to_string()),
replication_status: rustfs_filemeta::ReplicationStatusType::Failed,
replication_status: rustfs_replication::ReplicationStatusType::Failed,
..source.clone()
};
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+3 -3
View File
@@ -222,8 +222,8 @@ fn test_rebalance_delete_marker_opts_preserves_replication_state() {
let mod_time = OffsetDateTime::now_utc();
let version = FileInfo {
mod_time: Some(mod_time),
replication_state_internal: Some(rustfs_filemeta::ReplicationState {
replica_status: rustfs_filemeta::ReplicationStatusType::Replica,
replication_state_internal: Some(rustfs_replication::ReplicationState {
replica_status: rustfs_replication::ReplicationStatusType::Replica,
delete_marker: true,
replicate_decision_str: "existing".to_string(),
..Default::default()
@@ -241,7 +241,7 @@ fn test_rebalance_delete_marker_opts_preserves_replication_state() {
assert_eq!(opts.src_pool_idx, 7);
assert_eq!(opts.version_id.as_deref(), Some("version-id"));
assert_eq!(opts.mod_time, Some(mod_time));
assert_eq!(replication.replica_status, rustfs_filemeta::ReplicationStatusType::Replica);
assert_eq!(replication.replica_status, rustfs_replication::ReplicationStatusType::Replica);
assert!(replication.delete_marker);
assert_eq!(replication.replicate_decision_str, "existing");
}
crates/ecstore/src/set_disk/mod.rs
+2 -2
View File
@@ -7407,7 +7407,7 @@ mod tests {
delete_replication: Some(ReplicationState {
replicate_decision_str: "target=true;false;target;".to_string(),
replication_status_internal: Some("target=COMPLETED;".to_string()),
targets: rustfs_filemeta::replication_statuses_map("target=COMPLETED;"),
targets: rustfs_replication::replication_statuses_map("target=COMPLETED;"),
..Default::default()
}),
..Default::default()
@@ -7439,7 +7439,7 @@ mod tests {
version_id: Some(Uuid::new_v4().to_string()),
delete_replication: Some(ReplicationState {
version_purge_status_internal: Some("target=PENDING;".to_string()),
purge_targets: rustfs_filemeta::version_purge_statuses_map("target=PENDING;"),
purge_targets: rustfs_replication::version_purge_statuses_map("target=PENDING;"),
..Default::default()
}),
..Default::default()
crates/ecstore/src/store/object.rs
+5 -5
View File
@@ -1466,11 +1466,11 @@ mod tests {
transitioned_objname: "remote/object".to_string(),
transition_tier: "WARM".to_string(),
transition_version_id: Some(transition_version_id),
replication_state_internal: Some(rustfs_filemeta::ReplicationState {
replication_state_internal: Some(rustfs_replication::ReplicationState {
replication_status_internal: Some("arn:minio:replication:target=COMPLETED;".to_string()),
targets: rustfs_filemeta::replication_statuses_map("arn:minio:replication:target=COMPLETED;"),
targets: rustfs_replication::replication_statuses_map("arn:minio:replication:target=COMPLETED;"),
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
purge_targets: rustfs_filemeta::version_purge_statuses_map("arn:minio:replication:target=PENDING;"),
purge_targets: rustfs_replication::version_purge_statuses_map("arn:minio:replication:target=PENDING;"),
..Default::default()
}),
metadata: HashMap::from([
@@ -1527,7 +1527,7 @@ mod tests {
let source = tiered_equivalence_source();
let mut target = tiered_equivalence_target(&source);
target.replication_status_internal = Some("arn:minio:replication:target=FAILED;".to_string());
target.replication_status = rustfs_filemeta::ReplicationStatusType::Failed;
target.replication_status = rustfs_replication::ReplicationStatusType::Failed;
assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
}
@@ -1537,7 +1537,7 @@ mod tests {
let source = tiered_equivalence_source();
let mut target = tiered_equivalence_target(&source);
target.version_purge_status_internal = Some("arn:minio:replication:target=COMPLETE;".to_string());
target.version_purge_status = rustfs_filemeta::VersionPurgeStatusType::Complete;
target.version_purge_status = rustfs_replication::VersionPurgeStatusType::Complete;
assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
}
rustfs/src/admin/storage_api.rs
+1 -1
View File
@@ -150,7 +150,7 @@ pub(crate) trait AdminReplicationConfigExt {
impl AdminReplicationConfigExt for s3s::dto::ReplicationConfiguration {
fn filter_all_replication_target_arns(&self) -> Vec<String> {
let obj = ecstore_bucket::replication::ObjectOpts {
op_type: rustfs_filemeta::ReplicationType::All,
op_type: rustfs_replication::ReplicationType::All,
..Default::default()
};
<s3s::dto::ReplicationConfiguration as ecstore_bucket::replication::ReplicationConfigurationExt>::filter_target_arns(
rustfs/src/app/storage_api.rs
+11 -10
View File
@@ -612,7 +612,7 @@ pub(crate) mod bucket {
crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo;
type ObjectOpts = crate::storage::storage_api::ecstore_bucket::replication::ObjectOpts;
type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
pub(crate) type ReplicateDecision = rustfs_filemeta::ReplicateDecision;
pub(crate) type ReplicateDecision = rustfs_replication::ReplicateDecision;
pub(crate) async fn check_replicate_delete(
bucket: &str,
@@ -629,14 +629,14 @@ pub(crate) mod bucket {
object: &str,
user_defined: &HashMap<String, String>,
user_tags: String,
status: rustfs_filemeta::ReplicationStatusType,
status: rustfs_replication::ReplicationStatusType,
opts: crate::storage::storage_api::StorageObjectOptions,
) -> ReplicateDecision {
let mopts = ReplicationObjectBridge::must_replicate_options(
user_defined,
user_tags,
status,
rustfs_filemeta::ReplicationType::Object,
rustfs_replication::ReplicationType::Object,
opts,
);
ReplicationObjectBridge::must_replicate(bucket, object, mopts).await
@@ -647,7 +647,7 @@ pub(crate) mod bucket {
store: Arc<crate::storage::storage_api::ECStore>,
dsc: ReplicateDecision,
) {
ReplicationObjectBridge::schedule_object(oi, store, dsc, rustfs_filemeta::ReplicationType::Object).await;
ReplicationObjectBridge::schedule_object(oi, store, dsc, rustfs_replication::ReplicationType::Object).await;
}
pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
@@ -659,13 +659,13 @@ pub(crate) mod bucket {
obj_info: &crate::storage::storage_api::StorageObjectInfo,
version_id: Option<Uuid>,
replica: bool,
) -> Option<rustfs_filemeta::ReplicationState> {
) -> Option<rustfs_replication::ReplicationState> {
let opts = ObjectOpts {
name: obj_info.name.clone(),
user_tags: (*obj_info.user_tags).clone(),
version_id,
delete_marker: obj_info.delete_marker,
op_type: rustfs_filemeta::ReplicationType::Delete,
op_type: rustfs_replication::ReplicationType::Delete,
replica,
..Default::default()
};
@@ -685,23 +685,24 @@ pub(crate) mod bucket {
config,
&target_opts,
);
decision.set(rustfs_filemeta::ReplicateTargetDecision::new(target_arn, replicate, false));
decision.set(rustfs_replication::ReplicateTargetDecision::new(target_arn, replicate, false));
}
if !decision.replicate_any() {
return None;
}
let pending_status = decision.pending_status();
let mut state = rustfs_filemeta::ReplicationState {
let mut state = rustfs_replication::ReplicationState {
replicate_decision_str: decision.to_string(),
..Default::default()
};
if version_id.is_some() {
state.version_purge_status_internal = pending_status.clone();
state.purge_targets = rustfs_filemeta::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default());
state.purge_targets =
rustfs_replication::version_purge_statuses_map(pending_status.as_deref().unwrap_or_default());
} else {
state.replication_status_internal = pending_status.clone();
state.targets = rustfs_filemeta::replication_statuses_map(pending_status.as_deref().unwrap_or_default());
state.targets = rustfs_replication::replication_statuses_map(pending_status.as_deref().unwrap_or_default());
}
Some(state)
}
scripts/check_architecture_migration_rules.sh
+13
View File
@@ -212,6 +212,7 @@ REPLICATION_ERROR_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_error_bounda
REPLICATION_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypass_hits.txt"
REPLICATION_EVENT_HOST_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_host_bypass_hits.txt"
REPLICATION_FILEMETA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_filemeta_boundary_bypass_hits.txt"
REPLICATION_FILEMETA_DIRECT_PATH_BYPASS_HITS_FILE="${TMP_DIR}/replication_filemeta_direct_path_bypass_hits.txt"
REPLICATION_LOCAL_PATH_BYPASS_HITS_FILE="${TMP_DIR}/replication_local_path_bypass_hits.txt"
REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_lock_boundary_bypass_hits.txt"
REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt"
@@ -3034,6 +3035,18 @@ if [[ -s "$REPLICATION_FILEMETA_BOUNDARY_BYPASS_HITS_FILE" ]]; then
report_failure "replication filemeta contracts must stay behind replication filemeta boundary: $(paste -sd '; ' "$REPLICATION_FILEMETA_BOUNDARY_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'rustfs_filemeta::(Replicate|Replication|VersionPurge|replication_statuses_map|version_purge_statuses_map|get_replication_state|parse_replicate_decision|target_reset_header|REPLICATE_)' \
crates/ecstore/src rustfs/src crates/scanner/src \
--glob '*.rs' |
rg -v '^crates/ecstore/src/bucket/replication/replication_filemeta_boundary\.rs:' || true
) >"$REPLICATION_FILEMETA_DIRECT_PATH_BYPASS_HITS_FILE"
if [[ -s "$REPLICATION_FILEMETA_DIRECT_PATH_BYPASS_HITS_FILE" ]]; then
report_failure "replication filemeta fully-qualified paths must use rustfs_replication: $(paste -sd '; ' "$REPLICATION_FILEMETA_DIRECT_PATH_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |