diff --git a/Cargo.lock b/Cargo.lock index a65584942..e94d4c69b 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9366,6 +9366,7 @@ dependencies = [ "rustfs-object-capacity", "rustfs-policy", "rustfs-protos", + "rustfs-replication", "rustfs-rio", "rustfs-rio-v2", "rustfs-s3-types", @@ -9887,6 +9888,17 @@ dependencies = [ "tracing", ] +[[package]] +name = "rustfs-replication" +version = "1.0.0-beta.8" +dependencies = [ + "byteorder", + "rmp", + "rmp-serde", + "serde", + "time", +] + [[package]] name = "rustfs-rio" version = "1.0.0-beta.8" diff --git a/Cargo.toml b/Cargo.toml index eb893830c..b4338390c 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -39,6 +39,7 @@ members = [ "crates/protos", # Protocol buffer definitions "crates/rio", # Rust I/O utilities and abstractions "crates/rio-v2", # Next-generation Rust I/O compatibility layer + "crates/replication", # Replication contracts and wire formats "crates/concurrency", # Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling "crates/s3-types", # S3 event type definitions "crates/s3-ops", # S3 operation definitions and mapping @@ -106,6 +107,7 @@ rustfs-obs = { path = "crates/obs", version = "1.0.0-beta.8" } rustfs-policy = { path = "crates/policy", version = "1.0.0-beta.8" } rustfs-protos = { path = "crates/protos", version = "1.0.0-beta.8" } rustfs-protocols = { path = "crates/protocols", version = "1.0.0-beta.8" } +rustfs-replication = { path = "crates/replication", version = "1.0.0-beta.8" } rustfs-rio = { path = "crates/rio", version = "1.0.0-beta.8" } rustfs-rio-v2 = { path = "crates/rio-v2", version = "1.0.0-beta.8" } rustfs-s3-types = { path = "crates/s3-types", version = "1.0.0-beta.8" } diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 926d18946..49930c17e 100644 --- a/crates/ecstore/Cargo.toml +++ b/crates/ecstore/Cargo.toml @@ -49,6 +49,7 @@ rustfs-credentials = { workspace = true } rustfs-common.workspace = true rustfs-policy.workspace = true rustfs-protos.workspace = true +rustfs-replication.workspace = true rustfs-kms.workspace = true rustfs-s3-types = { workspace = true } rustfs-data-usage.workspace = true diff --git a/crates/ecstore/src/bucket/replication/README.md b/crates/ecstore/src/bucket/replication/README.md index b6864e527..c7b6563bb 100644 --- a/crates/ecstore/src/bucket/replication/README.md +++ b/crates/ecstore/src/bucket/replication/README.md @@ -1,18 +1,19 @@ # ECStore Replication Split Inventory -This directory still belongs to ECStore. It is not ready to become a standalone -crate because replication workers currently depend on ECStore object IO, bucket -target clients, bucket metadata systems, runtime state, notification events, -and lifecycle/heal scheduling paths. +This directory still owns the ECStore replication workers. The resync status +contracts and wire format now live in `rustfs-replication`, while worker runtime +code still depends on ECStore object IO, bucket target clients, bucket metadata +systems, runtime state, notification events, and lifecycle/heal scheduling +paths. ## Current Modules | Module | Current role | Split blocker | |---|---|---| | `config.rs` | Replication config helpers, rule matching, and tag filtering. | Uses replication-local filemeta/tagging boundaries and S3 DTOs directly. | -| `datatypes.rs` | Replication status and operation DTOs. | Explicitly re-exported through the ECStore replication facade. | +| `datatypes.rs` | ECStore compatibility re-export for resync status enums. | Re-exports `rustfs-replication` contracts while downstream facade consumers migrate. | | `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, metadata paths, and file metadata replication contracts through local boundaries, config storage, storage contracts through the replication storage boundary, runtime sources, and notification state. | -| `replication_resyncer.rs` | Object replication, delete replication, resync, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. | +| `replication_resyncer.rs` | Object replication, delete replication, resync execution, MRF encode/decode, target calls, and multipart target upload paths. | Depends on target calls and target config types through the replication target boundary, metadata paths and metadata systems through the replication metadata boundary, file metadata replication contracts through the filemeta boundary, error contracts through the error boundary, versioning systems, storage contracts through the replication storage boundary, config-derived storage class labels through the config store, runtime sources, notification events and local event host selection through the event sink, bandwidth reader wrapping, and SetDisks lock timing. | | `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources, file metadata replication contracts, error contracts, and bucket monitor handles through local boundaries, and owns shared replication pool/stat state. | | `replication_lifecycle_bridge.rs` | Lifecycle-originated delete replication admission and version-purge state construction. | Depends on replication config/rule matching, delete-replication decisions, and replication delete scheduling through a local contract type. | | `replication_migration_bridge.rs` | Bucket migration access to persisted replication resync codec helpers. | Keeps migration normalization behind a bridge instead of re-exporting resyncer codec helpers. | @@ -29,6 +30,7 @@ and lifecycle/heal scheduling paths. | `ReplicationObjectIO` | Object read/write primitives used by config, MRF, resync status, and multipart replication paths. | ECStore object API reader/writer types and storage-api object IO contracts are concentrated in `replication_storage_boundary.rs`. | | `ReplicationStorage` | Object read/write/delete, object walk, metadata update, and target object IO. | ECStore object API, storage-api contracts, and read option types are concentrated in `replication_storage_boundary.rs`. | | `ReplicationMetadataStore` | Replication config, MRF/resync state, target reset headers, and status persistence. | Metadata sys access and replication metadata path constants are exposed through the contract type in `replication_metadata_boundary.rs`; versioning sys and config storage imports remain separate contracts. | +| `ReplicationResyncContracts` | Resync options, target status, bucket status, status enum, and persisted resync status wire format. | Owned by `crates/replication`; ECStore maps its error type at the resyncer boundary. | | `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. | | `ReplicationErrorBoundary` | ECStore error/result contracts and replication-specific error classifiers. | `crate::error` imports are concentrated in `replication_error_boundary.rs`. | diff --git a/crates/ecstore/src/bucket/replication/datatypes.rs b/crates/ecstore/src/bucket/replication/datatypes.rs index 79ff131c1..83758ff09 100644 --- a/crates/ecstore/src/bucket/replication/datatypes.rs +++ b/crates/ecstore/src/bucket/replication/datatypes.rs @@ -12,36 +12,4 @@ // See the License for the specific language governing permissions and // limitations under the License. -use serde::{Deserialize, Serialize}; -use std::fmt; - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] -pub enum ResyncStatusType { - #[default] - NoResync, - ResyncPending, - ResyncCanceled, - ResyncStarted, - ResyncCompleted, - ResyncFailed, -} - -impl ResyncStatusType { - pub fn is_valid(&self) -> bool { - *self != ResyncStatusType::NoResync - } -} - -impl fmt::Display for ResyncStatusType { - fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { - let s = match self { - ResyncStatusType::ResyncStarted => "Ongoing", - ResyncStatusType::ResyncCompleted => "Completed", - ResyncStatusType::ResyncFailed => "Failed", - ResyncStatusType::ResyncPending => "Pending", - ResyncStatusType::ResyncCanceled => "Canceled", - ResyncStatusType::NoResync => "", - }; - write!(f, "{s}") - } -} +pub use rustfs_replication::ResyncStatusType; diff --git a/crates/ecstore/src/bucket/replication/mod.rs b/crates/ecstore/src/bucket/replication/mod.rs index 9a6a19597..1cf8f6a6d 100644 --- a/crates/ecstore/src/bucket/replication/mod.rs +++ b/crates/ecstore/src/bucket/replication/mod.rs @@ -46,11 +46,9 @@ pub use replication_pool::{ DynReplicationPool, ReplicationHealQueueResult, ReplicationPoolTrait, ReplicationQueueAdmission, get_global_replication_pool, get_global_replication_stats, init_background_replication, }; -pub use replication_resyncer::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationConfig, ResyncOpts, - TargetReplicationResyncStatus, -}; +pub use replication_resyncer::{DeletedObjectReplicationInfo, MustReplicateOptions, ReplicationConfig}; pub use replication_scanner_bridge::ReplicationScannerBridge; pub use replication_state::{BucketStats, ReplicationStats}; pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage}; pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge; +pub use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; diff --git a/crates/ecstore/src/bucket/replication/replication_migration_bridge.rs b/crates/ecstore/src/bucket/replication/replication_migration_bridge.rs index 316c01373..197185800 100644 --- a/crates/ecstore/src/bucket/replication/replication_migration_bridge.rs +++ b/crates/ecstore/src/bucket/replication/replication_migration_bridge.rs @@ -12,8 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +use super::BucketReplicationResyncStatus; use super::replication_error_boundary::Result; -use super::replication_resyncer::{BucketReplicationResyncStatus, decode_resync_file, encode_resync_file}; +use super::replication_resyncer::{decode_resync_file, encode_resync_file}; pub(crate) struct ReplicationMigrationBridge; diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index f610f78cf..fa56cd2ad 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -22,14 +22,14 @@ use super::replication_filemeta_boundary::{ }; use super::replication_metadata_boundary::ReplicationMetadataStore; use super::replication_resyncer::{ - BucketReplicationResyncStatus, DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer, ResyncOpts, - TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info, - replicate_delete, replicate_object, save_resync_status, + DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer, decode_mrf_file, decode_resync_file, encode_mrf_file, + 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_target_boundary::ReplicationTargetStore; use super::runtime_boundary as runtime_sources; +use super::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use lazy_static::lazy_static; use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str}; use std::any::Any; diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index e130a9605..1e72818fb 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -26,6 +26,7 @@ use super::replication_filemeta_boundary::{ }; use super::replication_lock_boundary::ReplicationLockTiming; use super::replication_metadata_boundary::ReplicationMetadataStore; +#[cfg(test)] use super::replication_msgp_boundary::ReplicationMsgpCodec; use super::replication_storage_boundary::{ AdvancedGetOptions, DeletedObject, EcstoreObjectOperations, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete, @@ -55,6 +56,7 @@ use http_body::Frame; use http_body_util::StreamBody; use regex::Regex; use rmp_serde; +use rustfs_replication::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus}; use rustfs_s3_types::EventName; use rustfs_utils::http::{ AMZ_BUCKET_REPLICATION_STATUS, AMZ_OBJECT_TAGGING, AMZ_TAGGING_DIRECTIVE, CONTENT_ENCODING, HeaderExt as _, @@ -73,8 +75,7 @@ use serde::Deserialize; use serde::Serialize; use std::any::Any; use std::collections::HashMap; -use std::io::{Cursor, Read}; -use std::sync::{Arc, LazyLock}; +use std::sync::Arc; use time::OffsetDateTime; use time::format_description::well_known::Rfc3339; use tokio::io::AsyncRead; @@ -98,29 +99,14 @@ const EVENT_RESYNC_TASK_FAILED: &str = "replication_resync_task_failed"; const EVENT_RESYNC_TARGET_OPERATION_FAILED: &str = "replication_resync_target_operation_failed"; const EVENT_RESYNC_RUNTIME_CHANNEL_FAILED: &str = "replication_resync_runtime_channel_failed"; -pub(crate) const RESYNC_META_FORMAT: u16 = 1; -pub(crate) const RESYNC_META_VERSION: u16 = 1; +pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT; +pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION; const MRF_META_FORMAT: u16 = 1; const MRF_META_VERSION: u16 = 1; const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60); -const WIRE_ZERO_TIME_UNIX: i64 = -62_135_596_800; - -static WIRE_ZERO_TIME: LazyLock = - LazyLock::new(|| OffsetDateTime::from_unix_timestamp(WIRE_ZERO_TIME_UNIX).unwrap_or(OffsetDateTime::UNIX_EPOCH)); static WARNED_MONITOR_UNINIT: std::sync::Once = std::sync::Once::new(); -fn wire_time_or_default(value: Option) -> OffsetDateTime { - value.unwrap_or(*WIRE_ZERO_TIME) -} - -fn normalize_wire_time(value: Option) -> Option { - match value { - Some(v) if v == *WIRE_ZERO_TIME || v == OffsetDateTime::UNIX_EPOCH => None, - other => other, - } -} - fn resync_state_accepts_update(state: &TargetReplicationResyncStatus, opts: &ResyncOpts) -> bool { state.resync_id.is_empty() || opts.resync_id.is_empty() || state.resync_id == opts.resync_id } @@ -203,152 +189,19 @@ fn content_matches(src: &ObjectInfo, tgt: &HeadObjectOutput) -> bool { src_etag.is_some() && src_etag == tgt_etag } -#[derive(Debug, Clone, Default)] -pub struct ResyncOpts { - pub bucket: String, - pub arn: String, - pub resync_id: String, - pub resync_before: Option, -} - -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct TargetReplicationResyncStatus { - pub start_time: Option, - pub last_update: Option, - pub resync_id: String, - pub resync_before_date: Option, - pub resync_status: ResyncStatusType, - pub failed_size: i64, - pub failed_count: i64, - pub replicated_size: i64, - pub replicated_count: i64, - pub bucket: String, - pub object: String, - pub error: Option, -} - -impl TargetReplicationResyncStatus { - pub fn new() -> Self { - Self::default() - } -} - -#[derive(Debug, Clone, Serialize, Deserialize, Default)] -pub struct BucketReplicationResyncStatus { - pub version: u16, - pub targets_map: HashMap, - pub id: i32, - pub last_update: Option, -} - -impl BucketReplicationResyncStatus { - pub fn new() -> Self { - Self { - version: RESYNC_META_VERSION, - ..Default::default() - } - } - - pub fn clone_tgt_stats(&self) -> HashMap { - self.targets_map.clone() - } - - pub fn marshal_msg(&self) -> Result> { - let mut wr = Vec::new(); - rmp::encode::write_map_len(&mut wr, 4)?; - rmp::encode::write_str(&mut wr, "v")?; - rmp::encode::write_i32(&mut wr, i32::from(self.version))?; - rmp::encode::write_str(&mut wr, "brs")?; - rmp::encode::write_map_len(&mut wr, self.targets_map.len() as u32)?; - for (arn, status) in &self.targets_map { - rmp::encode::write_str(&mut wr, arn)?; - status.marshal_wire_msg(&mut wr)?; - } - rmp::encode::write_str(&mut wr, "id")?; - rmp::encode::write_i32(&mut wr, self.id)?; - rmp::encode::write_str(&mut wr, "lu")?; - ReplicationMsgpCodec::write_time(&mut wr, wire_time_or_default(self.last_update))?; - Ok(wr) - } - - pub fn unmarshal_msg(data: &[u8]) -> Result { - let mut rd = Cursor::new(data); - let mut out = Self::new(); - let mut fields = rmp::decode::read_map_len(&mut rd)?; - - while fields > 0 { - fields -= 1; - let key = read_msgp_str(&mut rd)?; - match key.as_str() { - "v" => { - let v: i32 = rmp::decode::read_int(&mut rd)?; - out.version = u16::try_from(v).map_err(|_| Error::other("invalid resync version"))?; - } - "brs" => { - let map_len = rmp::decode::read_map_len(&mut rd)?; - let mut targets = HashMap::with_capacity(map_len as usize); - for _ in 0..map_len { - let arn = read_msgp_str(&mut rd)?; - let status = TargetReplicationResyncStatus::unmarshal_wire_msg(&mut rd)?; - targets.insert(arn, status); - } - out.targets_map = targets; - } - "id" => { - out.id = rmp::decode::read_int::(&mut rd)?; - } - "lu" => { - out.last_update = normalize_wire_time(read_msgp_time_or_nil(&mut rd)?); - } - _ => ReplicationMsgpCodec::skip_value(&mut rd)?, - } - } - Ok(out) - } - - pub fn unmarshal_legacy_msg(data: &[u8]) -> Result { - Ok(rmp_serde::from_slice(data)?) +fn map_resync_error(err: rustfs_replication::resync::Error) -> Error { + match err { + rustfs_replication::resync::Error::CorruptedFormat => Error::CorruptedFormat, + rustfs_replication::resync::Error::Other(err) => Error::other(err), } } pub(crate) fn encode_resync_file(status: &BucketReplicationResyncStatus) -> Result> { - let payload = status.marshal_msg()?; - let mut data = Vec::with_capacity(4 + payload.len()); - let mut major = [0u8; 2]; - byteorder::LittleEndian::write_u16(&mut major, RESYNC_META_FORMAT); - data.extend_from_slice(&major); - let mut minor = [0u8; 2]; - byteorder::LittleEndian::write_u16(&mut minor, RESYNC_META_VERSION); - data.extend_from_slice(&minor); - data.extend_from_slice(&payload); - Ok(data) + rustfs_replication::encode_resync_file(status).map_err(map_resync_error) } pub(crate) fn decode_resync_file(data: &[u8]) -> Result { - if data.len() <= 4 { - return Err(Error::CorruptedFormat); - } - - let mut major = [0u8; 2]; - major.copy_from_slice(&data[0..2]); - if byteorder::LittleEndian::read_u16(&major) != RESYNC_META_FORMAT { - return Err(Error::CorruptedFormat); - } - - let mut minor = [0u8; 2]; - minor.copy_from_slice(&data[2..4]); - if byteorder::LittleEndian::read_u16(&minor) != RESYNC_META_VERSION { - return Err(Error::CorruptedFormat); - } - - let status = match BucketReplicationResyncStatus::unmarshal_msg(&data[4..]) { - Ok(v) => v, - Err(_) => BucketReplicationResyncStatus::unmarshal_legacy_msg(&data[4..])?, - }; - if status.version != RESYNC_META_VERSION { - return Err(Error::CorruptedFormat); - } - Ok(status) + rustfs_replication::decode_resync_file(data).map_err(map_resync_error) } pub(crate) fn encode_mrf_file(entries: &[MrfReplicateEntry]) -> Result> { @@ -381,102 +234,6 @@ pub(crate) fn decode_mrf_file(data: &[u8]) -> Result> { rmp_serde::from_slice(&data[4..]).map_err(|e| Error::other(e.to_string())) } -impl TargetReplicationResyncStatus { - fn marshal_wire_msg(&self, wr: &mut Vec) -> Result<()> { - rmp::encode::write_map_len(wr, 11)?; - rmp::encode::write_str(wr, "st")?; - ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.start_time))?; - rmp::encode::write_str(wr, "lst")?; - ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.last_update))?; - rmp::encode::write_str(wr, "id")?; - rmp::encode::write_str(wr, &self.resync_id)?; - rmp::encode::write_str(wr, "rdt")?; - ReplicationMsgpCodec::write_time(wr, wire_time_or_default(self.resync_before_date))?; - rmp::encode::write_str(wr, "rst")?; - rmp::encode::write_i32(wr, resync_status_to_i32(self.resync_status))?; - rmp::encode::write_str(wr, "fs")?; - rmp::encode::write_i64(wr, self.failed_size)?; - rmp::encode::write_str(wr, "frc")?; - rmp::encode::write_i64(wr, self.failed_count)?; - rmp::encode::write_str(wr, "rs")?; - rmp::encode::write_i64(wr, self.replicated_size)?; - rmp::encode::write_str(wr, "rrc")?; - rmp::encode::write_i64(wr, self.replicated_count)?; - rmp::encode::write_str(wr, "bkt")?; - rmp::encode::write_str(wr, &self.bucket)?; - rmp::encode::write_str(wr, "obj")?; - rmp::encode::write_str(wr, &self.object)?; - Ok(()) - } - - fn unmarshal_wire_msg(rd: &mut R) -> Result { - let mut out = Self::new(); - let mut fields = rmp::decode::read_map_len(rd)?; - - while fields > 0 { - fields -= 1; - let key = read_msgp_str(rd)?; - match key.as_str() { - "st" => out.start_time = normalize_wire_time(read_msgp_time_or_nil(rd)?), - "lst" => out.last_update = normalize_wire_time(read_msgp_time_or_nil(rd)?), - "id" => out.resync_id = read_msgp_str(rd)?, - "rdt" => out.resync_before_date = normalize_wire_time(read_msgp_time_or_nil(rd)?), - "rst" => { - let v: i32 = rmp::decode::read_int(rd)?; - out.resync_status = resync_status_from_i32(v)?; - } - "fs" => out.failed_size = rmp::decode::read_int(rd)?, - "frc" => out.failed_count = rmp::decode::read_int(rd)?, - "rs" => out.replicated_size = rmp::decode::read_int(rd)?, - "rrc" => out.replicated_count = rmp::decode::read_int(rd)?, - "bkt" => out.bucket = read_msgp_str(rd)?, - "obj" => out.object = read_msgp_str(rd)?, - _ => ReplicationMsgpCodec::skip_value(rd)?, - } - } - Ok(out) - } -} - -fn read_msgp_str(rd: &mut R) -> Result { - let len = rmp::decode::read_str_len(rd)? as usize; - let mut buf = vec![0u8; len]; - rd.read_exact(&mut buf)?; - Ok(String::from_utf8(buf)?) -} - -fn read_msgp_time_or_nil(rd: &mut R) -> Result> { - let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; - match marker { - rmp::Marker::Null => Ok(None), - rmp::Marker::Ext8 => Ok(Some(ReplicationMsgpCodec::read_ext8_time(rd)?)), - other => Err(Error::other(format!("expected time ext or nil, got marker: {other:?}"))), - } -} - -fn resync_status_to_i32(status: ResyncStatusType) -> i32 { - match status { - ResyncStatusType::NoResync => 0, - ResyncStatusType::ResyncPending => 1, - ResyncStatusType::ResyncCanceled => 2, - ResyncStatusType::ResyncStarted => 3, - ResyncStatusType::ResyncCompleted => 4, - ResyncStatusType::ResyncFailed => 5, - } -} - -fn resync_status_from_i32(code: i32) -> Result { - match code { - 0 => Ok(ResyncStatusType::NoResync), - 1 => Ok(ResyncStatusType::ResyncPending), - 2 => Ok(ResyncStatusType::ResyncCanceled), - 3 => Ok(ResyncStatusType::ResyncStarted), - 4 => Ok(ResyncStatusType::ResyncCompleted), - 5 => Ok(ResyncStatusType::ResyncFailed), - _ => Err(Error::other(format!("invalid resync status code: {code}"))), - } -} - static RESYNC_WORKER_COUNT: usize = 10; #[derive(Debug)] @@ -4244,7 +4001,8 @@ mod tests { #[test] fn test_resync_none_time_encodes_as_wire_zero_and_decodes_to_none() { - let wire_zero = OffsetDateTime::from_unix_timestamp(WIRE_ZERO_TIME_UNIX).expect("valid wire zero timestamp"); + let wire_zero = OffsetDateTime::from_unix_timestamp(rustfs_replication::resync::WIRE_ZERO_TIME_UNIX) + .expect("valid wire zero timestamp"); let mut with_none = BucketReplicationResyncStatus::new(); with_none.id = 77; diff --git a/crates/replication/Cargo.toml b/crates/replication/Cargo.toml new file mode 100644 index 000000000..dd61297b3 --- /dev/null +++ b/crates/replication/Cargo.toml @@ -0,0 +1,39 @@ +# Copyright 2024 RustFS Team +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +[package] +name = "rustfs-replication" +edition.workspace = true +license.workspace = true +repository.workspace = true +rust-version.workspace = true +version.workspace = true +homepage.workspace = true +description = "Replication contracts and wire formats for RustFS." +keywords = ["replication", "storage", "rustfs", "Minio"] +categories = ["web-programming", "development-tools", "filesystem"] +documentation = "https://docs.rs/rustfs-replication/latest/rustfs_replication/" + +[dependencies] +byteorder.workspace = true +rmp.workspace = true +rmp-serde.workspace = true +serde.workspace = true +time.workspace = true + +[lints] +workspace = true + +[lib] +doctest = false diff --git a/crates/replication/src/lib.rs b/crates/replication/src/lib.rs new file mode 100644 index 000000000..fce854ad6 --- /dev/null +++ b/crates/replication/src/lib.rs @@ -0,0 +1,20 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +pub mod resync; + +pub use resync::{ + BucketReplicationResyncStatus, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, decode_resync_file, + encode_resync_file, +}; diff --git a/crates/replication/src/resync.rs b/crates/replication/src/resync.rs new file mode 100644 index 000000000..a75a79ad8 --- /dev/null +++ b/crates/replication/src/resync.rs @@ -0,0 +1,536 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::collections::HashMap; +use std::fmt; +use std::io::{Cursor, Read, Write}; + +use byteorder::{BigEndian, ByteOrder, LittleEndian}; +use rmp::Marker; +use serde::{Deserialize, Serialize}; +use time::OffsetDateTime; + +pub const RESYNC_META_FORMAT: u16 = 1; +pub const RESYNC_META_VERSION: u16 = 1; + +const MSGP_TIME_EXT_TYPE: i8 = 5; +const MSGP_TIME_LEN: u8 = 12; +pub const WIRE_ZERO_TIME_UNIX: i64 = -62_135_596_800; + +pub type Result = std::result::Result; + +#[derive(Debug)] +pub enum Error { + CorruptedFormat, + Other(String), +} + +impl Error { + fn other(err: impl Into) -> Self { + Self::Other(err.into()) + } +} + +impl fmt::Display for Error { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + match self { + Self::CorruptedFormat => write!(f, "corrupted format"), + Self::Other(err) => write!(f, "{err}"), + } + } +} + +impl std::error::Error for Error {} + +impl From for Error { + fn from(err: std::io::Error) -> Self { + Self::other(err.to_string()) + } +} + +impl From for Error { + fn from(err: std::string::FromUtf8Error) -> Self { + Self::other(err.to_string()) + } +} + +impl From for Error { + fn from(err: rmp::encode::ValueWriteError) -> Self { + Self::other(err.to_string()) + } +} + +impl From for Error { + fn from(err: rmp::decode::ValueReadError) -> Self { + Self::other(err.to_string()) + } +} + +impl From for Error { + fn from(err: rmp::decode::NumValueReadError) -> Self { + Self::other(err.to_string()) + } +} + +impl From for Error { + fn from(err: rmp_serde::decode::Error) -> Self { + Self::other(err.to_string()) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize, Default)] +pub enum ResyncStatusType { + #[default] + NoResync, + ResyncPending, + ResyncCanceled, + ResyncStarted, + ResyncCompleted, + ResyncFailed, +} + +impl ResyncStatusType { + pub fn is_valid(&self) -> bool { + *self != ResyncStatusType::NoResync + } +} + +impl fmt::Display for ResyncStatusType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + let s = match self { + ResyncStatusType::ResyncStarted => "Ongoing", + ResyncStatusType::ResyncCompleted => "Completed", + ResyncStatusType::ResyncFailed => "Failed", + ResyncStatusType::ResyncPending => "Pending", + ResyncStatusType::ResyncCanceled => "Canceled", + ResyncStatusType::NoResync => "", + }; + write!(f, "{s}") + } +} + +#[derive(Debug, Clone, Default)] +pub struct ResyncOpts { + pub bucket: String, + pub arn: String, + pub resync_id: String, + pub resync_before: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct TargetReplicationResyncStatus { + pub start_time: Option, + pub last_update: Option, + pub resync_id: String, + pub resync_before_date: Option, + pub resync_status: ResyncStatusType, + pub failed_size: i64, + pub failed_count: i64, + pub replicated_size: i64, + pub replicated_count: i64, + pub bucket: String, + pub object: String, + pub error: Option, +} + +impl TargetReplicationResyncStatus { + pub fn new() -> Self { + Self::default() + } + + fn marshal_wire_msg(&self, wr: &mut Vec) -> Result<()> { + rmp::encode::write_map_len(wr, 11)?; + rmp::encode::write_str(wr, "st")?; + write_msgp_time(wr, wire_time_or_default(self.start_time))?; + rmp::encode::write_str(wr, "lst")?; + write_msgp_time(wr, wire_time_or_default(self.last_update))?; + rmp::encode::write_str(wr, "id")?; + rmp::encode::write_str(wr, &self.resync_id)?; + rmp::encode::write_str(wr, "rdt")?; + write_msgp_time(wr, wire_time_or_default(self.resync_before_date))?; + rmp::encode::write_str(wr, "rst")?; + rmp::encode::write_i32(wr, resync_status_to_i32(self.resync_status))?; + rmp::encode::write_str(wr, "fs")?; + rmp::encode::write_i64(wr, self.failed_size)?; + rmp::encode::write_str(wr, "frc")?; + rmp::encode::write_i64(wr, self.failed_count)?; + rmp::encode::write_str(wr, "rs")?; + rmp::encode::write_i64(wr, self.replicated_size)?; + rmp::encode::write_str(wr, "rrc")?; + rmp::encode::write_i64(wr, self.replicated_count)?; + rmp::encode::write_str(wr, "bkt")?; + rmp::encode::write_str(wr, &self.bucket)?; + rmp::encode::write_str(wr, "obj")?; + rmp::encode::write_str(wr, &self.object)?; + Ok(()) + } + + fn unmarshal_wire_msg(rd: &mut R) -> Result { + let mut out = Self::new(); + let mut fields = rmp::decode::read_map_len(rd)?; + + while fields > 0 { + fields -= 1; + let key = read_msgp_str(rd)?; + match key.as_str() { + "st" => out.start_time = normalize_wire_time(read_msgp_time_or_nil(rd)?), + "lst" => out.last_update = normalize_wire_time(read_msgp_time_or_nil(rd)?), + "id" => out.resync_id = read_msgp_str(rd)?, + "rdt" => out.resync_before_date = normalize_wire_time(read_msgp_time_or_nil(rd)?), + "rst" => { + let v: i32 = rmp::decode::read_int(rd)?; + out.resync_status = resync_status_from_i32(v)?; + } + "fs" => out.failed_size = rmp::decode::read_int(rd)?, + "frc" => out.failed_count = rmp::decode::read_int(rd)?, + "rs" => out.replicated_size = rmp::decode::read_int(rd)?, + "rrc" => out.replicated_count = rmp::decode::read_int(rd)?, + "bkt" => out.bucket = read_msgp_str(rd)?, + "obj" => out.object = read_msgp_str(rd)?, + _ => skip_msgp_value(rd)?, + } + } + Ok(out) + } +} + +#[derive(Debug, Clone, Serialize, Deserialize, Default)] +pub struct BucketReplicationResyncStatus { + pub version: u16, + pub targets_map: HashMap, + pub id: i32, + pub last_update: Option, +} + +impl BucketReplicationResyncStatus { + pub fn new() -> Self { + Self { + version: RESYNC_META_VERSION, + ..Default::default() + } + } + + pub fn clone_tgt_stats(&self) -> HashMap { + self.targets_map.clone() + } + + pub fn marshal_msg(&self) -> Result> { + let mut wr = Vec::new(); + rmp::encode::write_map_len(&mut wr, 4)?; + rmp::encode::write_str(&mut wr, "v")?; + rmp::encode::write_i32(&mut wr, i32::from(self.version))?; + rmp::encode::write_str(&mut wr, "brs")?; + rmp::encode::write_map_len(&mut wr, self.targets_map.len() as u32)?; + for (arn, status) in &self.targets_map { + rmp::encode::write_str(&mut wr, arn)?; + status.marshal_wire_msg(&mut wr)?; + } + rmp::encode::write_str(&mut wr, "id")?; + rmp::encode::write_i32(&mut wr, self.id)?; + rmp::encode::write_str(&mut wr, "lu")?; + write_msgp_time(&mut wr, wire_time_or_default(self.last_update))?; + Ok(wr) + } + + pub fn unmarshal_msg(data: &[u8]) -> Result { + let mut rd = Cursor::new(data); + let mut out = Self::new(); + let mut fields = rmp::decode::read_map_len(&mut rd)?; + + while fields > 0 { + fields -= 1; + let key = read_msgp_str(&mut rd)?; + match key.as_str() { + "v" => { + let v: i32 = rmp::decode::read_int(&mut rd)?; + out.version = u16::try_from(v).map_err(|_| Error::other("invalid resync version"))?; + } + "brs" => { + let map_len = rmp::decode::read_map_len(&mut rd)?; + let mut targets = HashMap::with_capacity(map_len as usize); + for _ in 0..map_len { + let arn = read_msgp_str(&mut rd)?; + let status = TargetReplicationResyncStatus::unmarshal_wire_msg(&mut rd)?; + targets.insert(arn, status); + } + out.targets_map = targets; + } + "id" => { + out.id = rmp::decode::read_int::(&mut rd)?; + } + "lu" => { + out.last_update = normalize_wire_time(read_msgp_time_or_nil(&mut rd)?); + } + _ => skip_msgp_value(&mut rd)?, + } + } + Ok(out) + } + + pub fn unmarshal_legacy_msg(data: &[u8]) -> Result { + Ok(rmp_serde::from_slice(data)?) + } +} + +pub fn encode_resync_file(status: &BucketReplicationResyncStatus) -> Result> { + let payload = status.marshal_msg()?; + let mut data = Vec::with_capacity(4 + payload.len()); + let mut major = [0u8; 2]; + LittleEndian::write_u16(&mut major, RESYNC_META_FORMAT); + data.extend_from_slice(&major); + let mut minor = [0u8; 2]; + LittleEndian::write_u16(&mut minor, RESYNC_META_VERSION); + data.extend_from_slice(&minor); + data.extend_from_slice(&payload); + Ok(data) +} + +pub fn decode_resync_file(data: &[u8]) -> Result { + if data.len() <= 4 { + return Err(Error::CorruptedFormat); + } + + let mut major = [0u8; 2]; + major.copy_from_slice(&data[0..2]); + if LittleEndian::read_u16(&major) != RESYNC_META_FORMAT { + return Err(Error::CorruptedFormat); + } + + let mut minor = [0u8; 2]; + minor.copy_from_slice(&data[2..4]); + if LittleEndian::read_u16(&minor) != RESYNC_META_VERSION { + return Err(Error::CorruptedFormat); + } + + let status = match BucketReplicationResyncStatus::unmarshal_msg(&data[4..]) { + Ok(v) => v, + Err(_) => BucketReplicationResyncStatus::unmarshal_legacy_msg(&data[4..])?, + }; + if status.version != RESYNC_META_VERSION { + return Err(Error::CorruptedFormat); + } + Ok(status) +} + +fn wire_time_or_default(value: Option) -> OffsetDateTime { + value.unwrap_or_else(wire_zero_time) +} + +fn normalize_wire_time(value: Option) -> Option { + match value { + Some(v) if v == wire_zero_time() || v == OffsetDateTime::UNIX_EPOCH => None, + other => other, + } +} + +fn wire_zero_time() -> OffsetDateTime { + OffsetDateTime::from_unix_timestamp(WIRE_ZERO_TIME_UNIX).unwrap_or(OffsetDateTime::UNIX_EPOCH) +} + +fn read_msgp_str(rd: &mut R) -> Result { + let len = rmp::decode::read_str_len(rd)? as usize; + let mut buf = vec![0u8; len]; + rd.read_exact(&mut buf)?; + Ok(String::from_utf8(buf)?) +} + +fn read_msgp_time_or_nil(rd: &mut R) -> Result> { + let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; + match marker { + Marker::Null => Ok(None), + Marker::Ext8 => Ok(Some(read_msgp_ext8_time(rd)?)), + other => Err(Error::other(format!("expected time ext or nil, got marker: {other:?}"))), + } +} + +fn read_msgp_ext8_time(rd: &mut R) -> Result { + let mut len_buf = [0u8; 1]; + rd.read_exact(&mut len_buf)?; + let len = len_buf[0] as usize; + if len != MSGP_TIME_LEN as usize { + return Err(Error::other(format!("invalid msgp time len: {len}"))); + } + let mut type_buf = [0u8; 1]; + rd.read_exact(&mut type_buf)?; + if type_buf[0] != MSGP_TIME_EXT_TYPE as u8 { + return Err(Error::other(format!("invalid msgp time type: {}", type_buf[0]))); + } + let mut buf = [0u8; 12]; + rd.read_exact(&mut buf)?; + let sec = BigEndian::read_i64(&buf[0..8]); + let nsec = BigEndian::read_u32(&buf[8..12]); + OffsetDateTime::from_unix_timestamp(sec) + .map_err(|_| Error::other("invalid timestamp"))? + .replace_nanosecond(nsec) + .map_err(|_| Error::other("invalid nanosecond")) +} + +fn write_msgp_time(wr: &mut W, time: OffsetDateTime) -> Result<()> { + wr.write_all(&[0xc7, MSGP_TIME_LEN, MSGP_TIME_EXT_TYPE as u8])?; + let mut buf = [0u8; 12]; + BigEndian::write_i64(&mut buf[0..8], time.unix_timestamp()); + BigEndian::write_u32(&mut buf[8..12], time.nanosecond()); + wr.write_all(&buf)?; + Ok(()) +} + +fn skip_msgp_value(rd: &mut R) -> Result<()> { + let marker = rmp::decode::read_marker(rd).map_err(|e| Error::other(format!("{e:?}")))?; + let skip_len: usize = match marker { + Marker::Null | Marker::False | Marker::True => 0, + Marker::FixPos(_) | Marker::FixNeg(_) => 0, + Marker::U8 => 1, + Marker::U16 => 2, + Marker::U32 => 4, + Marker::U64 => 8, + Marker::I8 => 1, + Marker::I16 => 2, + Marker::I32 => 4, + Marker::I64 => 8, + Marker::F32 => 4, + Marker::F64 => 8, + Marker::FixStr(n) => n as usize, + Marker::Str8 | Marker::Bin8 => read_skip_len(rd, 1)?, + Marker::Str16 | Marker::Bin16 => read_skip_len(rd, 2)?, + Marker::Str32 | Marker::Bin32 => read_skip_len(rd, 4)?, + Marker::FixArray(n) => { + for _ in 0..n { + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::Array16 => { + let n = read_skip_len(rd, 2)?; + for _ in 0..n { + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::Array32 => { + let n = read_skip_len(rd, 4)?; + for _ in 0..n { + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::FixMap(n) => { + for _ in 0..n { + skip_msgp_value(rd)?; + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::Map16 => { + let n = read_skip_len(rd, 2)?; + for _ in 0..n { + skip_msgp_value(rd)?; + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::Map32 => { + let n = read_skip_len(rd, 4)?; + for _ in 0..n { + skip_msgp_value(rd)?; + skip_msgp_value(rd)?; + } + return Ok(()); + } + Marker::FixExt1 => 1, + Marker::FixExt2 => 2, + Marker::FixExt4 => 4, + Marker::FixExt8 => 8, + Marker::FixExt16 => 16, + Marker::Ext8 => 1 + read_skip_len(rd, 1)?, + Marker::Ext16 => 2 + read_skip_len(rd, 2)?, + Marker::Ext32 => 4 + read_skip_len(rd, 4)?, + Marker::Reserved => 0, + }; + if skip_len > 0 { + let mut buf = vec![0u8; skip_len]; + rd.read_exact(&mut buf)?; + } + Ok(()) +} + +fn read_skip_len(rd: &mut R, bytes: usize) -> Result { + let mut buf = vec![0u8; bytes]; + rd.read_exact(&mut buf)?; + let len = match bytes { + 1 => buf[0] as usize, + 2 => u16::from_be_bytes([buf[0], buf[1]]) as usize, + 4 => u32::from_be_bytes([buf[0], buf[1], buf[2], buf[3]]) as usize, + _ => return Err(Error::other("invalid MessagePack length width")), + }; + Ok(len) +} + +fn resync_status_to_i32(status: ResyncStatusType) -> i32 { + match status { + ResyncStatusType::NoResync => 0, + ResyncStatusType::ResyncPending => 1, + ResyncStatusType::ResyncCanceled => 2, + ResyncStatusType::ResyncStarted => 3, + ResyncStatusType::ResyncCompleted => 4, + ResyncStatusType::ResyncFailed => 5, + } +} + +fn resync_status_from_i32(code: i32) -> Result { + match code { + 0 => Ok(ResyncStatusType::NoResync), + 1 => Ok(ResyncStatusType::ResyncPending), + 2 => Ok(ResyncStatusType::ResyncCanceled), + 3 => Ok(ResyncStatusType::ResyncStarted), + 4 => Ok(ResyncStatusType::ResyncCompleted), + 5 => Ok(ResyncStatusType::ResyncFailed), + _ => Err(Error::other(format!("invalid resync status code: {code}"))), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resync_status_display_matches_admin_contract() { + assert_eq!(ResyncStatusType::ResyncStarted.to_string(), "Ongoing"); + assert_eq!(ResyncStatusType::ResyncCompleted.to_string(), "Completed"); + assert_eq!(ResyncStatusType::NoResync.to_string(), ""); + } + + #[test] + fn resync_file_round_trips_status() { + let mut status = BucketReplicationResyncStatus::new(); + status.targets_map.insert( + "arn:replication:a".to_string(), + TargetReplicationResyncStatus { + resync_id: "rid-1".to_string(), + resync_status: ResyncStatusType::ResyncStarted, + bucket: "bucket-a".to_string(), + object: "object-a".to_string(), + replicated_count: 7, + ..Default::default() + }, + ); + + let data = encode_resync_file(&status).expect("resync status should encode"); + let got = decode_resync_file(&data).expect("resync status should decode"); + + assert_eq!(got.version, RESYNC_META_VERSION); + assert_eq!(got.targets_map["arn:replication:a"].resync_id, "rid-1"); + assert_eq!(got.targets_map["arn:replication:a"].resync_status, ResyncStatusType::ResyncStarted); + assert_eq!(got.targets_map["arn:replication:a"].replicated_count, 7); + } +} diff --git a/docs/architecture/ecstore-module-split-plan.md b/docs/architecture/ecstore-module-split-plan.md index cd21c1c5a..70d94b3a9 100644 --- a/docs/architecture/ecstore-module-split-plan.md +++ b/docs/architecture/ecstore-module-split-plan.md @@ -98,7 +98,9 @@ Focused verification for the first code-bearing lifecycle PR: ## Replication Candidate -`bucket/replication` is not ready for a standalone crate yet. +`rustfs-replication` now owns the resync status contracts and persisted resync +status wire format. The remaining `bucket/replication` worker runtime is not +ready for a full standalone crate yet. Current coupling: @@ -120,6 +122,9 @@ Current coupling: - bucket metadata migration and bucket target removal checks use local replication bridges instead of importing resyncer codec or config helper internals; +- 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; - admin replication extension target filtering and resync request construction stay behind the admin storage boundary instead of exposing replication work DTO construction to handlers; @@ -189,6 +194,8 @@ Required contracts before crate movement: - `ReplicationMigrationBridge`: persisted resync status decode/encode access for bucket metadata migration is exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_migration_bridge.rs`. +- `ReplicationResyncContracts`: resync options, target/bucket resync status, + status labels, and persisted status encoding live in `crates/replication`. - `ReplicationObjectBridge`: app and SetDisks object write/delete replication decisions and scheduling are exposed through the contract type in `crates/ecstore/src/bucket/replication/replication_object_bridge.rs`. diff --git a/scripts/check_architecture_migration_rules.sh b/scripts/check_architecture_migration_rules.sh index 0f1abb04e..850a968fd 100755 --- a/scripts/check_architecture_migration_rules.sh +++ b/scripts/check_architecture_migration_rules.sh @@ -199,6 +199,7 @@ FUZZ_ECSTORE_COMPAT_BYPASS_HITS_FILE="${TMP_DIR}/fuzz_ecstore_compat_bypass_hits EXTERNAL_ECSTORE_API_BOUNDARY_HITS_FILE="${TMP_DIR}/external_ecstore_api_boundary_hits.txt" REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.txt" REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE="${TMP_DIR}/replication_facade_wildcard_export_hits.txt" +REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE="${TMP_DIR}/replication_resync_contract_backslide_hits.txt" STORAGE_REPLICATION_HANDLE_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/storage_replication_handle_boundary_bypass_hits.txt" ADMIN_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/admin_replication_dto_boundary_bypass_hits.txt" APP_REPLICATION_DTO_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/app_replication_dto_boundary_bypass_hits.txt" @@ -2540,6 +2541,17 @@ if [[ -s "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE" ]]; then report_failure "replication facade must use explicit compatibility exports instead of wildcard re-exports: $(paste -sd '; ' "$REPLICATION_FACADE_WILDCARD_EXPORT_HITS_FILE")" fi +( + cd "$ROOT_DIR" + rg -n --with-filename 'pub\s+(?:struct|enum)\s+(ResyncOpts|TargetReplicationResyncStatus|BucketReplicationResyncStatus|ResyncStatusType)\b' \ + crates/ecstore/src/bucket/replication \ + --glob '*.rs' || true +) >"$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE" + +if [[ -s "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE" ]]; then + report_failure "resync DTO contracts must stay in crates/replication: $(paste -sd '; ' "$REPLICATION_RESYNC_CONTRACT_BACKSLIDE_HITS_FILE")" +fi + ( cd "$ROOT_DIR" {