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

refactor(replication): own delete DTO contracts (#4253)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-04 04:00:27 +08:00
committed by GitHub
parent 8a0617865b
commit 125c228832
14 changed files with 159 additions and 74 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
-1
View File
@@ -9620,7 +9620,6 @@ dependencies = [
"rmp",
"rmp-serde",
"rustfs-filemeta",
"rustfs-storage-api",
"rustfs-utils",
"s3s",
"serde",
crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs
+4 -2
View File
@@ -20,7 +20,9 @@ use super::replication_filemeta_boundary::{
};
use super::replication_object_config::{ReplicationConfig, check_replicate_delete};
use super::replication_queue_boundary::DeletedObjectReplicationInfo;
use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete};
use super::replication_storage_boundary::{
DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
};
pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
@@ -64,7 +66,7 @@ impl ReplicationLifecycleBridge {
pub(crate) async fn schedule_delete(bucket: String, delete_object: DeletedObject) {
super::replication_pool::schedule_replication_delete(DeletedObjectReplicationInfo {
delete_object,
delete_object: deleted_object_for_replication(delete_object),
bucket,
event_type: REPLICATE_INCOMING_DELETE.to_string(),
..Default::default()
crates/ecstore/src/bucket/replication/replication_object_bridge.rs
+13 -1
View File
@@ -19,7 +19,9 @@ use super::replication_object_config::{check_replicate_delete, get_must_replicat
use super::replication_object_decision_boundary::MustReplicateOptions;
use super::replication_pool::{schedule_replication, schedule_replication_delete};
use super::replication_queue_boundary::DeletedObjectReplicationInfo;
use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage};
use super::replication_storage_boundary::{
DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, ReplicationStorage, deleted_object_for_replication,
};
pub struct ReplicationObjectBridge;
@@ -60,6 +62,16 @@ impl ReplicationObjectBridge {
pub async fn schedule_delete(delete_object: DeletedObjectReplicationInfo) {
schedule_replication_delete(delete_object).await;
}
pub async fn schedule_storage_delete(delete_object: DeletedObject, bucket: String, event_type: String) {
Self::schedule_delete(DeletedObjectReplicationInfo {
delete_object: deleted_object_for_replication(delete_object),
bucket,
event_type,
..Default::default()
})
.await;
}
}
#[cfg(test)]
crates/ecstore/src/bucket/replication/replication_object_config.rs
+3 -2
View File
@@ -30,7 +30,7 @@ use super::replication_object_decision_boundary::{
MustReplicateOptions, ReplicationDeleteSource, ReplicationResyncTargetObject, delete_replication_missing_source_decision,
delete_replication_object_opts, resync_target_for_object,
};
use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete};
use super::replication_storage_boundary::{ObjectInfo, ObjectOptions, ObjectToDelete, object_to_delete_for_replication};
use super::replication_target_boundary::{BucketTargets, ReplicationTargetStore};
use super::replication_versioning_boundary::ReplicationVersioningStore;
use super::runtime_boundary as runtime_sources;
@@ -194,8 +194,9 @@ pub(crate) async fn check_replicate_delete(
return ReplicateDecision::default();
}
let replication_delete = object_to_delete_for_replication(dobj);
let opts = delete_replication_object_opts(
dobj,
&replication_delete,
&ReplicationDeleteSource {
user_defined: oi.user_defined.as_ref(),
user_tags: oi.user_tags.as_str(),
crates/ecstore/src/bucket/replication/replication_pool.rs
+4 -2
View File
@@ -38,7 +38,9 @@ use super::replication_resyncer::{
ReplicationResyncer, get_heal_replicate_object_info, replicate_delete, replicate_object, save_resync_status,
};
use super::replication_state::ReplicationStats;
use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage};
use super::replication_storage_boundary::{
ObjectInfo, ObjectOptions, ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage,
};
use super::replication_target_boundary::ReplicationTargetStore;
use super::runtime_boundary as runtime_sources;
use lazy_static::lazy_static;
@@ -616,7 +618,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
// get_object_info here because the delete-marker or version may
// already be absent from the local store — that is expected.
let dv = DeletedObjectReplicationInfo {
delete_object: DeletedObject {
delete_object: ReplicationDeletedObject {
object_name: entry.object.clone(),
version_id: entry.version_id,
delete_marker_version_id: entry.delete_marker_version_id,
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+7 -7
View File
@@ -42,8 +42,8 @@ use super::replication_resync_boundary::{
#[cfg(test)]
use super::replication_resync_boundary::{RESYNC_META_FORMAT, RESYNC_META_VERSION, WIRE_ZERO_TIME_UNIX, decode_resync_file};
use super::replication_storage_boundary::{
AdvancedGetOptions, DeletedObject, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete,
ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions,
AdvancedGetOptions, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete,
ReplicationDeletedObject, ReplicationObjectIO, ReplicationStorage, StatObjectOptions, WalkOptions,
};
use super::replication_target_boundary::{
PutObjectOptions, PutObjectPartOptions, ReplicationTargetStore, TargetClient, replication_action_for_target_head,
@@ -671,7 +671,7 @@ impl ReplicationResyncer {
};
let doi = DeletedObjectReplicationInfo {
delete_object: DeletedObject {
delete_object: ReplicationDeletedObject {
object_name: roi.name.clone(),
delete_marker_version_id: dm_version_id,
version_id,
@@ -3131,7 +3131,7 @@ mod tests {
#[test]
fn test_is_version_delete_replication_for_delete_marker_version_purge() {
let dobj = DeletedObject {
let dobj = ReplicationDeletedObject {
delete_marker: false,
delete_marker_version_id: Some(Uuid::new_v4()),
..Default::default()
@@ -3145,7 +3145,7 @@ mod tests {
#[test]
fn test_is_version_delete_replication_for_delete_marker_creation() {
let dobj = DeletedObject {
let dobj = ReplicationDeletedObject {
delete_marker: true,
delete_marker_version_id: Some(Uuid::new_v4()),
..Default::default()
@@ -3159,7 +3159,7 @@ mod tests {
#[test]
fn test_should_retry_delete_marker_purge_for_version_purge() {
let dobj = DeletedObject {
let dobj = ReplicationDeletedObject {
delete_marker: false,
delete_marker_version_id: Some(Uuid::new_v4()),
..Default::default()
@@ -3173,7 +3173,7 @@ mod tests {
#[test]
fn test_should_retry_delete_marker_purge_for_delete_marker_creation() {
let dobj = DeletedObject {
let dobj = ReplicationDeletedObject {
delete_marker: true,
delete_marker_version_id: Some(Uuid::new_v4()),
..Default::default()
crates/ecstore/src/bucket/replication/replication_storage_boundary.rs
+25
View File
@@ -27,6 +27,7 @@ pub(crate) use crate::storage_api_contracts::object::{
DeletedObject, EcstoreObjectOperations, ObjectIO, ObjectOperations, ObjectToDelete,
};
pub(crate) use crate::storage_api_contracts::range::HTTPRangeSpec;
pub(crate) use rustfs_replication::{DeletedObject as ReplicationDeletedObject, ObjectToDelete as ReplicationObjectToDelete};
type ListObjectsV2Info = StorageListObjectsV2Info<ObjectInfo>;
type ListObjectVersionsInfo = StorageListObjectVersionsInfo<ObjectInfo>;
@@ -93,6 +94,30 @@ pub trait ReplicationStorage:
{
}
pub(crate) fn deleted_object_for_replication(delete_object: DeletedObject) -> ReplicationDeletedObject {
ReplicationDeletedObject {
delete_marker: delete_object.delete_marker,
delete_marker_version_id: delete_object.delete_marker_version_id,
object_name: delete_object.object_name,
version_id: delete_object.version_id,
delete_marker_mtime: delete_object.delete_marker_mtime,
replication_state: delete_object.replication_state,
found: delete_object.found,
force_delete: delete_object.force_delete,
}
}
pub(crate) fn object_to_delete_for_replication(object: &ObjectToDelete) -> ReplicationObjectToDelete {
ReplicationObjectToDelete {
object_name: object.object_name.clone(),
version_id: object.version_id,
delete_marker_replication_status: object.delete_marker_replication_status.clone(),
version_purge_status: object.version_purge_status.clone(),
version_purge_statuses: object.version_purge_statuses.clone(),
replicate_decision_str: object.replicate_decision_str.clone(),
}
}
impl<T> ReplicationStorage for T where
T: ObjectIO<
Error = Error,
crates/replication/Cargo.toml
-1
View File
@@ -30,7 +30,6 @@ byteorder.workspace = true
rmp.workspace = true
rmp-serde.workspace = true
rustfs-filemeta.workspace = true
rustfs-storage-api.workspace = true
rustfs-utils = { workspace = true, features = ["http", "path", "string"] }
s3s.workspace = true
serde.workspace = true
crates/replication/src/lib.rs
+1
View File
@@ -82,4 +82,5 @@ pub use stats::{
FailedMetric, InQueueMetric, InQueueStats, LatencyStats, ProxyMetric, ProxyStatsCache, QueueCache, QueueNode, QueueStats,
SRMetricsSummary, XferStats,
};
pub use storage_api::{DeletedObject, ObjectToDelete};
pub use tagging::{ReplicationTagFilter, decode_tags_to_map};
crates/replication/src/storage_api.rs
+54 -1
View File
@@ -12,4 +12,57 @@
// See the License for the specific language governing permissions and
// limitations under the License.
pub(crate) use rustfs_storage_api::{DeletedObject, ObjectToDelete};
use crate::{ReplicationState, ReplicationStatusType, VersionPurgeStatusType};
use time::OffsetDateTime;
use uuid::Uuid;
#[derive(Debug, Default, Clone)]
pub struct ObjectToDelete {
pub object_name: String,
pub version_id: Option<Uuid>,
pub delete_marker_replication_status: Option<String>,
pub version_purge_status: Option<VersionPurgeStatusType>,
pub version_purge_statuses: Option<String>,
pub replicate_decision_str: Option<String>,
}
impl ObjectToDelete {
pub fn replication_state(&self) -> ReplicationState {
ReplicationState {
replication_status_internal: self.delete_marker_replication_status.clone(),
version_purge_status_internal: self.version_purge_statuses.clone(),
replicate_decision_str: self.replicate_decision_str.clone().unwrap_or_default(),
targets: crate::replication_statuses_map(self.delete_marker_replication_status.as_deref().unwrap_or_default()),
purge_targets: crate::version_purge_statuses_map(self.version_purge_statuses.as_deref().unwrap_or_default()),
..Default::default()
}
}
}
#[derive(Debug, Default, Clone)]
pub struct DeletedObject {
pub delete_marker: bool,
pub delete_marker_version_id: Option<Uuid>,
pub object_name: String,
pub version_id: Option<Uuid>,
pub delete_marker_mtime: Option<OffsetDateTime>,
pub replication_state: Option<ReplicationState>,
pub found: bool,
pub force_delete: bool,
}
impl DeletedObject {
pub fn version_purge_status(&self) -> VersionPurgeStatusType {
self.replication_state
.as_ref()
.map(|v| v.composite_version_purge_status())
.unwrap_or(VersionPurgeStatusType::Empty)
}
pub fn delete_marker_replication_status(&self) -> ReplicationStatusType {
self.replication_state
.as_ref()
.map(|v| v.composite_replication_status())
.unwrap_or(ReplicationStatusType::Empty)
}
}
docs/architecture/ecstore-module-split-plan.md
+11 -10
View File
@@ -105,10 +105,9 @@ ready for a full standalone crate yet.
Current coupling:
- replication workers depend on `ReplicationStorage`, ECStore object APIs and
storage-api contracts through the replication storage boundary, bucket target
clients, bucket metadata, file metadata replication state through the
filemeta boundary, config-derived storage class labels through the config
store, scanner repair
owner storage-api contracts through the replication storage boundary, bucket
target clients, bucket metadata, file metadata replication state through the
filemeta boundary, config-derived storage class labels through the config store, scanner repair
classification, runtime replication pool/stat handles, bucket monitor and
bandwidth reader access through local boundaries, local node names, and
notification events;
@@ -127,8 +126,10 @@ Current coupling:
ECStore retaining only error mapping and MRF persistence locally;
- `crates/replication/src/filemeta.rs` is the only direct filemeta wire-contract
import boundary inside `rustfs-replication`;
- `crates/replication/src/storage_api.rs` is the only direct storage-api delete
DTO import boundary inside `rustfs-replication`;
- `ReplicationCrateStorageApiIndependence`: delete work DTOs are owned inside
`rustfs-replication`; ECStore converts storage-api delete DTOs at the
replication storage boundary instead of `rustfs-replication` importing
`rustfs-storage-api`;
- direct ECStore replication imports from `rustfs-replication` are limited to
`*_boundary.rs` modules;
- storage-api delete replication status/state helpers use the local
@@ -192,10 +193,10 @@ Required contracts before crate movement:
`rustfs-ecstore`; runtime code under `rustfs/src` does not import
`rustfs-replication` directly, and the RustFS runtime/scanner crates do not
depend on it.
- `StorageApiReplicationContracts`: storage-api delete DTO replication
state/status helpers are concentrated in `crates/storage-api/src/replication.rs`
until the underlying wire contracts can move without a
`rustfs-replication` / `rustfs-storage-api` dependency cycle.
- `StorageApiReplicationContracts`: owner-facing storage-api delete DTO
replication state/status helpers remain concentrated in
`crates/storage-api/src/replication.rs`, while replication worker DTOs live in
`rustfs-replication`.
- `ReplicationErrorBoundary`: ECStore error/result contracts and
replication-specific error classifiers. `crate::error` imports are
concentrated in
rustfs/src/app/object_usecase.rs
+26 -38
View File
@@ -40,10 +40,10 @@ use super::storage_api::object_usecase::bucket::{
predict_lifecycle_expiration,
quota::QuotaOperation,
replication::{
DeletedObjectReplicationInfo, 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,
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,
},
tagging::decode_tags,
validate_restore_request,
@@ -4852,13 +4852,7 @@ impl DefaultObjectUsecase {
dobj.version_id = Some(Uuid::nil());
}
let deleted_object = DeletedObjectReplicationInfo {
delete_object: dobj,
bucket: bucket.clone(),
event_type: REPLICATE_INCOMING_DELETE.to_string(),
..Default::default()
};
schedule_replication_delete(deleted_object).await;
schedule_replication_delete(dobj, bucket.clone(), REPLICATE_INCOMING_DELETE.to_string()).await;
}
}
@@ -5043,16 +5037,15 @@ impl DefaultObjectUsecase {
if obj_info.name.is_empty() {
if replicate_force_delete {
schedule_replication_delete(DeletedObjectReplicationInfo {
delete_object: StorageDeletedObject {
schedule_replication_delete(
StorageDeletedObject {
object_name: key.clone(),
force_delete: true,
..Default::default()
},
bucket: bucket.clone(),
event_type: REPLICATE_INCOMING_DELETE.to_string(),
..Default::default()
})
bucket.clone(),
REPLICATE_INCOMING_DELETE.to_string(),
)
.await;
}
// Prefix/force-delete returns empty ObjectInfo; still emit bucket notification so webhooks match S3 DELETE.
@@ -5091,30 +5084,25 @@ impl DefaultObjectUsecase {
if schedule_delete_replication {
let _activity_guard = DeleteTailActivityGuard::new(DeleteTailStage::Replication);
let mut deleted_object = DeletedObjectReplicationInfo {
delete_object: StorageDeletedObject {
delete_marker: deleted_object_source.delete_marker && !deleted_delete_marker_version,
delete_marker_version_id: if deleted_object_source.delete_marker {
deleted_object_source.version_id
} else {
None
},
object_name: key.clone(),
version_id: if deleted_object_source.delete_marker {
None
} else {
deleted_object_source.version_id
},
delete_marker_mtime: deleted_object_source.mod_time,
replication_state: Some(replication_state_source.replication_state()),
..Default::default()
let mut deleted_object = StorageDeletedObject {
delete_marker: deleted_object_source.delete_marker && !deleted_delete_marker_version,
delete_marker_version_id: if deleted_object_source.delete_marker {
deleted_object_source.version_id
} else {
None
},
bucket: bucket.clone(),
event_type: REPLICATE_INCOMING_DELETE.to_string(),
object_name: key.clone(),
version_id: if deleted_object_source.delete_marker {
None
} else {
deleted_object_source.version_id
},
delete_marker_mtime: deleted_object_source.mod_time,
replication_state: Some(replication_state_source.replication_state()),
..Default::default()
};
enrich_delete_replication_state_if_needed(&bucket, &mut deleted_object.delete_object, replication_state_source).await;
schedule_replication_delete(deleted_object).await;
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;
}
let delete_marker = obj_info.delete_marker;
rustfs/src/app/storage_api.rs
+6 -4
View File
@@ -611,8 +611,6 @@ pub(crate) mod bucket {
use crate::storage::storage_api::ecstore_bucket::replication as replication_contracts;
pub(crate) type DeletedObjectReplicationInfo =
crate::storage::storage_api::ecstore_bucket::replication::DeletedObjectReplicationInfo;
type ReplicationObjectBridge = crate::storage::storage_api::ecstore_bucket::replication::ReplicationObjectBridge;
pub(crate) type ReplicateDecision = replication_contracts::ReplicateDecision;
#[cfg(test)]
@@ -671,8 +669,12 @@ pub(crate) mod bucket {
ReplicationObjectBridge::schedule_object(oi, store, dsc, replication_contracts::ReplicationType::Object).await;
}
pub(crate) async fn schedule_replication_delete(dv: DeletedObjectReplicationInfo) {
ReplicationObjectBridge::schedule_delete(dv).await;
pub(crate) async fn schedule_replication_delete(
delete_object: crate::storage::storage_api::StorageDeletedObject,
bucket: String,
event_type: String,
) {
ReplicationObjectBridge::schedule_storage_delete(delete_object, bucket, event_type).await;
}
pub(crate) fn delete_replication_state_from_config(
scripts/check_architecture_migration_rules.sh
+5 -5
View File
@@ -87,7 +87,7 @@ require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ecstor
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "EcstoreReplicationBoundaryImports" "ECStore split plan replication boundary imports section"
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "RuntimeReplicationFacadeConsumers" "ECStore split plan runtime replication facade consumer section"
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateFileMetaFacade" "ECStore split plan replication crate filemeta facade section"
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateStorageApiBoundary" "ECStore split plan replication crate storage-api boundary section"
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ReplicationCrateStorageApiIndependence" "ECStore split plan replication crate storage-api independence section"
require_source_contains "docs/architecture/ecstore-module-split-plan.md" "StorageApiReplicationContracts" "ECStore split plan storage-api replication contract section"
require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "## Facade Group Inventory" "ECStore facade inventory group section"
require_source_contains "docs/architecture/ecstore-api-facade-inventory.md" "## External Consumer Boundaries" "ECStore facade inventory consumer boundary section"
@@ -2676,14 +2676,14 @@ fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'rustfs_storage_api::|use\s+rustfs_storage_api\b' \
rg -n --with-filename 'rustfs_storage_api::|use\s+rustfs_storage_api\b|^rustfs-storage-api\b' \
crates/replication/src \
--glob '*.rs' |
rg -v '^crates/replication/src/storage_api\.rs:' || true
--glob '*.rs' \
crates/replication/Cargo.toml || true
) >"$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE"
if [[ -s "$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE" ]]; then
report_failure "replication crate storage-api contracts must stay behind crates/replication/src/storage_api.rs: $(paste -sd '; ' "$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE")"
report_failure "replication crate delete DTO contracts must not import or depend on rustfs-storage-api: $(paste -sd '; ' "$REPLICATION_CRATE_STORAGE_API_BYPASS_HITS_FILE")"
fi
(