gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 13:27:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ecstore): route replication metadata paths (#4121)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-30 22:48:34 +08:00
committed by GitHub
parent 4efefd67b5
commit 942a8b505d
7 changed files with 309 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/README.md
+4 -4
View File
@@ -11,8 +11,8 @@ and lifecycle/heal scheduling paths.
|---|---|---|
| `config.rs` | Replication config helpers, rule matching, and tag filtering. | Uses replication-local tagging boundary and S3 DTOs directly. |
| `datatypes.rs` | Replication status and operation DTOs. | Publicly re-exported through the ECStore replication facade. |
| `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys, 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 through the replication target boundary, metadata/versioning systems, storage contracts through the replication storage boundary, disk paths, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
| `replication_pool.rs` | Replication queue, worker pool, MRF persistence, bucket stats, and delete/object scheduling. | Depends on bucket target sys, bucket metadata sys and metadata paths 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, versioning systems, storage contracts through the replication storage boundary, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
| `replication_state.rs` | Replication queue/stat state and worker accounting. | Reads runtime sources and owns shared replication pool/stat state. |
| `rule.rs` | Rule evaluation helpers for object replication options. | Depends on ECStore replication object option types. |
| `mod.rs` | Compatibility re-export facade for the current ECStore owner. | Must stay stable until downstream scanner, lifecycle, heal, and metrics paths compile through replacement contracts. |
@@ -23,8 +23,8 @@ 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. | Direct metadata sys, versioning sys, config storage, and file metadata imports. |
| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, and target operation option types. | Bucket target sys access and target operation types are concentrated in `replication_target_boundary.rs`. |
| `ReplicationMetadataStore` | Replication config, MRF/resync state, target reset headers, and status persistence. | Metadata sys access and replication metadata path constants are concentrated in `replication_metadata_boundary.rs`; versioning sys, config storage, and file metadata imports remain. |
| `ReplicationTargetStore` | Bucket target listing, target client lookup, target offline checks, target config types, and target operation option types. | Bucket target sys access, `BucketTargets`, and target operation types are concentrated in `replication_target_boundary.rs`. |
| `ReplicationRuntime` | Worker pool, queue sizing, stats, bucket monitor, local node identity, cancellation, and admission state. | Direct runtime source/global access and shared replication pool/stat state. |
| `ReplicationBandwidthLimiter` | Target reader wrapping for replication bandwidth accounting and throttling. | Direct bucket bandwidth reader imports from resyncer paths. |
| `ReplicationEventSink` | Notification and audit events for skipped, failed, pending, and completed replication operations. | Direct event notification service calls from worker code. |
crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs
+36
View File
@@ -13,10 +13,16 @@
// limitations under the License.
use crate::bucket::metadata_sys;
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
use rustfs_utils::path::path_join_buf;
use s3s::dto::ReplicationConfiguration;
use time::OffsetDateTime;
pub(crate) const REPLICATION_DIR: &str = ".replication";
pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin";
pub(crate) const MRF_REPLICATION_FILE: &str = "config/replication/mrf.bin";
pub(crate) async fn replication_config(bucket: &str) -> Result<(ReplicationConfiguration, OffsetDateTime)> {
metadata_sys::get_replication_config(bucket).await
}
@@ -33,3 +39,33 @@ pub(crate) async fn optional_replication_config(bucket: &str) -> Result<Option<R
};
Ok(config)
}
pub(crate) fn rustfs_meta_bucket() -> &'static str {
RUSTFS_META_BUCKET
}
pub(crate) fn resync_lock_key(bucket: &str, arn: &str) -> String {
format!("{REPLICATION_DIR}/{bucket}/{arn}")
}
pub(crate) fn bucket_resync_dir_path(bucket: &str) -> String {
path_join_buf(&[BUCKET_META_PREFIX, bucket, REPLICATION_DIR])
}
pub(crate) fn bucket_resync_file_path(bucket: &str) -> String {
let resync_dir_path = bucket_resync_dir_path(bucket);
path_join_buf(&[&resync_dir_path, RESYNC_FILE_NAME])
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn replication_metadata_paths_match_existing_layout() {
assert_eq!(resync_lock_key("bucket-a", "arn-a"), ".replication/bucket-a/arn-a");
assert_eq!(bucket_resync_dir_path("bucket-a"), "buckets/bucket-a/.replication");
assert_eq!(bucket_resync_file_path("bucket-a"), "buckets/bucket-a/.replication/resync.bin");
assert_eq!(MRF_REPLICATION_FILE, "config/replication/mrf.bin");
}
}
crates/ecstore/src/bucket/replication/replication_pool.rs
+16 -10
View File
@@ -22,12 +22,11 @@ use crate::bucket::replication::ResyncStatusType;
use crate::bucket::replication::replicate_delete;
use crate::bucket::replication::replicate_object;
use crate::bucket::replication::replication_resyncer::{
BucketReplicationResyncStatus, DeletedObjectReplicationInfo, MRF_REPLICATION_FILE, REPLICATION_DIR, RESYNC_FILE_NAME,
ReplicationConfig, ReplicationResyncer, TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file,
get_heal_replicate_object_info, save_resync_status,
BucketReplicationResyncStatus, DeletedObjectReplicationInfo, ReplicationConfig, ReplicationResyncer,
TargetReplicationResyncStatus, decode_mrf_file, decode_resync_file, encode_mrf_file, get_heal_replicate_object_info,
save_resync_status,
};
use crate::bucket::replication::replication_state::ReplicationStats;
use crate::disk::BUCKET_META_PREFIX;
use crate::error::Error as EcstoreError;
use lazy_static::lazy_static;
use rustfs_filemeta::MrfOpKind;
@@ -800,7 +799,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(), MRF_REPLICATION_FILE).await {
let data = match config_store::read(storage.clone(), metadata_boundary::MRF_REPLICATION_FILE).await {
Ok(d) => d,
Err(EcstoreError::ConfigNotFound) => return, // no file yet — normal on first start
Err(e) => {
@@ -824,7 +823,12 @@ 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(storage, MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default()).await;
let _ = config_store::save(
storage,
metadata_boundary::MRF_REPLICATION_FILE,
encode_mrf_file(&[]).unwrap_or_default(),
)
.await;
return;
}
};
@@ -883,7 +887,10 @@ 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, MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default()).await {
if let Err(e) =
config_store::save(storage, metadata_boundary::MRF_REPLICATION_FILE, encode_mrf_file(&[]).unwrap_or_default())
.await
{
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1255,7 +1262,7 @@ 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(), MRF_REPLICATION_FILE, data).await {
if let Err(e) = config_store::save(storage.clone(), metadata_boundary::MRF_REPLICATION_FILE, data).await {
warn!(
component = LOG_COMPONENT_ECSTORE,
subsystem = LOG_SUBSYSTEM_REPLICATION,
@@ -1287,8 +1294,7 @@ async fn load_bucket_resync_metadata<S: ReplicationObjectIO>(
) -> Result<BucketReplicationResyncStatus, EcstoreError> {
let mut brs = BucketReplicationResyncStatus::new();
let resync_dir_path = format!("{BUCKET_META_PREFIX}/{bucket}/{REPLICATION_DIR}");
let resync_file_path = format!("{resync_dir_path}/{RESYNC_FILE_NAME}");
let resync_file_path = metadata_boundary::bucket_resync_file_path(bucket);
let data = match config_store::read(obj_api, &resync_file_path).await {
Ok(data) => data,
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+7 -13
View File
@@ -25,14 +25,12 @@ use super::replication_storage_boundary::{
use super::replication_tagging_boundary as tagging_boundary;
use super::replication_target_boundary;
use super::replication_target_boundary::{
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
AdvancedPutOptions, BucketTargets, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
};
use super::replication_versioning_boundary as versioning_boundary;
use super::runtime_boundary as runtime_sources;
use crate::bucket::replication::ResyncStatusType;
use crate::bucket::replication::{ObjectOpts, ReplicationConfigurationExt as _};
use crate::bucket::target::BucketTargets;
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
use crate::error::{Error, Result, is_err_object_not_found, is_err_version_not_found};
use aws_sdk_s3::error::{ProvideErrorMetadata, SdkError};
use aws_sdk_s3::operation::head_object::{HeadObjectError, HeadObjectOutput};
@@ -68,7 +66,6 @@ use rustfs_utils::http::{
SUFFIX_REPLICATION_ACTUAL_OBJECT_SIZE, SUFFIX_REPLICATION_RESET_STATUS, SUFFIX_REPLICATION_SSEC_CRC, get_header_map, get_str,
has_internal_suffix, insert_header_map, insert_str, internal_key_strip_suffix_prefix, is_internal_key,
};
use rustfs_utils::path::path_join_buf;
use rustfs_utils::string::strings_has_prefix_fold;
use rustfs_utils::{DEFAULT_SIP_HASH_KEY, sip_hash};
use s3s::dto::ReplicationConfiguration;
@@ -101,14 +98,8 @@ 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 REPLICATION_DIR: &str = ".replication";
pub(crate) const RESYNC_FILE_NAME: &str = "resync.bin";
pub(crate) const RESYNC_META_FORMAT: u16 = 1;
pub(crate) const RESYNC_META_VERSION: u16 = 1;
// MRF (Most Recent Failures) persistence file — stored at
// `{RUSTFS_META_BUCKET}/config/replication/mrf.bin`, cross-bucket.
pub(crate) const MRF_REPLICATION_FILE: &str = "config/replication/mrf.bin";
const MRF_META_FORMAT: u16 = 1;
const MRF_META_VERSION: u16 = 1;
const RESYNC_TIME_INTERVAL: TokioDuration = TokioDuration::from_secs(60);
@@ -746,8 +737,11 @@ impl ReplicationResyncer {
// Acquire a cluster-wide leader lock for this (bucket, ARN) pair so that only
// one node runs the resync scan at a time. Without this, every cluster node would
// scan and replicate every object independently, causing N-fold duplicate traffic.
let resync_lock_key = format!("{}/{}/{}", REPLICATION_DIR, opts.bucket, opts.arn);
let resync_ns_lock = match storage.new_ns_lock(RUSTFS_META_BUCKET, &resync_lock_key).await {
let resync_lock_key = metadata_boundary::resync_lock_key(&opts.bucket, &opts.arn);
let resync_ns_lock = match storage
.new_ns_lock(metadata_boundary::rustfs_meta_bucket(), &resync_lock_key)
.await
{
Ok(l) => l,
Err(e) => {
warn!(
@@ -1234,7 +1228,7 @@ pub(crate) async fn save_resync_status<S: ReplicationObjectIO>(
) -> Result<()> {
let data = encode_resync_file(status)?;
let config_file = path_join_buf(&[BUCKET_META_PREFIX, bucket, REPLICATION_DIR, RESYNC_FILE_NAME]);
let config_file = metadata_boundary::bucket_resync_file_path(bucket);
config_store::save(api, &config_file, data).await?;
Ok(())
crates/ecstore/src/bucket/replication/replication_target_boundary.rs
+1 -1
View File
@@ -15,11 +15,11 @@
use std::sync::Arc;
use crate::bucket::bucket_target_sys::{BucketTargetError, BucketTargetSys};
use crate::bucket::target::BucketTargets;
pub(crate) use crate::bucket::bucket_target_sys::{
AdvancedPutOptions, PutObjectOptions, PutObjectPartOptions, RemoveObjectOptions, TargetClient,
};
pub(crate) use crate::bucket::target::BucketTargets;
pub(crate) async fn list_bucket_targets(bucket: &str) -> Result<BucketTargets, BucketTargetError> {
BucketTargetSys::get().list_bucket_targets(bucket).await
docs/architecture/ecstore-module-split-plan.md
+10 -5
View File
@@ -107,9 +107,9 @@ Current coupling:
classification, runtime replication pool/stat handles, bucket monitor and
bandwidth reader access through local boundaries, local node names, and
notification events;
- resync and delete replication paths call metadata directly, while bucket
target system access and target operation types are concentrated behind the
replication target boundary;
- resync and delete replication paths call metadata paths through the metadata
boundary, while bucket target system access, target config types, and target
operation types are concentrated behind the replication target boundary;
- lifecycle and heal paths schedule replication work through the current ECStore
module;
- global replication pool/stat initialization still lives with ECStore runtime
@@ -127,9 +127,14 @@ Required contracts before crate movement:
storage-api contracts, and read option types are concentrated in
`crates/ecstore/src/bucket/replication/replication_storage_boundary.rs`.
- `ReplicationMetadataStore`: replication config, target reset headers,
MRF/resync state, and status persistence.
MRF/resync state, and status persistence. Metadata sys access and replication
metadata path constants are concentrated in
`crates/ecstore/src/bucket/replication/replication_metadata_boundary.rs`.
- `ReplicationTargetStore`: bucket target listing, target client lookup,
target offline checks, and target operation option types.
target offline checks, target config types, and target operation option
types. Bucket target sys access, `BucketTargets`, and target operation types
are concentrated in
`crates/ecstore/src/bucket/replication/replication_target_boundary.rs`.
- `ReplicationRuntime`: pool, stats, worker admission, bucket monitor, local
node identity, cancellation, and queue sizing.
- `ReplicationBandwidthLimiter`: target reader wrapping for replication
scripts/check_architecture_migration_rules.sh
+235 -4
View File
@@ -2612,13 +2612,125 @@ fi
cd "$ROOT_DIR"
find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |
xargs -0 perl -0ne '
while (/(?:(?:crate::bucket|super(?:::super)+)(?:::\{[^;]*\bbucket_target_sys\b|::bucket_target_sys\b)|BucketTargetSys::get\(\))/sg) {
while (/(?:crate::bucket::(?:bucket_target_sys|target)\b|super(?:::super)+::(?:bucket_target_sys|target)\b|BucketTargetSys::get\(\))/sg) {
my $prefix = substr($_, 0, $-[0]);
my $line = ($prefix =~ tr/\n//) + 1;
my $match = $&;
$match =~ s/\s+/ /g;
print "$ARGV:$line:$match\n";
}
while (/crate::bucket::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
if ($part =~ /^\s*(?:bucket_target_sys|target)\b/s) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:crate::bucket::{$part}\n";
}
$part_start = $j + 1;
}
}
}
while (/super(?:::super)+::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
if ($part =~ /^\s*(?:bucket_target_sys|target)\b/s) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:super::super::{$part}\n";
}
$part_start = $j + 1;
}
}
}
while (/crate::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
my $hit = $part =~ /^\s*bucket\s*::\s*(?:bucket_target_sys|target)\b/s;
if (!$hit && $part =~ /^\s*bucket\s*::\s*\{(.*)\}\s*$/s) {
my $bucket_body = $1;
my $bucket_part_start = 0;
my $bucket_depth = 0;
for (my $k = 0; $k <= length($bucket_body); $k++) {
my $bucket_ch = $k < length($bucket_body) ? substr($bucket_body, $k, 1) : ",";
$bucket_depth++ if $bucket_ch eq "{";
$bucket_depth-- if $bucket_ch eq "}";
if ($bucket_ch eq "," && $bucket_depth == 0) {
my $bucket_part = substr($bucket_body, $bucket_part_start, $k - $bucket_part_start);
if ($bucket_part =~ /^\s*(?:bucket_target_sys|target)\b/s) {
$hit = 1;
last;
}
$bucket_part_start = $k + 1;
}
}
}
if ($hit) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:crate::{$part}\n";
}
$part_start = $j + 1;
}
}
}
' |
rg -v '^crates/ecstore/src/bucket/replication/replication_target_boundary\.rs:' || true
) >"$REPLICATION_TARGET_BOUNDARY_BYPASS_HITS_FILE"
@@ -2629,9 +2741,128 @@ fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'crate::bucket::metadata_sys|metadata_sys::get_replication_config' \
crates/ecstore/src/bucket/replication \
--glob '*.rs' |
find crates/ecstore/src/bucket/replication -type f -name '*.rs' -print0 |
xargs -0 perl -0ne '
while (/(?:crate::bucket::metadata_sys|super(?:::super)+::metadata_sys\b|metadata_sys::get_replication_config|crate::disk::(?:BUCKET_META_PREFIX|RUSTFS_META_BUCKET)\b)/sg) {
my $prefix = substr($_, 0, $-[0]);
my $line = ($prefix =~ tr/\n//) + 1;
my $match = $&;
$match =~ s/\s+/ /g;
print "$ARGV:$line:$match\n";
}
while (/super(?:::super)+::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
if ($part =~ /^\s*metadata_sys\b/s) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:super::super::{$part}\n";
}
$part_start = $j + 1;
}
}
}
while (/crate::disk::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
if ($part =~ /^\s*(?:BUCKET_META_PREFIX|RUSTFS_META_BUCKET)\b/s) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:crate::disk::{$part}\n";
}
$part_start = $j + 1;
}
}
}
while (/crate::\{/g) {
my $start = $-[0];
my $body_start = pos($_);
my $depth = 1;
my $i = $body_start;
while ($i < length($_) && $depth > 0) {
my $ch = substr($_, $i, 1);
$depth++ if $ch eq "{";
$depth-- if $ch eq "}";
$i++;
}
next if $depth != 0;
my $body = substr($_, $body_start, $i - $body_start - 1);
my $part_start = 0;
my $inner_depth = 0;
for (my $j = 0; $j <= length($body); $j++) {
my $ch = $j < length($body) ? substr($body, $j, 1) : ",";
$inner_depth++ if $ch eq "{";
$inner_depth-- if $ch eq "}";
if ($ch eq "," && $inner_depth == 0) {
my $part = substr($body, $part_start, $j - $part_start);
my $hit = $part =~ /^\s*disk\s*::\s*(?:BUCKET_META_PREFIX|RUSTFS_META_BUCKET)\b/s;
if (!$hit && $part =~ /^\s*disk\s*::\s*\{(.*)\}\s*$/s) {
my $disk_body = $1;
my $disk_part_start = 0;
my $disk_depth = 0;
for (my $k = 0; $k <= length($disk_body); $k++) {
my $disk_ch = $k < length($disk_body) ? substr($disk_body, $k, 1) : ",";
$disk_depth++ if $disk_ch eq "{";
$disk_depth-- if $disk_ch eq "}";
if ($disk_ch eq "," && $disk_depth == 0) {
my $disk_part = substr($disk_body, $disk_part_start, $k - $disk_part_start);
if ($disk_part =~ /^\s*(?:BUCKET_META_PREFIX|RUSTFS_META_BUCKET)\b/s) {
$hit = 1;
last;
}
$disk_part_start = $k + 1;
}
}
}
if ($hit) {
my $prefix = substr($_, 0, $start);
my $line = ($prefix =~ tr/\n//) + 1;
$part =~ s/\s+/ /g;
print "$ARGV:$line:crate::{$part}\n";
}
$part_start = $j + 1;
}
}
}
' |
rg -v '^crates/ecstore/src/bucket/replication/replication_metadata_boundary\.rs:' || true
) >"$REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE"