gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-26 16:28:15 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(ecstore): route replication event host (#4126)

Browse Source
This commit is contained in:
Zhengchao An
2026-07-01 06:00:50 +08:00
committed by GitHub
parent e917f1f4be
commit 50a3ed8bf3
5 changed files with 57 additions and 70 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/bucket/replication/README.md
+2 -2
View File
@@ -12,7 +12,7 @@ and lifecycle/heal scheduling paths.
| `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. | 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, 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, versioning systems, storage contracts through the replication storage boundary, runtime sources, notification events, bandwidth reader wrapping, and SetDisks lock timing. |
| `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, versioning systems, storage contracts through the replication storage boundary, 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 and file metadata replication contracts through local boundaries, 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. |
@@ -28,7 +28,7 @@ and lifecycle/heal scheduling paths.
| `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. |
| `ReplicationEventSink` | Notification and audit events for skipped, failed, pending, and completed replication operations. | Event notification service calls and local event host selection are concentrated in `replication_event_sink.rs`. |
| `ReplicationTagFilter` | Decode object tag strings for rule and metadata replication decisions. | Direct bucket tagging helper imports from replication workers. |
| `ReplicationLifecycleBridge` | Lifecycle-originated delete and version-purge scheduling. | Direct coupling between lifecycle worker paths and replication delete scheduling. |
crates/ecstore/src/bucket/replication/replication_event_sink.rs
+7
View File
@@ -14,6 +14,13 @@
pub(crate) use crate::services::event_notification::EventArgs;
use super::runtime_boundary;
pub(crate) fn send_event(args: EventArgs) {
crate::services::event_notification::send_event(args);
}
pub(crate) fn send_local_event(mut args: EventArgs) {
args.host = runtime_boundary::default_local_node_name();
send_event(args);
}
crates/ecstore/src/bucket/replication/replication_resyncer.rs
+34 -67
View File
@@ -14,7 +14,7 @@
use super::replication_bandwidth_boundary;
use super::replication_config_store as config_store;
use super::replication_event_sink::{EventArgs, send_event};
use super::replication_event_sink::{EventArgs, send_event, send_local_event};
use super::replication_filemeta_boundary::{
MrfOpKind, MrfReplicateEntry, REPLICATE_EXISTING, REPLICATE_EXISTING_DELETE, ReplicateDecision, ReplicateObjectInfo,
ReplicateTargetDecision, ReplicatedInfos, ReplicatedTargetInfo, ReplicationAction, ReplicationState, ReplicationStatusType,
@@ -1755,7 +1755,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "replication_config_missing",
"Skipping replication delete because replication config is missing"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -1766,7 +1766,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
@@ -1782,7 +1781,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "replication_config_lookup_failed",
"Skipping replication delete because replication config lookup failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -1793,7 +1792,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -1881,7 +1879,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "replicate_decision_parse_failed",
"Failed to parse replicate decision"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -1892,7 +1890,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -1914,7 +1911,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "ns_lock_unavailable",
"Skipping replication delete"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -1925,7 +1922,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -1945,7 +1941,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "write_lock_unavailable",
"Skipping replication delete"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -1956,7 +1952,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -1994,7 +1989,7 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
reason = "target_client_missing",
"Skipping replication delete because target client is unavailable"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2005,7 +2000,6 @@ pub async fn replicate_delete<S: ReplicationStorage>(dobj: DeletedObjectReplicat
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
continue;
@@ -2248,7 +2242,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
reason = "replication_config_missing",
"Skipping replication force-delete because replication config is missing"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2257,7 +2251,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -2272,7 +2265,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
reason = "replication_config_lookup_failed",
"Skipping replication force-delete because replication config lookup failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2281,7 +2274,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -2304,7 +2296,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
error = %e,
"Skipping replication force-delete"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2313,7 +2305,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -2333,7 +2324,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
error = %e,
"Skipping replication force-delete"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2342,7 +2333,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -2371,7 +2361,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
reason = "target_client_missing",
"Skipping replication force-delete because target client is unavailable"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2380,7 +2370,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
continue;
@@ -2401,7 +2390,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
endpoint = %tgt_client.to_url(),
"Skipping replication force-delete"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationFailed.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2410,7 +2399,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
return;
@@ -2444,7 +2432,7 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
error = %e,
"Replication target operation failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationFailed.to_string(),
bucket_name: bucket.clone(),
object: ObjectInfo {
@@ -2453,7 +2441,6 @@ async fn replicate_force_delete_to_targets<S: ReplicationStorage>(dobj: &Deleted
..Default::default()
},
user_agent: "Internal: [Replication]".to_string(),
host: runtime_sources::default_local_node_name(),
..Default::default()
});
}
@@ -2649,11 +2636,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
reason = "replication_config_missing",
"Skipping replication object because replication config is missing"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2669,11 +2655,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
error = %err,
"Failed to look up replication config for object replication"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2708,11 +2693,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
reason = "ns_lock_create_failed",
"Skipping replication object"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2732,11 +2716,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
reason = "ns_lock_write_lock_failed",
"Skipping replication object"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2757,11 +2740,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
reason = "target_client_missing",
"Skipping replication object target"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2800,11 +2782,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
error = %e,
"Replication resync task failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: roi.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2848,11 +2829,10 @@ pub async fn replicate_object<S: ReplicationStorage>(roi: ReplicateObjectInfo, s
EventName::ObjectReplicationFailed.to_string()
};
send_event(EventArgs {
send_local_event(EventArgs {
event_name,
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2925,11 +2905,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
endpoint = %tgt_client.to_url(),
"Skipping replication object target"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2965,11 +2944,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
"Skipping replication object target"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -2996,11 +2974,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "actual_size_unavailable",
"Skipping replication object target"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3018,11 +2995,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "target_bucket_empty",
"Skipping replication object target"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3113,11 +3089,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
error = %e,
"Replication target operation failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3216,11 +3191,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "target_offline",
"Skipped replication because target is offline"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3255,11 +3229,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "object_reader_unavailable",
"Skipped replication because object reader is unavailable"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: self.to_object_info(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3295,11 +3268,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "actual_size_unavailable",
"Skipped replication because actual object size is unavailable"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3319,11 +3291,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "target_bucket_empty",
"Skipped replication because target bucket is empty"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3382,11 +3353,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "newer_target_version_exists",
"Skipping replication because newer target version exists"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info.clone(),
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3445,11 +3415,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
reason = "head_object_fallback_failed",
"Failed replication head-object fallback"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3472,11 +3441,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
"Skipped replication because head-object failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
@@ -3508,11 +3476,10 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo {
error = %e,
"Replication target operation failed"
);
send_event(EventArgs {
send_local_event(EventArgs {
event_name: EventName::ObjectReplicationNotTracked.to_string(),
bucket_name: bucket.clone(),
object: object_info,
host: runtime_sources::default_local_node_name(),
user_agent: "Internal: [Replication]".to_string(),
..Default::default()
});
docs/architecture/ecstore-module-split-plan.md
+1 -1
View File
@@ -147,7 +147,7 @@ Required contracts before crate movement:
- `ReplicationBandwidthLimiter`: target reader wrapping for replication
bandwidth accounting and throttling.
- `ReplicationEventSink`: notification/audit events for skipped, failed, and
completed replication operations.
completed replication operations, including local event host selection.
- `ReplicationLifecycleBridge`: lifecycle-originated delete and version-purge
scheduling without importing lifecycle internals.
scripts/check_architecture_migration_rules.sh
+13
View File
@@ -201,6 +201,7 @@ REPLICATION_FACADE_BYPASS_HITS_FILE="${TMP_DIR}/replication_facade_bypass_hits.t
REPLICATION_BANDWIDTH_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_bandwidth_boundary_bypass_hits.txt"
REPLICATION_CONFIG_STORE_BYPASS_HITS_FILE="${TMP_DIR}/replication_config_store_bypass_hits.txt"
REPLICATION_EVENT_SINK_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_sink_bypass_hits.txt"
REPLICATION_EVENT_HOST_BYPASS_HITS_FILE="${TMP_DIR}/replication_event_host_bypass_hits.txt"
REPLICATION_FILEMETA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_filemeta_boundary_bypass_hits.txt"
REPLICATION_LOCK_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_lock_boundary_bypass_hits.txt"
REPLICATION_METADATA_BOUNDARY_BYPASS_HITS_FILE="${TMP_DIR}/replication_metadata_boundary_bypass_hits.txt"
@@ -2523,6 +2524,18 @@ if [[ -s "$REPLICATION_EVENT_SINK_BYPASS_HITS_FILE" ]]; then
report_failure "replication event notification access must stay behind replication event sink: $(paste -sd '; ' "$REPLICATION_EVENT_SINK_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename '\b(runtime_sources|runtime_boundary)::default_local_node_name\(\)' \
crates/ecstore/src/bucket/replication \
--glob '*.rs' |
rg -v '^crates/ecstore/src/bucket/replication/replication_event_sink\.rs:' || true
) >"$REPLICATION_EVENT_HOST_BYPASS_HITS_FILE"
if [[ -s "$REPLICATION_EVENT_HOST_BYPASS_HITS_FILE" ]]; then
report_failure "replication event host selection must stay behind replication event sink: $(paste -sd '; ' "$REPLICATION_EVENT_HOST_BYPASS_HITS_FILE")"
fi
(
cd "$ROOT_DIR"
rg -n --with-filename 'crate::config::com::\{[^}]*\b(read_config|save_config)\b|crate::config::com::(read_config|save_config)|\b(read_config|save_config)\(' \