diff --git a/Cargo.lock b/Cargo.lock index a92296021..f5cd91a06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9620,7 +9620,6 @@ dependencies = [ "rmp", "rmp-serde", "rustfs-filemeta", - "rustfs-storage-api", "rustfs-utils", "s3s", "serde", diff --git a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs index 73f3d89ef..3908389ba 100644 --- a/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs @@ -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() diff --git a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs index 9ddddd2b5..95ea82d6d 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_bridge.rs @@ -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)] diff --git a/crates/ecstore/src/bucket/replication/replication_object_config.rs b/crates/ecstore/src/bucket/replication/replication_object_config.rs index 803b898fe..2d23466d3 100644 --- a/crates/ecstore/src/bucket/replication/replication_object_config.rs +++ b/crates/ecstore/src/bucket/replication/replication_object_config.rs @@ -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(), diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 62977247d..88c628276 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -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 ReplicationPool { // 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, diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index 98daaf4bb..195c46c7b 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -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() diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs index d1f03548c..2d222841b 100644 --- a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -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; type ListObjectVersionsInfo = StorageListObjectVersionsInfo; @@ -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 ReplicationStorage for T where T: ObjectIO< Error = Error, diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index 5a582c0b7..7aa5b4e62 100644 --- a/crates/replication/Cargo.toml +++ b/crates/replication/Cargo.toml @@ -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 diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 778171c36..33a6849e0 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -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}; diff --git a/crates/replication/src/storage_api.rs b/crates/replication/src/storage_api.rs index 9de76ebe9..2a4e4180a 100644 --- a/crates/replication/src/storage_api.rs +++ b/crates/replication/src/storage_api.rs @@ -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, + pub delete_marker_replication_status: Option, + pub version_purge_status: Option, + pub version_purge_statuses: Option, + pub replicate_decision_str: Option, +} + +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, + pub object_name: String, + pub version_id: Option, + pub delete_marker_mtime: Option, + pub replication_state: Option, + 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) + } +} diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 37c2d33ef..9e806c90f 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -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 diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 436d0d072..2696906b3 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -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; diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index be8f4574e..829d3ba30 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -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( diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 1afa17301..d59df1107 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -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 (