diff --git a/Cargo.lock b/Cargo.lock index f5cd91a06..5ef41185b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9617,9 +9617,10 @@ name = "rustfs-replication" version = "1.0.0-beta.8" dependencies = [ "byteorder", + "bytes", + "regex", "rmp", "rmp-serde", - "rustfs-filemeta", "rustfs-utils", "s3s", "serde", diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 279055b1b..5b1b58130 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -23,8 +23,8 @@ use crate::bucket::lifecycle::lifecycle::{ }; use crate::bucket::lifecycle::replication_sink; use crate::bucket::lifecycle::replication_sink::{ - ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, - version_purge_statuses_map, + ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, + replication_statuses_map, version_purge_statuses_map, }; use crate::bucket::lifecycle::tier_delete_journal::{process_tier_delete_journal_entry, run_tier_delete_journal_recovery_loop}; use crate::bucket::lifecycle::tier_free_version_recovery::{DEFAULT_FREE_VERSION_RECOVERY_LIMIT, recover_tier_free_versions}; @@ -2881,7 +2881,7 @@ async fn schedule_lifecycle_replication_delete_if_needed(oi: &ObjectInfo, dobj: return; } - delete_object.replication_state = replication_state; + delete_object.replication_state = replication_state.as_ref().map(replication_state_to_filemeta); replication_sink::schedule_delete(oi.bucket.clone(), delete_object).await; } diff --git a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs index fab063750..349472743 100644 --- a/crates/ecstore/src/bucket/lifecycle/replication_sink.rs +++ b/crates/ecstore/src/bucket/lifecycle/replication_sink.rs @@ -18,8 +18,8 @@ use crate::bucket::lifecycle::lifecycle::ObjectOpts; #[cfg(test)] pub(crate) use crate::bucket::replication::ReplicateTargetDecision; pub(crate) use crate::bucket::replication::{ - ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, - version_purge_statuses_map, + ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, + replication_statuses_map, version_purge_statuses_map, }; use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; use crate::object_api::{ObjectInfo, ObjectOptions}; diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index 042282292..1d07e99c7 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -37,11 +37,11 @@ paths. | `EcstoreReplicationBoundaryImports` | ECStore-side imports from `rustfs-replication`. | Direct `rustfs-replication` imports under `crates/ecstore/src/bucket/replication` stay in `*_boundary.rs` modules, including config and resync facade re-exports. | | `RuntimeReplicationFacadeConsumers` | Runtime owner consumers of replication DTOs and status types. | Scanner, admin, and storage owner facades import replication DTOs/status types through `rustfs-ecstore`; app storage keeps the remaining direct object/delete helper calls behind its local storage API boundary. | | `ReplicationResyncContracts` | Resync options, target status, bucket status, status classifiers, and persisted resync/MRF status wire format. | Owned by `crates/replication`; ECStore imports them through `replication_resync_boundary.rs`, which maps crate errors to ECStore errors. | -| `ReplicationCrateFileMetaFacade` | Replication facade compatibility symbols that still originate in filemeta wire contracts. | `crates/replication/src/filemeta.rs` is the only direct `rustfs-filemeta` import boundary inside `rustfs-replication`. | +| `ReplicationCrateFileMetaIndependence` | Replication status, decision, MRF, resync, and target-reset wire contracts owned by `rustfs-replication`. | `crates/replication/src/filemeta.rs` owns these contracts; `rustfs-replication` must not import or depend on `rustfs-filemeta`. | | `ReplicationConfigStore` | Replication config persistence and config-derived labels used by target options. | Config read/save helpers and storage class labels are exposed through the contract type in `replication_config_store.rs`. | -| `ReplicationFileMeta` | Replication status, decisions, MRF entries, resync decisions, and target reset helpers. | `rustfs_filemeta` replication contracts are concentrated in `replication_filemeta_boundary.rs`; `FileInfo` remains in the storage boundary for storage trait bindings and walk options. | -| `StorageApiReplicationContracts` | Storage-api delete DTO replication state/status helpers. | The temporary `rustfs-filemeta` dependency is isolated in `crates/storage-api/src/replication.rs` until the wire contracts can move without creating a `rustfs-replication` / `rustfs-storage-api` cycle. | -| `ReplicationCrateStorageApiBoundary` | Storage API delete DTOs consumed by `rustfs-replication`. | `crates/replication/src/storage_api.rs` is the only direct `rustfs-storage-api` import boundary inside `rustfs-replication`. | +| `ReplicationFileMeta` | ECStore compatibility conversions for filemeta replication state/status. | `rustfs_filemeta` to `rustfs_replication` conversions are concentrated in `replication_filemeta_boundary.rs`; `FileInfo` remains in the storage boundary for storage trait bindings and walk options. | +| `StorageApiReplicationContracts` | Storage-api delete DTO replication state/status helpers. | Storage-api owner DTOs keep their local replication boundary; ECStore converts them in `replication_storage_boundary.rs` before queueing replication work. | +| `ReplicationCrateStorageApiIndependence` | Delete work DTOs consumed by `rustfs-replication`. | `crates/replication/src/storage_api.rs` owns these DTOs; `rustfs-replication` must not import or depend on `rustfs-storage-api`. | | `ReplicationObjectDecisionContracts` | Object replication options, delete replication decisions, resync target projection, multipart planning, and delete-marker retry classifiers. | Owned by `crates/replication`; ECStore imports them through `replication_object_decision_boundary.rs`. | | `ReplicationQueueContracts` | Queue admission, heal queue results/actions, worker operations, worker sizing, and backpressure decisions. | Owned by `crates/replication`; ECStore imports them through `replication_queue_boundary.rs`. | | `ReplicationStatsContracts` | Bucket stats, replication target stats, queue/proxy metrics, and worker metric snapshots. | Owned by `crates/replication`; ECStore imports them through `replication_stats_boundary.rs`. | @@ -88,13 +88,12 @@ paths. 10. Keep ECStore owner modules outside `bucket/replication` behind bridge contracts when they need replication codec or config helper behavior. 11. Keep storage-api replication status/state helpers behind - `crates/storage-api/src/replication.rs` until the underlying wire contracts - can move without a `rustfs-replication` / `rustfs-storage-api` dependency - cycle. -12. Keep direct `rustfs-filemeta` imports inside `rustfs-replication` - concentrated in `crates/replication/src/filemeta.rs`. -13. Keep direct `rustfs-storage-api` imports inside `rustfs-replication` - concentrated in `crates/replication/src/storage_api.rs`. + `crates/storage-api/src/replication.rs`; ECStore converts owner DTOs at the + replication storage boundary. +12. Keep `rustfs-replication` independent from `rustfs-filemeta`; ECStore + compatibility conversions live in `replication_filemeta_boundary.rs`. +13. Keep `rustfs-replication` independent from `rustfs-storage-api`; ECStore + compatibility conversions live in `replication_storage_boundary.rs`. 14. Keep direct `rustfs-replication` imports inside ECStore replication concentrated in `*_boundary.rs` modules. 15. Keep scanner, admin, and storage-owner replication status/DTO consumers diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 54b6b06dc..c7723fec1 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -52,7 +52,11 @@ pub(crate) use replication_filemeta_boundary::ReplicateTargetDecision; pub(crate) use replication_filemeta_boundary::version_purge_statuses_map; pub use replication_filemeta_boundary::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType, - VersionPurgeStatusType, replication_statuses_map, + VersionPurgeStatusType, replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, + version_purge_status_to_filemeta, +}; +pub(crate) use replication_filemeta_boundary::{ + replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta, }; pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig}; pub(crate) use replication_migration_bridge::ReplicationMigrationBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs index 625c987595..f394921be 100644 --- a/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs @@ -22,3 +22,65 @@ pub use rustfs_replication::{ REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicationState, ReplicationStatusType, ReplicationType, VersionPurgeStatusType, replication_statuses_map, }; + +pub(crate) fn replication_status_from_filemeta(status: rustfs_filemeta::ReplicationStatusType) -> ReplicationStatusType { + ReplicationStatusType::from(status.as_str()) +} + +pub(crate) fn version_purge_status_from_filemeta(status: rustfs_filemeta::VersionPurgeStatusType) -> VersionPurgeStatusType { + VersionPurgeStatusType::from(status.as_str()) +} + +pub(crate) fn replication_state_from_filemeta(state: &rustfs_filemeta::ReplicationState) -> ReplicationState { + ReplicationState { + replica_timestamp: state.replica_timestamp, + replica_status: replication_status_from_filemeta(state.replica_status.clone()), + delete_marker: state.delete_marker, + replication_timestamp: state.replication_timestamp, + replication_status_internal: state.replication_status_internal.clone(), + version_purge_status_internal: state.version_purge_status_internal.clone(), + replicate_decision_str: state.replicate_decision_str.clone(), + targets: state + .targets + .iter() + .map(|(arn, status)| (arn.clone(), replication_status_from_filemeta(status.clone()))) + .collect(), + purge_targets: state + .purge_targets + .iter() + .map(|(arn, status)| (arn.clone(), version_purge_status_from_filemeta(status.clone()))) + .collect(), + reset_statuses_map: state.reset_statuses_map.clone(), + } +} + +pub fn replication_status_to_filemeta(status: ReplicationStatusType) -> rustfs_filemeta::ReplicationStatusType { + rustfs_filemeta::ReplicationStatusType::from(status.as_str()) +} + +pub fn version_purge_status_to_filemeta(status: VersionPurgeStatusType) -> rustfs_filemeta::VersionPurgeStatusType { + rustfs_filemeta::VersionPurgeStatusType::from(status.as_str()) +} + +pub fn replication_state_to_filemeta(state: &ReplicationState) -> rustfs_filemeta::ReplicationState { + rustfs_filemeta::ReplicationState { + replica_timestamp: state.replica_timestamp, + replica_status: replication_status_to_filemeta(state.replica_status.clone()), + delete_marker: state.delete_marker, + replication_timestamp: state.replication_timestamp, + replication_status_internal: state.replication_status_internal.clone(), + version_purge_status_internal: state.version_purge_status_internal.clone(), + replicate_decision_str: state.replicate_decision_str.clone(), + targets: state + .targets + .iter() + .map(|(arn, status)| (arn.clone(), replication_status_to_filemeta(status.clone()))) + .collect(), + purge_targets: state + .purge_targets + .iter() + .map(|(arn, status)| (arn.clone(), version_purge_status_to_filemeta(status.clone()))) + .collect(), + reset_statuses_map: state.reset_statuses_map.clone(), + } +} diff --git a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs index 2d222841b..e06ab91fc 100644 --- a/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs +++ b/crates/ecstore/src/bucket/replication/replication_storage_boundary.rs @@ -16,6 +16,7 @@ use rustfs_filemeta::FileInfo; use tokio_util::sync::CancellationToken; use super::replication_error_boundary::Error; +use super::replication_filemeta_boundary::{replication_state_from_filemeta, version_purge_status_from_filemeta}; pub(crate) type ReplicationObjectStore = crate::store::ECStore; pub(crate) use crate::client::api_get_options::{AdvancedGetOptions, StatObjectOptions}; pub(crate) use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader}; @@ -101,7 +102,7 @@ pub(crate) fn deleted_object_for_replication(delete_object: DeletedObject) -> Re 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, + replication_state: delete_object.replication_state.as_ref().map(replication_state_from_filemeta), found: delete_object.found, force_delete: delete_object.force_delete, } @@ -112,7 +113,7 @@ pub(crate) fn object_to_delete_for_replication(object: &ObjectToDelete) -> Repli 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_status: object.version_purge_status.clone().map(version_purge_status_from_filemeta), version_purge_statuses: object.version_purge_statuses.clone(), replicate_decision_str: object.replicate_decision_str.clone(), } diff --git a/crates/ecstore/src/client/object_handlers_common.rs b/crates/ecstore/src/client/object_handlers_common.rs index 69a44901d..c51d4f1bf 100644 --- a/crates/ecstore/src/client/object_handlers_common.rs +++ b/crates/ecstore/src/client/object_handlers_common.rs @@ -21,7 +21,7 @@ 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, ReplicationState}; +use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationState, replication_state_to_filemeta}; use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::object_api::ObjectOptions; @@ -106,7 +106,7 @@ pub async fn delete_object_versions(api: &Arc, bucket: &str, to_del: &[ let Some(replication_state) = replication_candidates.get(i).and_then(|c| c.clone()) else { continue; }; - deleted_obj.replication_state = Some(replication_state); + deleted_obj.replication_state = Some(replication_state_to_filemeta(&replication_state)); ReplicationLifecycleBridge::schedule_delete(bucket.to_string(), deleted_obj.clone()).await; } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 79ceb5d00..7e09d5eaa 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -12,6 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. +use crate::bucket::replication::replication_state_from_filemeta; use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ @@ -2132,7 +2133,10 @@ fn decommission_delete_marker_opts( data_movement: true, delete_marker: true, skip_decommissioned: true, - delete_replication: version.replication_state_internal.clone(), + delete_replication: version + .replication_state_internal + .as_ref() + .map(replication_state_from_filemeta), ..Default::default() } } @@ -4236,12 +4240,12 @@ mod tests { let mod_time = OffsetDateTime::now_utc(); let version = rustfs_filemeta::FileInfo { mod_time: Some(mod_time), - replication_state_internal: Some(ReplicationState { + replication_state_internal: Some(crate::bucket::replication::replication_state_to_filemeta(&ReplicationState { replica_status: ReplicationStatusType::Replica, delete_marker: true, replicate_decision_str: "existing".to_string(), ..Default::default() - }), + })), ..Default::default() }; diff --git a/crates/ecstore/src/object_api/mod.rs b/crates/ecstore/src/object_api/mod.rs index 15d06bebf..67fb33229 100644 --- a/crates/ecstore/src/object_api/mod.rs +++ b/crates/ecstore/src/object_api/mod.rs @@ -17,8 +17,8 @@ use crate::bucket::metadata_sys::get_versioning_config; use crate::bucket::replication::{ - ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_statuses_map, - version_purge_statuses_map, + ReplicateDecision, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_status_from_filemeta, + replication_statuses_map, version_purge_status_from_filemeta, version_purge_statuses_map, }; use crate::bucket::versioning::VersioningApi as _; use crate::config::storageclass; diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 44da1e9c3..53715d520 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -445,7 +445,7 @@ impl ObjectInfo { .map(|v| v.replicate_decision_str.clone()) .unwrap_or_default(); - let mut replication_status = fi.replication_status(); + let mut replication_status = replication_status_from_filemeta(fi.replication_status()); if replication_status.is_empty() && let Some(status) = fi.metadata.get(AMZ_BUCKET_REPLICATION_STATUS).cloned() && status == ReplicationStatusType::Replica.as_str() @@ -453,7 +453,7 @@ impl ObjectInfo { replication_status = ReplicationStatusType::Replica; } - let version_purge_status = fi.version_purge_status(); + let version_purge_status = version_purge_status_from_filemeta(fi.version_purge_status()); let transitioned_object = TransitionedObject { name: fi.transitioned_objname.clone(), @@ -1052,10 +1052,10 @@ mod tests { #[test] fn from_file_info_preserves_replication_decision() { let fi = FileInfo { - replication_state_internal: Some(ReplicationState { + replication_state_internal: Some(crate::bucket::replication::replication_state_to_filemeta(&ReplicationState { replicate_decision_str: "arn=true;false;arn:replication::1:dest;rule-id".to_string(), ..Default::default() - }), + })), ..Default::default() }; diff --git a/crates/ecstore/src/services/rebalance/migration.rs b/crates/ecstore/src/services/rebalance/migration.rs index 1c13f46e4..998c84fea 100644 --- a/crates/ecstore/src/services/rebalance/migration.rs +++ b/crates/ecstore/src/services/rebalance/migration.rs @@ -1,4 +1,5 @@ use super::worker::{is_transient_rebalance_error, rebalance_migration_retry_delay, sleep_rebalance_migration_retry}; +use crate::bucket::replication::replication_state_from_filemeta; use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found}; use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions}; @@ -32,7 +33,10 @@ pub(super) fn rebalance_delete_marker_opts(version: &FileInfo, version_id: Optio data_movement: true, delete_marker: true, skip_decommissioned: true, - delete_replication: version.replication_state_internal.clone(), + delete_replication: version + .replication_state_internal + .as_ref() + .map(replication_state_from_filemeta), ..Default::default() } } diff --git a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs index 41e9cefd3..7661e001a 100644 --- a/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs +++ b/crates/ecstore/src/services/rebalance/rebalance_unit_tests.rs @@ -49,7 +49,7 @@ use super::{ DiskStat, GetObjectReader, ObjectInfo, ObjectOptions, RebalSaveOpt, RebalStatus, RebalanceBucketOutcome, RebalanceCleanupWarnings, RebalanceEntryOutcome, RebalanceInfo, RebalanceMeta, RebalanceStats, }; -use crate::bucket::replication::{ReplicationState, ReplicationStatusType}; +use crate::bucket::replication::{ReplicationState, ReplicationStatusType, replication_state_to_filemeta}; use crate::data_movement; use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::disk::RUSTFS_META_BUCKET; @@ -223,12 +223,12 @@ 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(ReplicationState { + replication_state_internal: Some(replication_state_to_filemeta(&ReplicationState { replica_status: ReplicationStatusType::Replica, delete_marker: true, replicate_decision_str: "existing".to_string(), ..Default::default() - }), + })), ..version_deleted() }; diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 033691b57..e002919e4 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -22,6 +22,7 @@ use crate::bucket::metadata_sys; use crate::bucket::object_lock::objectlock_sys::check_retention_for_modification; use crate::bucket::replication::{ ReplicateDecision, ReplicationObjectBridge, ReplicationState, ReplicationStatusType, VersionPurgeStatusType, + replication_state_to_filemeta, }; use crate::bucket::versioning::VersioningApi; use crate::bucket::versioning_sys::BucketVersioningSys; @@ -2944,7 +2945,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks { } record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks); - fi.replication_state_internal = Some(opts.put_replication_state()); + fi.replication_state_internal = Some(replication_state_to_filemeta(&opts.put_replication_state())); fi.is_latest = true; @@ -4152,7 +4153,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { deleted: delete_marker, mark_deleted: mark_delete, mod_time: Some(mod_time), - replication_state_internal: opts.delete_replication.clone(), + replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta), ..Default::default() // TODO: Transition }; @@ -4190,7 +4191,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { mark_deleted: mark_delete, deleted: delete_marker, mod_time: Some(mod_time), - replication_state_internal: opts.delete_replication.clone(), + replication_state_internal: opts.delete_replication.as_ref().map(replication_state_to_filemeta), ..Default::default() }; diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index e9d6636c1..f8b4e59d3 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -1347,7 +1347,8 @@ 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, + ReplicationState, ReplicationStatusType, VersionPurgeStatusType, replication_state_to_filemeta, replication_statuses_map, + version_purge_statuses_map, }; use crate::layout::{ endpoints::{Endpoints, PoolEndpoints}, @@ -1513,13 +1514,13 @@ mod tests { transitioned_objname: "remote/object".to_string(), transition_tier: "WARM".to_string(), transition_version_id: Some(transition_version_id), - replication_state_internal: Some(ReplicationState { + replication_state_internal: Some(replication_state_to_filemeta(&ReplicationState { replication_status_internal: Some("arn:minio:replication:target=COMPLETED;".to_string()), targets: replication_statuses_map("arn:minio:replication:target=COMPLETED;"), version_purge_status_internal: Some("arn:minio:replication:target=PENDING;".to_string()), purge_targets: version_purge_statuses_map("arn:minio:replication:target=PENDING;"), ..Default::default() - }), + })), metadata: HashMap::from([ ("etag".to_string(), "etag-value".to_string()), ("x-amz-meta-key".to_string(), "metadata-value".to_string()), diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml index 7aa5b4e62..40076cdd6 100644 --- a/crates/replication/Cargo.toml +++ b/crates/replication/Cargo.toml @@ -26,10 +26,11 @@ categories = ["web-programming", "development-tools", "filesystem"] documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/" [dependencies] +bytes.workspace = true byteorder.workspace = true +regex.workspace = true rmp.workspace = true rmp-serde.workspace = true -rustfs-filemeta.workspace = true rustfs-utils = { workspace = true, features = ["http", "path", "string"] } s3s.workspace = true serde.workspace = true diff --git a/crates/replication/src/filemeta.rs b/crates/replication/src/filemeta.rs index e02d9cf4c..14121110a 100644 --- a/crates/replication/src/filemeta.rs +++ b/crates/replication/src/filemeta.rs @@ -12,10 +12,1015 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub use rustfs_filemeta::{ - MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, - REPLICATE_INCOMING_DELETE, ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, - ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType, ReplicationType, - ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, - parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, -}; +use bytes::Bytes; +use core::fmt; +use regex::Regex; +use rustfs_utils::http::internal_key_rustfs; +use serde::{Deserialize, Serialize}; +use std::any::Any; +use std::collections::HashMap; +use std::sync::LazyLock; +use std::time::Duration; +use time::OffsetDateTime; +use uuid::Uuid; + +pub const REPLICATION_RESET: &str = "replication-reset"; +pub const REPLICATION_STATUS: &str = "replication-status"; + +// ReplicateQueued - replication being queued trail +pub const REPLICATE_QUEUED: &str = "replicate:queue"; + +// ReplicateExisting - audit trail for existing objects replication +pub const REPLICATE_EXISTING: &str = "replicate:existing"; +// ReplicateExistingDelete - audit trail for delete replication triggered for existing delete markers +pub const REPLICATE_EXISTING_DELETE: &str = "replicate:existing:delete"; + +// ReplicateMRF - audit trail for replication from Most Recent Failures (MRF) queue +pub const REPLICATE_MRF: &str = "replicate:mrf"; +// ReplicateIncoming - audit trail of inline replication +pub const REPLICATE_INCOMING: &str = "replicate:incoming"; +// ReplicateIncomingDelete - audit trail of inline replication of deletes. +pub const REPLICATE_INCOMING_DELETE: &str = "replicate:incoming:delete"; + +// ReplicateHeal - audit trail for healing of failed/pending replications +pub const REPLICATE_HEAL: &str = "replicate:heal"; +// ReplicateHealDelete - audit trail of healing of failed/pending delete replications. +pub const REPLICATE_HEAL_DELETE: &str = "replicate:heal:delete"; + +/// StatusType of Replication for x-amz-replication-status header +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default, Hash)] +pub enum ReplicationStatusType { + /// Pending - replication is pending. + Pending, + /// Completed - replication completed ok. + Completed, + /// CompletedLegacy was called "COMPLETE" incorrectly. + CompletedLegacy, + /// Failed - replication failed. + Failed, + /// Replica - this is a replica. + Replica, + #[default] + Empty, +} + +impl ReplicationStatusType { + /// Returns string representation of status + pub fn as_str(&self) -> &'static str { + match self { + ReplicationStatusType::Pending => "PENDING", + ReplicationStatusType::Completed => "COMPLETED", + ReplicationStatusType::CompletedLegacy => "COMPLETE", + ReplicationStatusType::Failed => "FAILED", + ReplicationStatusType::Replica => "REPLICA", + ReplicationStatusType::Empty => "", + } + } + pub fn is_empty(&self) -> bool { + matches!(self, ReplicationStatusType::Empty) + } +} + +impl fmt::Display for ReplicationStatusType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl From<&str> for ReplicationStatusType { + fn from(s: &str) -> Self { + match s { + "PENDING" => ReplicationStatusType::Pending, + "COMPLETED" => ReplicationStatusType::Completed, + "COMPLETE" => ReplicationStatusType::CompletedLegacy, + "FAILED" => ReplicationStatusType::Failed, + "REPLICA" => ReplicationStatusType::Replica, + _ => ReplicationStatusType::Empty, + } + } +} + +impl From for ReplicationStatusType { + fn from(status: VersionPurgeStatusType) -> Self { + match status { + VersionPurgeStatusType::Pending => ReplicationStatusType::Pending, + VersionPurgeStatusType::Complete => ReplicationStatusType::Completed, + VersionPurgeStatusType::Failed => ReplicationStatusType::Failed, + VersionPurgeStatusType::Empty => ReplicationStatusType::Empty, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, Default)] +pub enum VersionPurgeStatusType { + Pending, + Complete, + Failed, + #[default] + Empty, +} + +impl VersionPurgeStatusType { + /// Returns string representation of version purge status + pub fn as_str(&self) -> &'static str { + match self { + VersionPurgeStatusType::Pending => "PENDING", + VersionPurgeStatusType::Complete => "COMPLETE", + VersionPurgeStatusType::Failed => "FAILED", + VersionPurgeStatusType::Empty => "", + } + } + + /// Returns true if the version is pending purge. + pub fn is_pending(&self) -> bool { + matches!(self, VersionPurgeStatusType::Pending | VersionPurgeStatusType::Failed) + } + + pub fn is_empty(&self) -> bool { + matches!(self, VersionPurgeStatusType::Empty) + } +} + +impl fmt::Display for VersionPurgeStatusType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl From<&str> for VersionPurgeStatusType { + fn from(s: &str) -> Self { + match s { + "PENDING" => VersionPurgeStatusType::Pending, + "COMPLETE" => VersionPurgeStatusType::Complete, + "FAILED" => VersionPurgeStatusType::Failed, + _ => VersionPurgeStatusType::Empty, + } + } +} + +/// Type - replication type enum +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +pub enum ReplicationType { + #[default] + Unset, + Object, + Delete, + Metadata, + Heal, + ExistingObject, + Resync, + All, +} + +impl ReplicationType { + pub fn as_str(&self) -> &'static str { + match self { + ReplicationType::Unset => "", + ReplicationType::Object => "OBJECT", + ReplicationType::Delete => "DELETE", + ReplicationType::Metadata => "METADATA", + ReplicationType::Heal => "HEAL", + ReplicationType::ExistingObject => "EXISTING_OBJECT", + ReplicationType::Resync => "RESYNC", + ReplicationType::All => "ALL", + } + } + + pub fn is_valid(&self) -> bool { + matches!( + self, + ReplicationType::Object + | ReplicationType::Delete + | ReplicationType::Metadata + | ReplicationType::Heal + | ReplicationType::ExistingObject + | ReplicationType::Resync + | ReplicationType::All + ) + } + + pub fn is_data_replication(&self) -> bool { + matches!(self, ReplicationType::Object | ReplicationType::Delete | ReplicationType::Heal) + } +} + +impl fmt::Display for ReplicationType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl From<&str> for ReplicationType { + fn from(s: &str) -> Self { + match s { + "UNSET" => ReplicationType::Unset, + "OBJECT" => ReplicationType::Object, + "DELETE" => ReplicationType::Delete, + "METADATA" => ReplicationType::Metadata, + "HEAL" => ReplicationType::Heal, + "EXISTING_OBJECT" => ReplicationType::ExistingObject, + "RESYNC" => ReplicationType::Resync, + "ALL" => ReplicationType::All, + _ => ReplicationType::Unset, + } + } +} + +/// ReplicationState represents internal replication state +#[derive(Debug, Clone, Serialize, Deserialize, Default, PartialEq, Eq)] +pub struct ReplicationState { + pub replica_timestamp: Option, + pub replica_status: ReplicationStatusType, + pub delete_marker: bool, + pub replication_timestamp: Option, + pub replication_status_internal: Option, + pub version_purge_status_internal: Option, + pub replicate_decision_str: String, + pub targets: HashMap, + pub purge_targets: HashMap, + pub reset_statuses_map: HashMap, +} + +impl ReplicationState { + pub fn new() -> Self { + Self::default() + } + + /// Returns true if replication state is identical for version purge statuses and replication statuses + pub fn equal(&self, other: &ReplicationState) -> bool { + self.replica_status == other.replica_status + && self.replication_status_internal == other.replication_status_internal + && self.version_purge_status_internal == other.version_purge_status_internal + } + + /// Returns overall replication status for the object version being replicated + pub fn composite_replication_status(&self) -> ReplicationStatusType { + if let Some(replication_status_internal) = &self.replication_status_internal { + match ReplicationStatusType::from(replication_status_internal.as_str()) { + ReplicationStatusType::Pending + | ReplicationStatusType::Completed + | ReplicationStatusType::Failed + | ReplicationStatusType::Replica => { + return ReplicationStatusType::from(replication_status_internal.as_str()); + } + _ => { + let repl_status = get_composite_replication_status(&self.targets); + + if self.replica_timestamp.is_none() { + return repl_status; + } + + if repl_status == ReplicationStatusType::Completed + && let (Some(replica_timestamp), Some(replication_timestamp)) = + (self.replica_timestamp, self.replication_timestamp) + && replica_timestamp > replication_timestamp + { + return self.replica_status.clone(); + } + + return repl_status; + } + } + } else if !self.replica_status.is_empty() { + return self.replica_status.clone(); + } + + ReplicationStatusType::default() + } + + /// Returns overall replication purge status for the permanent delete being replicated + pub fn composite_version_purge_status(&self) -> VersionPurgeStatusType { + match VersionPurgeStatusType::from(self.version_purge_status_internal.clone().unwrap_or_default().as_str()) { + VersionPurgeStatusType::Pending | VersionPurgeStatusType::Complete | VersionPurgeStatusType::Failed => { + VersionPurgeStatusType::from(self.version_purge_status_internal.clone().unwrap_or_default().as_str()) + } + _ => get_composite_version_purge_status(&self.purge_targets), + } + } + + /// Returns replicatedInfos struct initialized with the previous state of replication + pub fn target_state(&self, arn: &str) -> ReplicatedTargetInfo { + let resync_timestamp = self + .reset_statuses_map + .get(&target_reset_header(arn)) + .or_else(|| self.reset_statuses_map.get(arn)) + .cloned() + .unwrap_or_default(); + + ReplicatedTargetInfo { + arn: arn.to_string(), + prev_replication_status: self.targets.get(arn).cloned().unwrap_or_default(), + version_purge_status: self.purge_targets.get(arn).cloned().unwrap_or_default(), + resync_timestamp, + ..Default::default() + } + } +} + +pub fn get_composite_replication_status(targets: &HashMap) -> ReplicationStatusType { + if targets.is_empty() { + return ReplicationStatusType::Empty; + } + + let mut completed = 0; + for status in targets.values() { + match status { + ReplicationStatusType::Failed => return ReplicationStatusType::Failed, + ReplicationStatusType::Completed => completed += 1, + _ => {} + } + } + + if completed == targets.len() { + ReplicationStatusType::Completed + } else { + ReplicationStatusType::Pending + } +} + +pub fn get_composite_version_purge_status(targets: &HashMap) -> VersionPurgeStatusType { + if targets.is_empty() { + return VersionPurgeStatusType::default(); + } + + let mut completed = 0; + for status in targets.values() { + match status { + VersionPurgeStatusType::Failed => return VersionPurgeStatusType::Failed, + VersionPurgeStatusType::Complete => completed += 1, + _ => {} + } + } + + if completed == targets.len() { + VersionPurgeStatusType::Complete + } else { + VersionPurgeStatusType::Pending + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +pub enum ReplicationAction { + /// Replicate all data + All, + /// Replicate only metadata + Metadata, + /// Do not replicate + #[default] + None, +} + +impl ReplicationAction { + /// Returns string representation of replication action + pub fn as_str(&self) -> &'static str { + match self { + ReplicationAction::All => "all", + ReplicationAction::Metadata => "metadata", + ReplicationAction::None => "none", + } + } +} + +impl fmt::Display for ReplicationAction { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.as_str()) + } +} + +impl From<&str> for ReplicationAction { + fn from(s: &str) -> Self { + match s { + "all" => ReplicationAction::All, + "metadata" => ReplicationAction::Metadata, + "none" => ReplicationAction::None, + _ => ReplicationAction::None, + } + } +} + +/// ReplicatedTargetInfo struct represents replication info on a target +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ReplicatedTargetInfo { + pub arn: String, + pub size: i64, + pub duration: Duration, + pub replication_action: ReplicationAction, + pub op_type: ReplicationType, + pub replication_status: ReplicationStatusType, + pub prev_replication_status: ReplicationStatusType, + pub version_purge_status: VersionPurgeStatusType, + pub resync_timestamp: String, + pub replication_resynced: bool, + pub endpoint: String, + pub secure: bool, + pub error: Option, +} + +impl ReplicatedTargetInfo { + /// Returns true for a target if arn is empty + pub fn is_empty(&self) -> bool { + self.arn.is_empty() + } +} + +/// ReplicatedInfos struct contains replication information for multiple targets +#[derive(Debug, Clone)] +pub struct ReplicatedInfos { + pub replication_timestamp: Option, + pub targets: Vec, +} + +impl ReplicatedInfos { + /// Returns the total size of completed replications + pub fn completed_size(&self) -> i64 { + let mut sz = 0i64; + for target in &self.targets { + if target.is_empty() { + continue; + } + if target.replication_status == ReplicationStatusType::Completed + && target.prev_replication_status != ReplicationStatusType::Completed + { + sz += target.size; + } + } + sz + } + + /// Returns true if replication was attempted on any of the targets for the object version queued + pub fn replication_resynced(&self) -> bool { + for target in &self.targets { + if target.is_empty() || !target.replication_resynced { + continue; + } + return true; + } + false + } + + /// Returns internal representation of replication status for all targets + pub fn replication_status_internal(&self) -> Option { + let mut result = String::new(); + for target in &self.targets { + if target.is_empty() { + continue; + } + result.push_str(&format!("{}={};", target.arn, target.replication_status)); + } + if result.is_empty() { None } else { Some(result) } + } + + /// Returns overall replication status across all targets + pub fn replication_status(&self) -> ReplicationStatusType { + if self.targets.is_empty() { + return ReplicationStatusType::Empty; + } + + let mut completed = 0; + for target in &self.targets { + match target.replication_status { + ReplicationStatusType::Failed => return ReplicationStatusType::Failed, + ReplicationStatusType::Completed => completed += 1, + _ => {} + } + } + + if completed == self.targets.len() { + ReplicationStatusType::Completed + } else { + ReplicationStatusType::Pending + } + } + + /// Returns overall version purge status across all targets + pub fn version_purge_status(&self) -> VersionPurgeStatusType { + if self.targets.is_empty() { + return VersionPurgeStatusType::Empty; + } + + let mut completed = 0; + for target in &self.targets { + match target.version_purge_status { + VersionPurgeStatusType::Failed => return VersionPurgeStatusType::Failed, + VersionPurgeStatusType::Complete => completed += 1, + _ => {} + } + } + + if completed == self.targets.len() { + VersionPurgeStatusType::Complete + } else { + VersionPurgeStatusType::Pending + } + } + + /// Returns internal representation of version purge status for all targets + pub fn version_purge_status_internal(&self) -> Option { + let mut result = String::new(); + for target in &self.targets { + if target.is_empty() || target.version_purge_status.is_empty() { + continue; + } + result.push_str(&format!("{}={};", target.arn, target.version_purge_status)); + } + if result.is_empty() { None } else { Some(result) } + } + + /// Returns replication action based on target that actually performed replication + pub fn action(&self) -> ReplicationAction { + for target in &self.targets { + if target.is_empty() { + continue; + } + // rely on replication action from target that actually performed replication now. + if target.prev_replication_status != ReplicationStatusType::Completed { + return target.replication_action; + } + } + ReplicationAction::None + } +} + +/// Distinguishes the kind of operation stored in [`MrfReplicateEntry`]. +/// +/// Old serialized files lack the `op` key; `default` maps to `Object`, which preserves +/// the pre-existing replay behaviour for entries written before this field existed. +#[derive(Serialize, Deserialize, Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum MrfOpKind { + #[default] + #[serde(rename = "object")] + Object, + #[serde(rename = "delete")] + Delete, +} + +#[derive(Serialize, Deserialize, Debug, Clone)] +pub struct MrfReplicateEntry { + #[serde(rename = "bucket")] + pub bucket: String, + + #[serde(rename = "object")] + pub object: String, + + // Persisted so recovery after restart can replay the exact version. + // Old serialized files lack this key; `default` fills in None safely. + #[serde(rename = "versionID", skip_serializing_if = "Option::is_none", default)] + pub version_id: Option, + + #[serde(rename = "retryCount")] + pub retry_count: i32, + + #[serde(rename = "size", default)] + pub size: i64, + + // Operation kind. Old files lack this key; default=Object preserves existing behaviour. + #[serde(rename = "op", default)] + pub op: MrfOpKind, + + // For delete entries: the delete-marker version id (distinct from version_id, which is + // the version being purged). Old files lack this; default=None is correct. + #[serde(rename = "deleteMarkerVersionID", skip_serializing_if = "Option::is_none", default)] + pub delete_marker_version_id: Option, + + // For delete entries: whether this is a delete-marker vs a versioned-object delete. + // Old files lack this; default=false is correct. + #[serde(rename = "deleteMarker", default)] + pub delete_marker: bool, +} + +fn retry_count_to_mrf(retry_count: u32) -> i32 { + i32::try_from(retry_count).unwrap_or(i32::MAX) +} + +pub trait ReplicationWorkerOperation: Any + Send + Sync { + fn to_mrf_entry(&self) -> MrfReplicateEntry; + fn as_any(&self) -> &dyn Any; + fn get_bucket(&self) -> &str; + fn get_object(&self) -> &str; + fn get_size(&self) -> i64; + fn is_delete_marker(&self) -> bool; + fn get_op_type(&self) -> ReplicationType; +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ReplicateTargetDecision { + pub replicate: bool, + pub synchronous: bool, + pub arn: String, + pub id: String, +} + +impl ReplicateTargetDecision { + pub fn new(arn: String, replicate: bool, sync: bool) -> Self { + Self { + replicate, + synchronous: sync, + arn, + id: String::new(), + } + } +} + +impl fmt::Display for ReplicateTargetDecision { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{};{};{};{}", self.replicate, self.synchronous, self.arn, self.id) + } +} + +/// ReplicateDecision represents replication decision for each target +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ReplicateDecision { + pub targets_map: HashMap, +} + +impl ReplicateDecision { + pub fn new() -> Self { + Self { + targets_map: HashMap::new(), + } + } + + /// Returns true if at least one target qualifies for replication + pub fn replicate_any(&self) -> bool { + self.targets_map.values().any(|t| t.replicate) + } + + /// Returns true if at least one target qualifies for synchronous replication + pub fn is_synchronous(&self) -> bool { + self.targets_map.values().any(|t| t.synchronous) + } + + /// Updates ReplicateDecision with target's replication decision + pub fn set(&mut self, target: ReplicateTargetDecision) { + self.targets_map.insert(target.arn.clone(), target); + } + + /// Returns a stringified representation of internal replication status with all targets marked as `PENDING` + pub fn pending_status(&self) -> Option { + let mut result = String::new(); + for target in self.targets_map.values() { + if target.replicate { + result.push_str(&format!("{}={};", target.arn, ReplicationStatusType::Pending.as_str())); + } + } + if result.is_empty() { None } else { Some(result) } + } +} + +impl fmt::Display for ReplicateDecision { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let mut result = String::new(); + for (key, value) in &self.targets_map { + result.push_str(&format!("{key}={value},")); + } + write!(f, "{}", result.trim_end_matches(',')) + } +} + +impl Default for ReplicateDecision { + fn default() -> Self { + Self::new() + } +} + +// parse k-v pairs of target ARN to stringified ReplicateTargetDecision delimited by ',' into a +// ReplicateDecision struct +pub fn parse_replicate_decision(_bucket: &str, s: &str) -> std::io::Result { + let mut decision = ReplicateDecision::new(); + + if s.is_empty() { + return Ok(decision); + } + + for p in s.split(',') { + if p.is_empty() { + continue; + } + + let slc = p.split('=').collect::>(); + if slc.len() != 2 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid replicate decision format: {s}"), + )); + } + + let tgt_str = slc[1].trim_matches('"'); + let tgt = tgt_str.split(';').collect::>(); + if tgt.len() != 4 { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidInput, + format!("invalid replicate decision format: {s}"), + )); + } + + let tgt = ReplicateTargetDecision { + replicate: tgt[0] == "true", + synchronous: tgt[1] == "true", + arn: tgt[2].to_string(), + id: tgt[3].to_string(), + }; + decision.targets_map.insert(slc[0].to_string(), tgt); + } + + Ok(decision) + + // r = ReplicateDecision{ + // targetsMap: make(map[string]replicateTargetDecision), + // } + // if len(s) == 0 { + // return + // } + // for _, p := range strings.Split(s, ",") { + // if p == "" { + // continue + // } + // slc := strings.Split(p, "=") + // if len(slc) != 2 { + // return r, errInvalidReplicateDecisionFormat + // } + // tgtStr := strings.TrimSuffix(strings.TrimPrefix(slc[1], `"`), `"`) + // tgt := strings.Split(tgtStr, ";") + // if len(tgt) != 4 { + // return r, errInvalidReplicateDecisionFormat + // } + // r.targetsMap[slc[0]] = replicateTargetDecision{Replicate: tgt[0] == "true", Synchronous: tgt[1] == "true", Arn: tgt[2], ID: tgt[3]} + // } +} + +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ReplicateObjectInfo { + pub name: String, + pub size: i64, + pub actual_size: i64, + pub bucket: String, + pub version_id: Option, + pub etag: Option, + pub mod_time: Option, + pub replication_status: ReplicationStatusType, + pub replication_status_internal: Option, + pub delete_marker: bool, + pub version_purge_status_internal: Option, + pub version_purge_status: VersionPurgeStatusType, + pub replication_state: Option, + pub op_type: ReplicationType, + pub event_type: String, + pub dsc: ReplicateDecision, + pub existing_obj_resync: ResyncDecision, + pub target_statuses: HashMap, + pub target_purge_statuses: HashMap, + pub replication_timestamp: Option, + pub ssec: bool, + pub user_tags: String, + pub checksum: Option, + pub retry_count: u32, +} + +impl ReplicationWorkerOperation for ReplicateObjectInfo { + fn as_any(&self) -> &dyn Any { + self + } + + fn to_mrf_entry(&self) -> MrfReplicateEntry { + MrfReplicateEntry { + bucket: self.bucket.clone(), + object: self.name.clone(), + version_id: self.version_id, + retry_count: retry_count_to_mrf(self.retry_count), + size: self.size, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + } + } + + fn get_bucket(&self) -> &str { + &self.bucket + } + + fn get_object(&self) -> &str { + &self.name + } + + fn get_size(&self) -> i64 { + self.size + } + + fn is_delete_marker(&self) -> bool { + self.delete_marker + } + + fn get_op_type(&self) -> ReplicationType { + self.op_type + } +} + +static REPL_STATUS_REGEX: LazyLock = LazyLock::new(|| match Regex::new(r"([^=].*?)=([^,].*?);") { + Ok(regex) => regex, + Err(err) => panic!("replication status regex must compile: {err}"), +}); + +impl ReplicateObjectInfo { + /// Returns replication status of a target + pub fn target_replication_status(&self, arn: &str) -> ReplicationStatusType { + let binding = self.replication_status_internal.clone().unwrap_or_default(); + let captures = REPL_STATUS_REGEX.captures_iter(&binding); + for cap in captures { + if cap.len() == 3 && &cap[1] == arn { + return ReplicationStatusType::from(&cap[2]); + } + } + ReplicationStatusType::default() + } + + /// Returns the relevant info needed by MRF + pub fn to_mrf_entry(&self) -> MrfReplicateEntry { + MrfReplicateEntry { + bucket: self.bucket.clone(), + object: self.name.clone(), + version_id: self.version_id, + retry_count: retry_count_to_mrf(self.retry_count), + size: self.size, + op: MrfOpKind::Object, + delete_marker_version_id: None, + delete_marker: false, + } + } +} + +// constructs a replication status map from string representation +pub fn replication_statuses_map(s: &str) -> HashMap { + let mut targets = HashMap::new(); + let rep_stat_matches = REPL_STATUS_REGEX.captures_iter(s).map(|c| c.extract()); + for (_, [arn, status]) in rep_stat_matches { + if arn.is_empty() { + continue; + } + let status = ReplicationStatusType::from(status); + targets.insert(arn.to_string(), status); + } + targets +} + +// constructs a version purge status map from string representation +pub fn version_purge_statuses_map(s: &str) -> HashMap { + let mut targets = HashMap::new(); + let purge_status_matches = REPL_STATUS_REGEX.captures_iter(s).map(|c| c.extract()); + for (_, [arn, status]) in purge_status_matches { + if arn.is_empty() { + continue; + } + let status = VersionPurgeStatusType::from(status); + targets.insert(arn.to_string(), status); + } + targets +} + +fn replication_statuses_string(targets: &HashMap) -> Option { + let mut result = String::new(); + for (arn, status) in targets { + if arn.is_empty() || status.is_empty() { + continue; + } + result.push_str(&format!("{arn}={status};")); + } + if result.is_empty() { None } else { Some(result) } +} + +fn version_purge_statuses_string(targets: &HashMap) -> Option { + let mut result = String::new(); + for (arn, status) in targets { + if arn.is_empty() || status.is_empty() { + continue; + } + result.push_str(&format!("{arn}={status};")); + } + if result.is_empty() { None } else { Some(result) } +} + +pub fn get_replication_state(rinfos: &ReplicatedInfos, prev_state: &ReplicationState, _vid: Option) -> ReplicationState { + let reset_status_map: Vec<(String, String)> = rinfos + .targets + .iter() + .filter(|v| !v.resync_timestamp.is_empty()) + .map(|t| (target_reset_header(t.arn.as_str()), t.resync_timestamp.clone())) + .collect(); + + let mut targets = prev_state.targets.clone(); + for (arn, status) in replication_statuses_map(&rinfos.replication_status_internal().unwrap_or_default()) { + targets.insert(arn, status); + } + + let mut purge_targets = prev_state.purge_targets.clone(); + for (arn, status) in version_purge_statuses_map(&rinfos.version_purge_status_internal().unwrap_or_default()) { + purge_targets.insert(arn, status); + } + + let repl_statuses = replication_statuses_string(&targets); + let vpurge_statuses = version_purge_statuses_string(&purge_targets); + + let mut reset_statuses_map = prev_state.reset_statuses_map.clone(); + for (key, value) in reset_status_map { + reset_statuses_map.insert(key, value); + } + + ReplicationState { + replicate_decision_str: prev_state.replicate_decision_str.clone(), + reset_statuses_map, + replica_timestamp: prev_state.replica_timestamp, + replica_status: prev_state.replica_status.clone(), + targets, + replication_status_internal: repl_statuses, + replication_timestamp: rinfos.replication_timestamp, + purge_targets, + version_purge_status_internal: vpurge_statuses, + + ..Default::default() + } +} + +pub fn target_reset_header(arn: &str) -> String { + internal_key_rustfs(&format!("{REPLICATION_RESET}-{arn}")) +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct ResyncTargetDecision { + pub replicate: bool, + pub reset_id: String, + pub reset_before_date: Option, +} + +/// ResyncDecision is a struct representing a map with target's individual resync decisions +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ResyncDecision { + pub targets: HashMap, +} + +impl ResyncDecision { + pub fn new() -> Self { + Self { targets: HashMap::new() } + } + + /// Returns true if no targets with resync decision present + pub fn is_empty(&self) -> bool { + self.targets.is_empty() + } + + pub fn must_resync(&self) -> bool { + self.targets.values().any(|v| v.replicate) + } + + pub fn must_resync_target(&self, tgt_arn: &str) -> bool { + self.targets.get(tgt_arn).map(|v| v.replicate).unwrap_or(false) + } +} + +impl Default for ResyncDecision { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn target_state_reads_resync_timestamp_from_target_reset_header_key() { + let arn = "arn:rustfs:replication:us-east-1:target:bucket"; + let timestamp = "2026-06-30T00:00:00Z;reset-1".to_string(); + let mut state = ReplicationState::default(); + state.reset_statuses_map.insert(target_reset_header(arn), timestamp.clone()); + + let target_state = state.target_state(arn); + + assert_eq!(target_state.resync_timestamp, timestamp); + } + + #[test] + fn get_replication_state_preserves_untouched_target_statuses() { + let target_a = "arn:target:a".to_string(); + let target_b = "arn:target:b".to_string(); + let mut prev_state = ReplicationState::default(); + prev_state.targets.insert(target_a.clone(), ReplicationStatusType::Failed); + prev_state.targets.insert(target_b.clone(), ReplicationStatusType::Completed); + + let rinfos = ReplicatedInfos { + replication_timestamp: None, + targets: vec![ReplicatedTargetInfo { + arn: target_a.clone(), + replication_status: ReplicationStatusType::Completed, + ..Default::default() + }], + }; + + let state = get_replication_state(&rinfos, &prev_state, None); + + assert_eq!(state.targets.get(&target_a), Some(&ReplicationStatusType::Completed)); + assert_eq!(state.targets.get(&target_b), Some(&ReplicationStatusType::Completed)); + assert_eq!( + replication_statuses_map(&state.replication_status_internal.unwrap_or_default()).get(&target_b), + Some(&ReplicationStatusType::Completed) + ); + } +} diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs index 33a6849e0..7cf1d4797 100644 --- a/crates/replication/src/lib.rs +++ b/crates/replication/src/lib.rs @@ -36,9 +36,10 @@ pub use delete::{ should_retry_delete_marker_purge, }; pub use filemeta::{ - REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING_DELETE, - ReplicateDecision, ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, - ReplicationState, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, + REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, REPLICATE_INCOMING, + REPLICATE_INCOMING_DELETE, REPLICATE_MRF, REPLICATE_QUEUED, REPLICATION_RESET, REPLICATION_STATUS, ReplicateDecision, + ReplicateObjectInfo, ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, + ReplicationStatusType, ReplicationType, ReplicationWorkerOperation, ResyncDecision, ResyncTargetDecision, VersionPurgeStatusType, get_replication_state, parse_replicate_decision, replication_statuses_map, target_reset_header, version_purge_statuses_map, }; diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index 9e806c90f..9c893811f 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -124,8 +124,9 @@ Current coupling: - resync options, bucket/target resync status DTOs, status display labels, and the persisted resync status wire format live in `crates/replication`, with 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`; +- `ReplicationCrateFileMetaIndependence`: replication status, decision, MRF, + resync, and target-reset wire contracts are owned inside `rustfs-replication` + instead of importing `rustfs-filemeta`; - `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 @@ -133,8 +134,8 @@ Current coupling: - direct ECStore replication imports from `rustfs-replication` are limited to `*_boundary.rs` modules; - storage-api delete replication status/state helpers use the local - `crates/storage-api/src/replication.rs` contract boundary while the - underlying wire types remain in `rustfs-filemeta`; + `crates/storage-api/src/replication.rs` contract boundary; ECStore converts + those owner DTOs at the replication storage boundary before queueing work; - admin replication extension target filtering and resync request construction stay behind the admin storage boundary instead of exposing replication work DTO construction to handlers; @@ -175,17 +176,18 @@ Required contracts before crate movement: labels are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_config_store.rs`. - `ReplicationFileMeta`: replication status, decisions, MRF entries, resync - decisions, and target reset helpers. `rustfs_filemeta` replication contracts - are concentrated in + decisions, and target reset helpers. ECStore concentrates filemeta-to- + replication compatibility conversions in `crates/ecstore/src/bucket/replication/replication_filemeta_boundary.rs`, while `FileInfo` remains in the storage boundary for storage trait bindings and walk options. -- `ReplicationCrateFileMetaFacade`: replication facade compatibility symbols - that still originate in filemeta wire contracts are concentrated in - `crates/replication/src/filemeta.rs` inside `rustfs-replication`. -- `ReplicationCrateStorageApiBoundary`: storage API delete DTOs consumed by - replication delete/queue/operation helpers are concentrated in - `crates/replication/src/storage_api.rs` inside `rustfs-replication`. +- `ReplicationCrateFileMetaIndependence`: filemeta wire contracts consumed by + replication workers are owned in `crates/replication/src/filemeta.rs`, and + `rustfs-replication` must not import or depend on `rustfs-filemeta`. +- `ReplicationCrateStorageApiIndependence`: delete work DTOs consumed by + replication delete/queue/operation helpers are owned in + `crates/replication/src/storage_api.rs`, and `rustfs-replication` must not + import or depend on `rustfs-storage-api`. - `EcstoreReplicationBoundaryImports`: ECStore-side imports from `rustfs-replication` are concentrated in replication `*_boundary.rs` modules. - `RuntimeReplicationFacadeConsumers`: scanner, admin, storage-owner, and app diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 2696906b3..9cf20dac4 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -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() diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 829d3ba30..b69eddf1f 100644 --- a/rustfs/src/app/storage_api.rs +++ b/rustfs/src/app/storage_api.rs @@ -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, diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index d59df1107..98abe53ae 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -86,7 +86,7 @@ require_source_contains "docs/architecture/overview.md" "ecstore-api-facade-inve require_source_contains "docs/architecture/ecstore-module-split-plan.md" "ecstore-api-facade-inventory.md" "ECStore split plan facade inventory link" 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" "ReplicationCrateFileMetaIndependence" "ECStore split plan replication crate filemeta independence 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" @@ -976,7 +976,8 @@ fi 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 + --glob '*.rs' | + rg -v '^crates/ecstore/src/bucket/replication/' || true } ) >"$ECSTORE_REPLICATION_CONTRACT_BYPASS_HITS_FILE" @@ -2664,14 +2665,14 @@ fi ( cd "$ROOT_DIR" - rg -n --with-filename 'rustfs_filemeta::|use\s+rustfs_filemeta\b' \ - crates/replication/src \ - --glob '*.rs' | - rg -v '^crates/replication/src/filemeta\.rs:' || true + { + rg -n --with-filename 'rustfs_filemeta::|use\s+rustfs_filemeta\b' crates/replication/src --glob '*.rs' + rg -n --with-filename '^rustfs-filemeta\b' crates/replication/Cargo.toml + } || true ) >"$REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE" if [[ -s "$REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE" ]]; then - report_failure "replication crate filemeta contracts must stay behind crates/replication/src/filemeta.rs: $(paste -sd '; ' "$REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE")" + report_failure "rustfs-replication must not import or depend on rustfs-filemeta: $(paste -sd '; ' "$REPLICATION_CRATE_FILEMETA_BYPASS_HITS_FILE")" fi (