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): isolate ecstore owner contracts (#4239)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-03 21:42:45 +08:00
committed by GitHub
parent 92402a3bde
commit 6a81445a96
10 changed files with 55 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/bucket_target_sys.rs
+1 -2
View File
@@ -15,7 +15,7 @@
use crate::bucket::metadata::BucketMetadata;
use crate::bucket::metadata_sys::get_bucket_targets_config;
use crate::bucket::metadata_sys::get_replication_config;
use crate::bucket::replication::ReplicationTargetConfigBridge;
use crate::bucket::replication::{ReplicationStatusType, ReplicationTargetConfigBridge};
use crate::bucket::target::ARN;
use crate::bucket::target::BucketTargetType;
use crate::bucket::target::{self, BucketTarget, BucketTargets, Credentials};
@@ -49,7 +49,6 @@ use hyper_util::client::legacy::Client as HyperClient;
use hyper_util::rt::{TokioExecutor, TokioTimer};
use reqwest::Client as HttpClient;
use rustfs_config::{DEFAULT_TRUST_LEAF_CERT_AS_CA, ENV_TRUST_LEAF_CERT_AS_CA, RUSTFS_CA_CERT, RUSTFS_TLS_CERT};
use rustfs_replication::ReplicationStatusType;
use rustfs_utils::egress::{OutboundUrlError, validate_outbound_url};
use rustfs_utils::http::{
AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_LOCK_BYPASS_GOVERNANCE, AMZ_OBJECT_LOCK_LEGAL_HOLD, AMZ_OBJECT_LOCK_MODE,
crates/ecstore/src/client/object_handlers_common.rs
+1 -2
View File
@@ -21,14 +21,13 @@ const EVENT_LIFECYCLE_CLEANUP_SKIPPED: &str = "lifecycle_cleanup_skipped";
const EVENT_LIFECYCLE_CLEANUP_FAILED: &str = "lifecycle_cleanup_failed";
use crate::bucket::lifecycle::lifecycle;
use crate::bucket::replication::ReplicationLifecycleBridge;
use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationState};
use crate::bucket::versioning::VersioningApi;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::object_api::ObjectOptions;
use crate::storage_api_contracts::object::{ObjectOperations as _, ObjectToDelete};
use crate::store::ECStore;
use rustfs_lock::MAX_DELETE_LIST;
use rustfs_replication::ReplicationState;
pub async fn delete_object_versions(api: &Arc<ECStore>, bucket: &str, to_del: &[ObjectToDelete], _lc_event: lifecycle::Event) {
let version_suspended = match BucketVersioningSys::get(bucket).await {
crates/ecstore/src/core/pools.rs
+4 -3
View File
@@ -4081,6 +4081,7 @@ impl ECStore {
#[allow(clippy::items_after_test_module)]
mod tests {
use super::*;
use crate::bucket::replication::{ReplicationState, ReplicationStatusType};
use serde::Serialize;
#[test]
@@ -4235,8 +4236,8 @@ mod tests {
let mod_time = OffsetDateTime::now_utc();
let version = rustfs_filemeta::FileInfo {
mod_time: Some(mod_time),
replication_state_internal: Some(rustfs_replication::ReplicationState {
replica_status: rustfs_replication::ReplicationStatusType::Replica,
replication_state_internal: Some(ReplicationState {
replica_status: ReplicationStatusType::Replica,
delete_marker: true,
replicate_decision_str: "existing".to_string(),
..Default::default()
@@ -4254,7 +4255,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_replication::ReplicationStatusType::Replica);
assert_eq!(replication.replica_status, ReplicationStatusType::Replica);
assert!(replication.delete_marker);
assert_eq!(replication.replicate_decision_str, "existing");
}
crates/ecstore/src/data_movement/mod.rs
+7 -6
View File
@@ -929,6 +929,7 @@ pub(crate) async fn migrate_object(
#[cfg(test)]
mod tests {
use super::*;
use crate::bucket::replication::{ReplicationStatusType, VersionPurgeStatusType};
use rustfs_rio::HashReaderMut;
use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE};
use std::collections::HashMap;
@@ -1437,9 +1438,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_replication::ReplicationStatusType::Completed,
replication_status: ReplicationStatusType::Completed,
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
version_purge_status: rustfs_replication::VersionPurgeStatusType::Pending,
version_purge_status: VersionPurgeStatusType::Pending,
parts: Arc::new(vec![part]),
..Default::default()
};
@@ -1696,9 +1697,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_replication::ReplicationStatusType::Completed,
replication_status: ReplicationStatusType::Completed,
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
version_purge_status: rustfs_replication::VersionPurgeStatusType::Pending,
version_purge_status: VersionPurgeStatusType::Pending,
parts: Arc::new(vec![part]),
..Default::default()
}
@@ -1771,7 +1772,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_replication::VersionPurgeStatusType::Complete,
version_purge_status: VersionPurgeStatusType::Complete,
..source.clone()
};
@@ -1783,7 +1784,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_replication::ReplicationStatusType::Failed,
replication_status: ReplicationStatusType::Failed,
..source.clone()
};
crates/ecstore/src/object_api/mod.rs
+5 -5
View File
@@ -16,6 +16,10 @@
#![allow(dead_code)]
use crate::bucket::metadata_sys::get_versioning_config;
use crate::bucket::replication::{
ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map,
version_purge_statuses_map,
};
use crate::bucket::versioning::VersioningApi as _;
use crate::config::storageclass;
use crate::error::{Error, Result};
@@ -28,11 +32,7 @@ use crate::store::utils::clean_metadata;
use crate::{bucket::lifecycle::bucket_lifecycle_audit::LcAuditEvent, bucket::lifecycle::lifecycle::TransitionOptions};
use bytes::Bytes;
use http::{HeaderMap, HeaderValue};
use rustfs_filemeta::{
FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, ReplicateDecision, ReplicationState, ReplicationStatusType,
RestoreStatusOps as _, VersionPurgeStatusType, parse_restore_obj_status, replication_statuses_map,
version_purge_statuses_map,
};
use rustfs_filemeta::{FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, RestoreStatusOps as _, parse_restore_obj_status};
use rustfs_rio::Checksum;
use rustfs_utils::CompressionAlgorithm;
use rustfs_utils::http::headers::AMZ_OBJECT_TAGGING;
crates/ecstore/src/object_api/types.rs
-1
View File
@@ -791,7 +791,6 @@ fn versions_after_marker(file_infos: &rustfs_filemeta::FileInfoVersions, marker:
mod tests {
use super::*;
use rustfs_filemeta::{FileInfo, FileMeta, MetaCacheEntry, TRANSITION_COMPLETE};
use rustfs_replication::ReplicationState;
#[test]
fn versions_after_marker_handles_null_version_marker() {
crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs
+4 -3
View File
@@ -49,6 +49,7 @@ use super::{
DiskStat, GetObjectReader, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, RebalanceBucketOutcome,
RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats,
};
use crate::bucket::replication::{ReplicationState, ReplicationStatusType};
use crate::data_movement;
use crate::data_usage::DATA_USAGE_CACHE_NAME;
use crate::disk::RUSTFS_META_BUCKET;
@@ -222,8 +223,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_replication::ReplicationState {
replica_status: rustfs_replication::ReplicationStatusType::Replica,
replication_state_internal: Some(ReplicationState {
replica_status: ReplicationStatusType::Replica,
delete_marker: true,
replicate_decision_str: "existing".to_string(),
..Default::default()
@@ -241,7 +242,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_replication::ReplicationStatusType::Replica);
assert_eq!(replication.replica_status, ReplicationStatusType::Replica);
assert!(replication.delete_marker);
assert_eq!(replication.replicate_decision_str, "existing");
}
crates/ecstore/src/set_disk/mod.rs
+7 -6
View File
@@ -20,7 +20,9 @@
use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
use crate::bucket::metadata_sys;
use crate::bucket::object_lock::objectlock_sys::check_retention_for_modification;
use crate::bucket::replication::ReplicationObjectBridge;
use crate::bucket::replication::{
ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType,
};
use crate::bucket::versioning::VersioningApi;
use crate::bucket::versioning_sys::BucketVersioningSys;
use crate::client::{object_api_utils::get_raw_etag, transition_api::ReaderImpl};
@@ -98,8 +100,7 @@ use rustfs_common::heal_channel::{
use rustfs_config::MI_B;
use rustfs_filemeta::{
FileInfo, FileMeta, FileMetaShallowVersion, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, ObjectPartInfo,
RawFileInfo, ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, file_info_from_raw,
merge_file_meta_versions,
RawFileInfo, file_info_from_raw, merge_file_meta_versions,
};
use rustfs_io_metrics::{
record_object_lock_diag_acquire_duration, record_object_lock_diag_enabled, record_object_lock_diag_hold_duration,
@@ -7023,6 +7024,7 @@ pub fn is_infrequent_access_class(storage_class: &str) -> bool {
#[cfg(test)]
mod tests {
use super::*;
use crate::bucket::replication::{replication_statuses_map, version_purge_statuses_map};
use crate::disk::CHECK_PART_UNKNOWN;
use crate::disk::CHECK_PART_VOLUME_NOT_FOUND;
use crate::disk::DiskOption;
@@ -7048,7 +7050,6 @@ mod tests {
use rustfs_filemeta::MetaCacheEntry;
use rustfs_lock::client::local::LocalClient;
use rustfs_lock::{LockError, LockInfo, LockResponse, LockStats};
use rustfs_replication::ReplicationState;
use serial_test::serial;
use std::collections::HashMap;
use tempfile::TempDir;
@@ -7500,7 +7501,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_replication::replication_statuses_map("target=COMPLETED;"),
targets: replication_statuses_map("target=COMPLETED;"),
..Default::default()
}),
..Default::default()
@@ -7532,7 +7533,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_replication::version_purge_statuses_map("target=PENDING;"),
purge_targets: version_purge_statuses_map("target=PENDING;"),
..Default::default()
}),
..Default::default()
crates/ecstore/src/store/object.rs
+8 -5
View File
@@ -1346,6 +1346,9 @@ impl ECStore {
mod tests {
use super::*;
use crate::bucket::lifecycle::core::TRANSITION_COMPLETE;
use crate::bucket::replication::{
ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, version_purge_statuses_map,
};
use crate::layout::{
endpoints::{Endpoints, PoolEndpoints},
format::FormatV3,
@@ -1510,11 +1513,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_replication::ReplicationState {
replication_state_internal: Some(ReplicationState {
replication_status_internal: Some("arn:minio:replication:target=COMPLETED;".to_string()),
targets: rustfs_replication::replication_statuses_map("arn:minio:replication:target=COMPLETED;"),
targets: replication_statuses_map("arn:minio:replication:target=COMPLETED;"),
version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()),
purge_targets: rustfs_replication::version_purge_statuses_map("arn:minio:replication:target=PENDING;"),
purge_targets: version_purge_statuses_map("arn:minio:replication:target=PENDING;"),
..Default::default()
}),
metadata: HashMap::from([
@@ -1571,7 +1574,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_replication::ReplicationStatusType::Failed;
target.replication_status = ReplicationStatusType::Failed;
assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
}
@@ -1581,7 +1584,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_replication::VersionPurgeStatusType::Complete;
target.version_purge_status = VersionPurgeStatusType::Complete;
assert!(!is_equivalent_data_movement_tiered_object(&source, &target));
}
scripts/check_architecture_migration_rules.sh
+18
View File
@@ -137,6 +137,7 @@ LIFECYCLE_TAGGING_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/lifecycle_tagging_bounda
LIFECYCLE_OBJECT_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/lifecycle_object_lock_boundary_bypass_hits.txt"
LIFECYCLE_REPLICATION_SINK_BYPASS_HITS_FILE="${TMP_DIR}/lifecycle_replication_sink_bypass_hits.txt"
LIFECYCLE_REPLICATION_CRATE_BYPASS_HITS_FILE="${TMP_DIR}/lifecycle_replication_crate_bypass_hits.txt"
ECSTORE_REPLICATION_CONTRACT_BYPASS_HITS_FILE="${TMP_DIR}/ecstore_replication_contract_bypass_hits.txt"
STORE_API_EXTERNAL_LIST_CONSUMER_HITS_FILE="${TMP_DIR}/store_api_external_list_consumer_hits.txt"
STORE_API_EXTERNAL_OPERATION_CONSUMER_HITS_FILE="${TMP_DIR}/store_api_external_operation_consumer_hits.txt"
STORE_API_OBJECT_OPERATION_LOCAL_METHOD_HITS_FILE="${TMP_DIR}/store_api_object_operation_local_method_hits.txt"
@@ -954,6 +955,23 @@ if [[ -s "$LIFECYCLE_REPLICATION_CRATE_BYPASS_HITS_FILE" ]]; then
report_failure "lifecycle replication status/state contracts must stay behind lifecycle replication_sink: $(paste -sd '; ' "$LIFECYCLE_REPLICATION_CRATE_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
{
rg -n --with-filename 'rustfs_replication::|use\s+rustfs_replication\b' \
crates/ecstore/src \
--glob '*.rs' |
rg -v '^crates/ecstore/src/bucket/replication/' || true
rg -n -U --with-filename 'use\s+rustfs_filemeta::\{[^}]*\b(ReplicateDecision|ReplicationState|ReplicationStatusType|VersionPurgeStatusType|replication_statuses_map|version_purge_statuses_map)\b|use\s+rustfs_filemeta::(ReplicateDecision|ReplicationState|ReplicationStatusType|VersionPurgeStatusType|replication_statuses_map|version_purge_statuses_map)\b|rustfs_filemeta::(Replicate|Replication|VersionPurge|replication_statuses_map|version_purge_statuses_map)' \
crates/ecstore/src \
--glob '*.rs' || true
}
) >"$ECSTORE_REPLICATION_CONTRACT_BYPASS_HITS_FILE"
if [[ -s "$ECSTORE_REPLICATION_CONTRACT_BYPASS_HITS_FILE" ]]; then
report_failure "ECStore owner replication contracts must stay behind crates/ecstore/src/bucket/replication: $(paste -sd '; ' "$ECSTORE_REPLICATION_CONTRACT_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --no-heading 'rustfs_ecstore::store_api(?:::\{[^}]*\b(?:ListObjectVersionsInfo|ListObjectsV2Info|ObjectInfoOrErr)\b|::(?:ListObjectVersionsInfo|ListObjectsV2Info|ObjectInfoOrErr)\b)' \