gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-31 09:18:28 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ecstore): name replication boundary contracts (#4134)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-01 17:55:12 +08:00
committed by GitHub
parent f5adb5abc1
commit 28e87c1cab
12 changed files with 221 additions and 148 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/replication_pool.rs
+19 -14
View File
@@ -13,14 +13,14 @@
// limitations under the License.
use super::datatypes::ResyncStatusType;
use super::replication_config_store as config_store;
use super::replication_config_store::ReplicationConfigStore;
use super::replication_error_boundary::Error as EcstoreError;
use super::replication_filemeta_boundary::{
MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_HEAL, REPLICATE_HEAL_DELETE, ReplicateDecision,
ReplicateObjectInfo, ReplicatedTargetInfo, ReplicationStatusType, ReplicationType, ReplicationWorkerOperation,
ResyncDecision, VersionPurgeStatusType, replication_statuses_map, version_purge_statuses_map,
};
use super::replication_metadata_boundary as metadata_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,
@@ -28,7 +28,7 @@ use super::replication_resyncer::{
};
use super::replication_state::ReplicationStats;
use super::replication_storage_boundary::{DeletedObject, ObjectInfo, ObjectOptions, ReplicationObjectIO, ReplicationStorage};
use super::replication_target_boundary as target_boundary;
use super::replication_target_boundary::ReplicationTargetStore;
use super::runtime_boundary as runtime_sources;
use lazy_static::lazy_static;
use rustfs_utils::http::{SUFFIX_REPLICATION_TIMESTAMP, get_str};
@@ -788,7 +788,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
let storage = self.storage.clone();
let handle = tokio::spawn(async move {
let data = match config_store::read(storage.clone(), metadata_boundary::MRF_REPLICATION_FILE).await {
let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
Ok(d) => d,
Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start
Err(e) => {
@@ -812,9 +812,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
"Failed to decode MRF recovery file — discarding corrupt data"
);
// Overwrite the corrupt file so we don't fail again on next restart.
let _ = config_store::save(
let _ = ReplicationConfigStore::save(
storage,
metadata_boundary::MRF_REPLICATION_FILE,
ReplicationMetadataStore::MRF_REPLICATION_FILE,
encode_mrf_file(&[]).unwrap_or_default(),
)
.await;
@@ -876,9 +876,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
// Clear AFTER all entries are processed so a crash mid-replay causes at-most-twice
// delivery (idempotent) rather than entry loss.
if let Err(e) =
config_store::save(storage, metadata_boundary::MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default())
.await
if let Err(e) = ReplicationConfigStore::save(
storage,
ReplicationMetadataStore::MRF_REPLICATION_FILE,
encode_mrf_file(&[]).unwrap_or_default(),
)
.await
{
warn!(
component = LOG_COMPONENT_ECSTORE,
@@ -1251,7 +1254,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
async fn flush_mrf_to_disk<S: ReplicationObjectIO>(entries: &[MrfReplicateEntry], storage: &Arc<S>) -> bool {
match encode_mrf_file(entries) {
Ok(data) => {
if let Err(e) = config_store::save(storage.clone(), metadata_boundary::MRF_REPLICATION_FILE, data).await {
if let Err(e) =
ReplicationConfigStore::save(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE, data).await
{
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1283,9 +1288,9 @@ async fn load_bucket_resync_metadata<S: ReplicationObjectIO>(
) -> Result<BucketReplicationResyncStatus, EcstoreError> {
let mut brs = BucketReplicationResyncStatus::new();
let resync_file_path = metadata_boundary::bucket_resync_file_path(bucket);
let resync_file_path = ReplicationMetadataStore::bucket_resync_file_path(bucket);
let data = match config_store::read(obj_api, &resync_file_path).await {
let data = match ReplicationConfigStore::read(obj_api, &resync_file_path).await {
Ok(data) => data,
Err(EcstoreError::ConfigNotFound) => return Ok(brs),
Err(err) => return Err(err),
@@ -1484,7 +1489,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
return;
}
let rcfg = match metadata_boundary::replication_config(bucket).await {
let rcfg = match ReplicationMetadataStore::replication_config(bucket).await {
Ok((config, _)) => config,
Err(err) => {
debug!(
@@ -1501,7 +1506,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
}
};
let tgts = match target_boundary::list_bucket_targets(bucket).await {
let tgts = match ReplicationTargetStore::list_bucket_targets(bucket).await {
Ok(targets) => Some(targets),
Err(err) => {
debug!(