refactor(replication): add metadata boundary (#4070)

This commit is contained in:
Zhengchao An
2026-06-29 23:24:39 +08:00
committed by GitHub
parent f1ec0bd765
commit 04e8c1bb6b
5 changed files with 53 additions and 13 deletions
@@ -15,6 +15,7 @@
mod config;
pub mod datatypes;
mod replication_event_sink;
mod replication_metadata_boundary;
mod replication_pool;
mod replication_resyncer;
mod replication_state;
@@ -0,0 +1,35 @@
// 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 crate::bucket::metadata_sys;
use crate::error::{Error, Result};
use s3s::dto::ReplicationConfiguration;
use time::OffsetDateTime;
pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
metadata_sys::get_replication_config(bucket).await
}
pub(crate) async fn optional_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
let config = match replication_config(bucket).await {
Ok((config, _)) => Some(config),
Err(err) => {
if err != Error::ConfigNotFound {
return Err(err);
}
None
}
};
Ok(config)
}
@@ -12,9 +12,9 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use super::replication_metadata_boundary as metadata_boundary;
use super::replication_target_boundary as target_boundary;
use super::runtime_boundary as runtime_sources;
use crate::bucket::metadata_sys;
use crate::bucket::replication::ResyncOpts;
use crate::bucket::replication::ResyncStatusType;
use crate::bucket::replication::replicate_delete;
@@ -1491,7 +1491,7 @@ pub async fn queue_replication_heal(bucket: &str, oi: ObjectInfo, retry_count: u
return;
}
let rcfg = match metadata_sys::get_replication_config(bucket).await {
let rcfg = match metadata_boundary::replication_config(bucket).await {
Ok((config, _)) => config,
Err(err) => {
debug!(
@@ -13,13 +13,13 @@
// limitations under the License.
use super::replication_event_sink::{EventArgs, send_event};
use super::replication_metadata_boundary as metadata_boundary;
use super::replication_target_boundary as target_boundary;
use super::runtime_boundary as runtime_sources;
use crate::bucket::bandwidth::reader::{BucketOptions, MonitorReaderOptions, MonitoredReader};
use crate::bucket::bucket_target_sys::{
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
};
use crate::bucket::metadata_sys;
use crate::bucket::msgp_decode::{read_msgp_ext8_time, skip_msgp_value, write_msgp_time};
use crate::bucket::replication::ResyncStatusType;
use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _};
@@ -1318,16 +1318,7 @@ pub(crate) async fn save_resync_status<S: EcstoreObjectIO>(
}
async fn get_replication_config(bucket: &str) -> Result<Option<ReplicationConfiguration>> {
let config = match metadata_sys::get_replication_config(bucket).await {
Ok((config, _)) => Some(config),
Err(err) => {
if err != Error::ConfigNotFound {
return Err(err);
}
None
}
};
Ok(config)
metadata_boundary::optional_replication_config(bucket).await
}
#[derive(Debug, Clone, Default)]
@@ -184,6 +184,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_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypass_hits.txt"
REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt"
REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_target_boundary_bypass_hits.txt"
REPLICATION_RUNTIME_SOURCE_BYPASS_HITS_FILE="${TMP_DIR}/replication_runtime_source_bypass_hits.txt"
GLOBAL_REPLICATION_STATE_BYPASS_HITS_FILE="${TMP_DIR}/global_replication_state_bypass_hits.txt"
@@ -2364,6 +2365,18 @@ if [[ -s "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE" ]]; then
report_failure "replication bucket-target access must stay behind replication target boundary: $(paste -sd '; ' "$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'crate::bucket::metadata_sys|metadata_sys::get_replication_config' \
crates/ecstore/src/bucket/replication \
--glob '*.rs' |
rg -v '^crates/ecstore/src/bucket/replication/replication_metadata_boundary\.rs:' || true
) >"$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE"
if [[ -s "$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE" ]]; then
report_failure "replication metadata access must stay behind replication metadata boundary: $(paste -sd '; ' "$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename '\bGLOBAL_REPLICATION_(POOL|STATS)\b' \