fix(replication): honor target TLS in health checks

crates/e2e_test/src/replication_extension_test.rs

@@ -1940,6 +1940,28 @@ async fn wait_for_remote_target_arn(env: &RustFSTestEnvironment, bucket: &str) -
     Err(format!("site replication did not configure a remote target for bucket {bucket} in time").into())
 }
 
+async fn wait_for_remote_target_health_check(
+    env: &RustFSTestEnvironment,
+    bucket: &str,
+    arn: &str,
+) -> Result<(), Box<dyn Error + Send + Sync>> {
+    for _ in 0..40 {
+        let response = list_replication_targets_request(env, Some(bucket)).await?;
+        if response.status() == StatusCode::OK {
+            let targets: Vec<serde_json::Value> = response.json().await?;
+            if targets.iter().any(|target| {
+                target.get("arn").and_then(|value| value.as_str()) == Some(arn)
+                    && target.get("lastOnline").is_some_and(|value| !value.is_null())
+            }) {
+                return Ok(());
+            }
+        }
+        sleep(Duration::from_millis(250)).await;
+    }
+
+    Err(format!("replication target {arn} did not complete a successful health check in time").into())
+}
+
 async fn site_replication_add(
     env: &RustFSTestEnvironment,
     sites: &[PeerSite],
@@ -2907,6 +2929,8 @@ async fn test_set_remote_target_allows_self_signed_https_target_with_skip_tls_ve
     let target_bucket = "replication-self-signed-ok-dst";
     let object_key = "self-signed-replication.txt";
     let body = "replication over self-signed https should succeed";
+    let post_health_check_key = "self-signed-replication-after-health-check.txt";
+    let post_health_check_body = "replication should remain available after the target health check";
 
     let source_client = source_env.create_s3_client();
     source_client
@@ -2955,6 +2979,24 @@ async fn test_set_remote_target_allows_self_signed_https_target_with_skip_tls_ve
 
     wait_for_replicated_object_over_https(&https_client, &target_env, target_bucket, object_key, body).await?;
 
+    wait_for_remote_target_health_check(&source_env, source_bucket, &target_arn).await?;
+    source_client
+        .put_object()
+        .bucket(source_bucket)
+        .key(post_health_check_key)
+        .body(ByteStream::from(post_health_check_body.as_bytes().to_vec()))
+        .send()
+        .await?;
+
+    wait_for_replicated_object_over_https(
+        &https_client,
+        &target_env,
+        target_bucket,
+        post_health_check_key,
+        post_health_check_body,
+    )
+    .await?;
+
     Ok(())
 }
 

crates/ecstore/src/api/mod.rs

@@ -173,9 +173,8 @@ pub mod bucket {
 
     pub mod replication {
         pub use crate::bucket::replication::replication_pool::{
-            DurableMrfBacklogSummary, DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBacklogObservabilitySummary,
-            MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, durable_mrf_target_backlog_snapshot,
-            mrf_backlog_observability_snapshot,
+            DurableMrfBacklogSummary, DurableMrfBucketBacklog, MrfBacklogObservabilitySummary, MrfBucketBacklogObservability,
+            durable_mrf_backlog_summary_snapshot, mrf_backlog_observability_snapshot,
         };
         pub use crate::bucket::replication::{
             BucketReplicationResyncStatus, BucketStats, DeletedObjectReplicationInfo, DurableMrfBacklog, DynReplicationPool,
@@ -184,13 +183,13 @@ pub mod bucket {
             ReplicationDeleteStateSource, ReplicationHealQueueResult, ReplicationObjectBridge, ReplicationObjectIO,
             ReplicationOperation, ReplicationPoolTrait, ReplicationPriority, ReplicationQueueAdmission, ReplicationScannerBridge,
             ReplicationState, ReplicationStats, ReplicationStatusType, ReplicationStorage, ReplicationTargetValidationError,
-            ReplicationType, ResyncOpts, ResyncStatusType, RuntimeReplicationTargetBacklog, TargetReplicationResyncStatus,
-            VersionPurgeStatusType, delete_replication_state_from_config, delete_replication_version_id,
-            get_global_replication_pool, get_global_replication_stats, init_background_replication, read_durable_mrf_backlog,
-            replication_state_to_filemeta, replication_status_to_filemeta, replication_statuses_map, replication_target_arns,
-            resync_start_conflict_id, should_remove_replication_target, should_schedule_delete_replication,
-            should_use_existing_delete_replication_info, should_use_existing_delete_replication_source,
-            unsupported_replication_config_field, validate_replication_config_target_arns, version_purge_status_to_filemeta,
+            ReplicationType, ResyncOpts, ResyncStatusType, TargetReplicationResyncStatus, VersionPurgeStatusType,
+            delete_replication_state_from_config, delete_replication_version_id, get_global_replication_pool,
+            get_global_replication_stats, init_background_replication, read_durable_mrf_backlog, replication_state_to_filemeta,
+            replication_status_to_filemeta, replication_statuses_map, replication_target_arns, resync_start_conflict_id,
+            should_remove_replication_target, should_schedule_delete_replication, should_use_existing_delete_replication_info,
+            should_use_existing_delete_replication_source, unsupported_replication_config_field,
+            validate_replication_config_target_arns, version_purge_status_to_filemeta,
         };
     }
 

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -44,6 +44,7 @@ use aws_smithy_runtime_api::client::http::{
 use aws_smithy_runtime_api::client::orchestrator::{HttpRequest, HttpResponse};
 use aws_smithy_runtime_api::client::result::ConnectorError;
 use aws_smithy_types::body::SdkBody;
+use futures::{StreamExt, stream};
 use http::{HeaderMap, HeaderName, HeaderValue, StatusCode, Uri};
 use hyper_util::client::legacy::Client as HyperClient;
 use hyper_util::rt::{TokioExecutor, TokioTimer};
@@ -68,6 +69,7 @@ use std::path::Path;
 use std::str::FromStr as _;
 use std::sync::Arc;
 use std::sync::OnceLock;
+use std::sync::Weak;
 use std::time::{Duration, Instant};
 use time::{OffsetDateTime, format_description::well_known::Rfc3339};
 use tokio::sync::Mutex;
@@ -79,6 +81,7 @@ use url::Url;
 use uuid::Uuid;
 
 const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
+const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
 
 pub static GLOBAL_BUCKET_TARGET_SYS: OnceLock<BucketTargetSys> = OnceLock::new();
@@ -255,6 +258,16 @@ fn endpoint_health_key(url: &Url) -> String {
     }
 }
 
+fn target_health(target: &TargetClient) -> EpHealth {
+    let url = target.to_url();
+    EpHealth {
+        endpoint: endpoint_health_key(&url),
+        scheme: url.scheme().to_string(),
+        online: true,
+        ..Default::default()
+    }
+}
+
 fn update_endpoint_health(health: &mut EpHealth, online: bool, latency: Duration, now: OffsetDateTime) {
     let prev_online = health.online;
     health.online = online;
@@ -272,14 +285,26 @@ fn update_endpoint_health(health: &mut EpHealth, online: bool, latency: Duration
     health.offline_duration += latency;
 }
 
+#[cfg(test)]
+#[derive(Clone, Debug)]
+struct TargetClientBuildProbe {
+    arn: String,
+    started: Arc<tokio::sync::Semaphore>,
+    release: Arc<tokio::sync::Semaphore>,
+}
+
 #[derive(Debug, Default)]
 pub struct BucketTargetSys {
     pub arn_remotes_map: Arc<RwLock<HashMap<String, ArnTarget>>>,
     pub targets_map: Arc<RwLock<HashMap<String, Vec<BucketTarget>>>>,
     pub h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
+    target_h_mutex: Arc<RwLock<HashMap<String, EpHealth>>>,
     pub hc_client: Arc<HttpClient>,
     pub a_mutex: Arc<Mutex<HashMap<String, ArnErrs>>>,
     pub arn_errs_map: Arc<RwLock<HashMap<String, ArnErrs>>>,
+    target_update_mutexes: Arc<Mutex<HashMap<String, Weak<Mutex<()>>>>>,
+    #[cfg(test)]
+    target_client_build_probe: Arc<Mutex<Option<TargetClientBuildProbe>>>,
     heartbeat_started: OnceLock<()>,
 }
 
@@ -293,9 +318,13 @@ impl BucketTargetSys {
             arn_remotes_map: Arc::new(RwLock::new(HashMap::new())),
             targets_map: Arc::new(RwLock::new(HashMap::new())),
             h_mutex: Arc::new(RwLock::new(HashMap::new())),
+            target_h_mutex: Arc::new(RwLock::new(HashMap::new())),
             hc_client: Arc::new(HttpClient::new()),
             a_mutex: Arc::new(Mutex::new(HashMap::new())),
             arn_errs_map: Arc::new(RwLock::new(HashMap::new())),
+            target_update_mutexes: Arc::new(Mutex::new(HashMap::new())),
+            #[cfg(test)]
+            target_client_build_probe: Arc::new(Mutex::new(None)),
             heartbeat_started: OnceLock::new(),
         }
     }
@@ -310,6 +339,17 @@ impl BucketTargetSys {
         });
     }
 
+    async fn target_update_mutex(&self, bucket: &str) -> Arc<Mutex<()>> {
+        let mut mutexes = self.target_update_mutexes.lock().await;
+        mutexes.retain(|_, mutex| mutex.strong_count() > 0);
+        if let Some(mutex) = mutexes.get(bucket).and_then(Weak::upgrade) {
+            return mutex;
+        }
+        let mutex = Arc::new(Mutex::new(()));
+        mutexes.insert(bucket.to_string(), Arc::downgrade(&mutex));
+        mutex
+    }
+
     pub async fn is_offline(&self, url: &Url) -> bool {
         let key = endpoint_health_key(url);
         {
@@ -318,7 +358,6 @@ impl BucketTargetSys {
                 return !health.online;
             }
         }
-        // Initialize health check if not exists
         self.init_hc(url).await;
         false
     }
@@ -345,42 +384,121 @@ impl BucketTargetSys {
         );
     }
 
+    pub(crate) async fn is_target_offline(&self, target: &Arc<TargetClient>) -> bool {
+        // Lock order: arn_remotes_map, then target_h_mutex. A stale client must not
+        // read or initialize the health state of its replacement.
+        let remotes = self.arn_remotes_map.read().await;
+        let Some(current) = remotes.get(&target.arn).and_then(|remote| remote.client.as_ref()) else {
+            return true;
+        };
+        if !Arc::ptr_eq(current, target) {
+            return true;
+        }
+        {
+            let health_map = self.target_h_mutex.read().await;
+            if let Some(health) = health_map.get(&target.arn) {
+                return !health.online;
+            }
+        }
+        let mut health_map = self.target_h_mutex.write().await;
+        let health = health_map.entry(target.arn.clone()).or_insert_with(|| target_health(target));
+        !health.online
+    }
+
+    pub(crate) async fn mark_target_offline(&self, target: &Arc<TargetClient>) {
+        // Lock order: arn_remotes_map, then target_h_mutex. Ignore failures reported
+        // by a client that has already been replaced.
+        let remotes = self.arn_remotes_map.read().await;
+        let Some(current) = remotes.get(&target.arn).and_then(|remote| remote.client.as_ref()) else {
+            return;
+        };
+        if !Arc::ptr_eq(current, target) {
+            return;
+        }
+        let mut health_map = self.target_h_mutex.write().await;
+        let health = health_map.entry(target.arn.clone()).or_insert_with(|| target_health(target));
+        update_endpoint_health(health, false, Duration::from_secs(0), OffsetDateTime::now_utc());
+    }
+
+    #[cfg(test)]
+    async fn init_target_health(&self, target: &TargetClient) {
+        let mut health_map = self.target_h_mutex.write().await;
+        health_map.insert(target.arn.clone(), target_health(target));
+        drop(health_map);
+        self.init_hc(&target.to_url()).await;
+    }
+
     pub async fn heartbeat(&self) {
         // Probe interval: `RUSTFS_REPL_HEALTH_CHECK_INTERVAL_MS` (default 5000ms,
         // clamped to >=10ms), read once when the heartbeat task starts.
         let mut interval = tokio::time::interval(crate::bucket::replication::replication_timing::health_check_interval());
         loop {
             interval.tick().await;
-
-            let endpoints = {
-                let health_map = self.h_mutex.read().await;
-                health_map
-                    .iter()
-                    .map(|(endpoint, health)| (endpoint.clone(), health.scheme.clone()))
-                    .collect::<Vec<_>>()
-            };
-
-            for (endpoint, scheme) in endpoints {
-                // Perform health check
-                let start = Instant::now();
-                let online = self.check_endpoint_health(&endpoint, &scheme).await;
-                let duration = start.elapsed();
-
-                {
-                    let mut health_map = self.h_mutex.write().await;
-                    if let Some(health) = health_map.get_mut(&endpoint) {
-                        update_endpoint_health(health, online, duration, OffsetDateTime::now_utc());
-                    }
-                }
-            }
+            self.heartbeat_once().await;
         }
     }
 
-    async fn check_endpoint_health(&self, endpoint: &str, scheme: &str) -> bool {
-        let scheme = if scheme.is_empty() { "https" } else { scheme };
-        let url = format!("{scheme}://{endpoint}/");
-        match self.hc_client.get(url).timeout(Duration::from_secs(3)).send().await {
-            Ok(response) => response.status().as_u16() < 500,
+    async fn heartbeat_once(&self) {
+        let targets = {
+            let remotes = self.arn_remotes_map.read().await;
+            remotes
+                .values()
+                .filter_map(|target| target.client.clone())
+                .collect::<Vec<_>>()
+        };
+
+        let checks = stream::iter(targets.into_iter().map(|target| async move {
+            let start = Instant::now();
+            let online = Self::check_endpoint_health(&target).await;
+            (target, online, start.elapsed())
+        }));
+        let mut checks = checks.buffer_unordered(MAX_CONCURRENT_TARGET_HEALTH_CHECKS);
+        let mut endpoint_checks = HashMap::<String, (String, bool, Duration)>::new();
+
+        while let Some((target, online, duration)) = checks.next().await {
+            let url = target.to_url();
+
+            {
+                // Lock order: arn_remotes_map, then target_h_mutex. Keeping the remote
+                // read guard prevents a replaced client from receiving stale health.
+                let remotes = self.arn_remotes_map.read().await;
+                let Some(current) = remotes.get(&target.arn).and_then(|remote| remote.client.as_ref()) else {
+                    continue;
+                };
+                if !Arc::ptr_eq(current, &target) {
+                    continue;
+                }
+                let mut health_map = self.target_h_mutex.write().await;
+                let health = health_map.entry(target.arn.clone()).or_insert_with(|| target_health(&target));
+                update_endpoint_health(health, online, duration, OffsetDateTime::now_utc());
+            }
+
+            let endpoint = endpoint_health_key(&url);
+            endpoint_checks
+                .entry(endpoint)
+                .and_modify(|(_, endpoint_online, endpoint_duration)| {
+                    *endpoint_online |= online;
+                    *endpoint_duration = (*endpoint_duration).max(duration);
+                })
+                .or_insert_with(|| (url.scheme().to_string(), online, duration));
+        }
+
+        let mut health_map = self.h_mutex.write().await;
+        for (endpoint, (scheme, online, duration)) in endpoint_checks {
+            let health = health_map.entry(endpoint.clone()).or_insert_with(|| EpHealth {
+                endpoint,
+                scheme,
+                online: true,
+                ..Default::default()
+            });
+            update_endpoint_health(health, online, duration, OffsetDateTime::now_utc());
+        }
+    }
+
+    async fn check_endpoint_health(target: &TargetClient) -> bool {
+        match tokio::time::timeout(Duration::from_secs(3), target.client.head_bucket().bucket(&target.bucket).send()).await {
+            Ok(Ok(_)) => true,
+            Ok(Err(err)) => err.raw_response().is_some_and(|response| response.status().as_u16() < 500),
             Err(_) => false,
         }
     }
@@ -390,15 +508,20 @@ impl BucketTargetSys {
         health_map.clone()
     }
 
+    async fn target_health_stats(&self) -> HashMap<String, EpHealth> {
+        let health_map = self.target_h_mutex.read().await;
+        health_map.clone()
+    }
+
     pub async fn list_targets(&self, bucket: &str, arn_type: &str) -> Vec<BucketTarget> {
-        let health_stats = self.health_stats().await;
+        let health_stats = self.target_health_stats().await;
         let mut targets = Vec::new();
 
         if !bucket.is_empty() {
             if let Ok(bucket_targets) = self.list_bucket_targets(bucket).await {
                 for mut target in bucket_targets.targets {
                     if arn_type.is_empty() || target.target_type.to_string() == arn_type {
-                        if let Some(health) = health_stats.get(&target.endpoint) {
+                        if let Some(health) = health_stats.get(&target.arn) {
                             target.total_downtime = health.offline_duration;
                             target.online = health.online;
                             target.last_online = health.last_online;
@@ -420,7 +543,7 @@ impl BucketTargetSys {
         for bucket_targets in targets_map.values() {
             for mut target in bucket_targets.iter().cloned() {
                 if arn_type.is_empty() || target.target_type.to_string() == arn_type {
-                    if let Some(health) = health_stats.get(&target.endpoint) {
+                    if let Some(health) = health_stats.get(&target.arn) {
                         target.total_downtime = health.offline_duration;
                         target.online = health.online;
                         target.last_online = health.last_online;
@@ -453,12 +576,18 @@ impl BucketTargetSys {
     }
 
     pub async fn delete(&self, bucket: &str) {
+        let update_mutex = self.target_update_mutex(bucket).await;
+        let _update_guard = update_mutex.lock().await;
+
+        // Lock order: targets_map, then arn_remotes_map, then target_h_mutex.
         let mut targets_map = self.targets_map.write().await;
         let mut arn_remotes_map = self.arn_remotes_map.write().await;
+        let mut health_map = self.target_h_mutex.write().await;
 
         if let Some(targets) = targets_map.remove(bucket) {
             for target in targets {
                 arn_remotes_map.remove(&target.arn);
+                health_map.remove(&target.arn);
             }
         }
     }
@@ -690,6 +819,22 @@ impl BucketTargetSys {
     }
 
     pub async fn get_remote_target_client_internal(&self, target: &BucketTarget) -> Result<TargetClient, BucketTargetError> {
+        #[cfg(test)]
+        {
+            let probe = self.target_client_build_probe.lock().await.clone();
+            if let Some(probe) = probe
+                && probe.arn == target.arn
+            {
+                probe.started.add_permits(1);
+                probe
+                    .release
+                    .acquire()
+                    .await
+                    .expect("test probe semaphore should remain open")
+                    .forget();
+            }
+        }
+
         let Some(credentials) = &target.credentials else {
             return Err(BucketTargetError::BucketRemoteTargetNotFound {
                 bucket: target.target_bucket.clone(),
@@ -792,12 +937,25 @@ impl BucketTargetSys {
     }
 
     pub async fn update_all_targets(&self, bucket: &str, targets: Option<&BucketTargets>) {
+        let update_mutex = self.target_update_mutex(bucket).await;
+        let _update_guard = update_mutex.lock().await;
+
+        let mut clients = Vec::new();
+        if let Some(new_targets) = targets {
+            for target in &new_targets.targets {
+                clients.push((target, self.get_remote_target_client_internal(target).await.map(Arc::new)));
+            }
+        }
+
+        // Lock order: targets_map, then arn_remotes_map, then target_h_mutex.
         let mut targets_map = self.targets_map.write().await;
         let mut arn_remotes_map = self.arn_remotes_map.write().await;
+        let mut health_map = self.target_h_mutex.write().await;
         // Remove existing targets
         if let Some(existing_targets) = targets_map.remove(bucket) {
             for target in existing_targets {
                 arn_remotes_map.remove(&target.arn);
+                health_map.remove(&target.arn);
                 self.update_bandwidth_limit(bucket, &target.arn, 0);
             }
         }
@@ -806,16 +964,17 @@ impl BucketTargetSys {
         if let Some(new_targets) = targets
             && !new_targets.is_empty()
         {
-            for target in &new_targets.targets {
-                match self.get_remote_target_client_internal(target).await {
+            for (target, client) in clients {
+                match client {
                     Ok(client) => {
                         arn_remotes_map.insert(
                             target.arn.clone(),
                             ArnTarget {
-                                client: Some(Arc::new(client)),
+                                client: Some(client.clone()),
                                 last_refresh: OffsetDateTime::now_utc(),
                             },
                         );
+                        health_map.insert(client.arn.clone(), target_health(&client));
                         self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
                     }
                     // The target stays in `targets_map`, so it keeps showing up in
@@ -845,25 +1004,7 @@ impl BucketTargetSys {
             return;
         }
 
-        for target in config.targets.iter() {
-            let cli = match self.get_remote_target_client_internal(target).await {
-                Ok(cli) => cli,
-                Err(e) => {
-                    warn!("get_remote_target_client_internal error:{}", e);
-                    continue;
-                }
-            };
-
-            {
-                let arn_target = ArnTarget::with_client(Arc::new(cli));
-                let mut arn_remotes_map = self.arn_remotes_map.write().await;
-                arn_remotes_map.insert(target.arn.clone(), arn_target);
-            }
-            self.update_bandwidth_limit(bucket, &target.arn, target.bandwidth_limit);
-        }
-
-        let mut targets_map = self.targets_map.write().await;
-        targets_map.insert(bucket.to_string(), config.targets.clone());
+        self.update_all_targets(bucket, Some(config)).await;
     }
 
     // getRemoteARN gets existing ARN for an endpoint or generates a new one.
@@ -1995,7 +2136,7 @@ mod tests {
     use super::*;
     use rcgen::generate_simple_self_signed;
 
-    fn spawn_single_request_https_server(cert: &rcgen::CertifiedKey<rcgen::KeyPair>) -> (u16, std::thread::JoinHandle<()>) {
+    fn spawn_https_server(cert: &rcgen::CertifiedKey<rcgen::KeyPair>, requests: usize) -> (u16, std::thread::JoinHandle<()>) {
         use std::io::{Read, Write};
 
         ensure_rustls_crypto_provider();
@@ -2014,42 +2155,116 @@ mod tests {
             .expect("test TLS server config should build");
 
         let handle = std::thread::spawn(move || {
-            let (stream, _) = listener.accept().expect("test TLS client should connect");
-            stream
-                .set_read_timeout(Some(Duration::from_secs(10)))
-                .expect("test TLS read timeout should configure");
-            stream
-                .set_write_timeout(Some(Duration::from_secs(10)))
-                .expect("test TLS write timeout should configure");
-            let connection = rustls::ServerConnection::new(Arc::new(server_config)).expect("test TLS connection should build");
-            let mut stream = rustls::StreamOwned::new(connection, stream);
-            let mut request = [0_u8; 8192];
-            let _ = stream.read(&mut request).expect("test TLS request should be readable");
-            stream
-                .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
-                .expect("test TLS response should be written");
-            stream.flush().expect("test TLS response should flush");
+            let server_config = Arc::new(server_config);
+            for _ in 0..requests {
+                let (stream, _) = listener.accept().expect("test TLS client should connect");
+                stream
+                    .set_read_timeout(Some(Duration::from_secs(10)))
+                    .expect("test TLS read timeout should configure");
+                stream
+                    .set_write_timeout(Some(Duration::from_secs(10)))
+                    .expect("test TLS write timeout should configure");
+                let connection = rustls::ServerConnection::new(server_config.clone()).expect("test TLS connection should build");
+                let mut stream = rustls::StreamOwned::new(connection, stream);
+                let mut request = [0_u8; 8192];
+                if stream.read(&mut request).is_err() {
+                    continue;
+                }
+                stream
+                    .write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                    .expect("test TLS response should be written");
+                stream.flush().expect("test TLS response should flush");
+            }
         });
 
         (port, handle)
     }
 
-    fn s3_client_with_http_client(port: u16, http_client: SharedHttpClient) -> S3Client {
+    fn spawn_http_status_server(status: u16) -> (u16, std::thread::JoinHandle<()>) {
+        use std::io::{Read, Write};
+
+        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("test HTTP listener should bind");
+        let port = listener
+            .local_addr()
+            .expect("test HTTP listener should have an address")
+            .port();
+        let handle = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().expect("test HTTP client should connect");
+            let mut request = [0_u8; 8192];
+            let bytes_read = stream.read(&mut request).expect("test HTTP request should be read");
+            assert!(bytes_read > 0, "test HTTP request should not be empty");
+            write!(stream, "HTTP/1.1 {status} Test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("test HTTP response should be written");
+        });
+        (port, handle)
+    }
+
+    fn spawn_delayed_http_server() -> (
+        u16,
+        tokio::sync::oneshot::Receiver<()>,
+        std::sync::mpsc::Sender<()>,
+        std::thread::JoinHandle<()>,
+    ) {
+        use std::io::{Read, Write};
+
+        let listener = std::net::TcpListener::bind(("127.0.0.1", 0)).expect("test HTTP listener should bind");
+        let port = listener
+            .local_addr()
+            .expect("test HTTP listener should have an address")
+            .port();
+        let (accepted_tx, accepted_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = std::sync::mpsc::channel();
+        let handle = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().expect("test HTTP client should connect");
+            let mut request = [0_u8; 8192];
+            let bytes_read = stream.read(&mut request).expect("test HTTP request should be read");
+            assert!(bytes_read > 0, "test HTTP request should not be empty");
+            accepted_tx.send(()).expect("test should wait for request");
+            release_rx.recv().expect("test should release response");
+            stream
+                .write_all(b"HTTP/1.1 500 Test\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("test HTTP response should be written");
+        });
+        (port, accepted_rx, release_tx, handle)
+    }
+
+    fn s3_client_for_test(port: u16, http_client: Option<SharedHttpClient>) -> S3Client {
+        s3_client_for_endpoint_test(format!("https://localhost:{port}"), http_client)
+    }
+
+    fn s3_client_for_endpoint_test(endpoint: String, http_client: Option<SharedHttpClient>) -> S3Client {
         let credentials = SdkCredentials::builder()
             .access_key_id("test-access")
             .secret_access_key("test-secret")
             .provider_name("bucket_target_tls_test")
             .build();
-        let config = S3Config::builder()
-            .endpoint_url(format!("https://localhost:{port}"))
+        let mut config = S3Config::builder()
+            .endpoint_url(endpoint)
             .credentials_provider(SharedCredentialsProvider::new(credentials))
             .region(SdkRegion::new("us-east-1"))
             .force_path_style(true)
-            .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
-            .http_client(http_client)
-            .build();
+            .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest());
+        if let Some(http_client) = http_client {
+            config = config.http_client(http_client);
+        }
 
-        S3Client::from_conf(config)
+        S3Client::from_conf(config.build())
+    }
+
+    fn target_client_for_test(arn: &str, endpoint: String, client: S3Client) -> Arc<TargetClient> {
+        Arc::new(TargetClient {
+            endpoint,
+            credentials: None,
+            bucket: "target-bucket".to_string(),
+            storage_class: String::new(),
+            disable_proxy: false,
+            arn: arn.to_string(),
+            reset_id: String::new(),
+            secure: true,
+            health_check_duration: Duration::from_secs(5),
+            replicate_sync: false,
+            client: Arc::new(client),
+        })
     }
 
     #[test]
@@ -2169,17 +2384,32 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn list_targets_applies_health_stats_for_endpoint_with_port() {
+    async fn list_targets_applies_health_stats_by_arn_and_preserves_endpoint_port() {
         let sys = BucketTargetSys::default();
-        let url = Url::parse("https://remote.example:9443").expect("url should parse");
-        sys.init_hc(&url).await;
-        sys.mark_offline(&url).await;
+        let arn = "arn:rustfs:replication:us-east-1:bucket:id";
+        let endpoint = "https://remote.example:9443".to_string();
+        let client = target_client_for_test(
+            arn,
+            endpoint.clone(),
+            S3Client::from_conf(
+                S3Config::builder()
+                    .endpoint_url(endpoint)
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
+        );
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(arn.to_string(), ArnTarget::with_client(client.clone()));
+        sys.init_target_health(&client).await;
+        sys.mark_target_offline(&client).await;
 
         sys.targets_map.write().await.insert(
             "bucket".to_string(),
             vec![BucketTarget {
                 endpoint: "remote.example:9443".to_string(),
-                arn: "arn:rustfs:replication:us-east-1:bucket:id".to_string(),
+                arn: arn.to_string(),
                 target_type: BucketTargetType::ReplicationService,
                 ..Default::default()
             }],
@@ -2190,6 +2420,97 @@ mod tests {
         assert_eq!(targets.len(), 1);
         assert!(!targets[0].online);
         assert_eq!(targets[0].offline_count, 1);
+        assert_eq!(sys.target_health_stats().await[arn].endpoint, "remote.example:9443");
+    }
+
+    #[tokio::test]
+    async fn target_health_is_isolated_by_arn_for_shared_endpoint() {
+        let sys = BucketTargetSys::default();
+        let endpoint = "https://shared.example:9443".to_string();
+        let config = || {
+            S3Config::builder()
+                .endpoint_url(endpoint.clone())
+                .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                .build()
+        };
+        let first = target_client_for_test("arn:first", endpoint.clone(), S3Client::from_conf(config()));
+        let second = target_client_for_test("arn:second", endpoint.clone(), S3Client::from_conf(config()));
+
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(first.arn.clone(), ArnTarget::with_client(first.clone()));
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(second.arn.clone(), ArnTarget::with_client(second.clone()));
+        sys.init_target_health(&first).await;
+        sys.init_target_health(&second).await;
+        sys.mark_target_offline(&first).await;
+
+        assert!(sys.is_target_offline(&first).await);
+        assert!(!sys.is_target_offline(&second).await);
+    }
+
+    #[tokio::test]
+    async fn stale_client_cannot_change_replacement_health_for_same_arn() {
+        let sys = BucketTargetSys::default();
+        let arn = "arn:replacement";
+        let config = |endpoint: &str| {
+            S3Config::builder()
+                .endpoint_url(endpoint)
+                .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                .build()
+        };
+        let stale = target_client_for_test(
+            arn,
+            "https://stale.example:9443".to_string(),
+            S3Client::from_conf(config("https://stale.example:9443")),
+        );
+        let current = target_client_for_test(
+            arn,
+            "https://current.example:9443".to_string(),
+            S3Client::from_conf(config("https://current.example:9443")),
+        );
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(arn.to_string(), ArnTarget::with_client(current.clone()));
+        sys.init_target_health(&current).await;
+
+        sys.mark_target_offline(&stale).await;
+
+        assert!(sys.is_target_offline(&stale).await);
+        assert!(!sys.is_target_offline(&current).await);
+    }
+
+    #[tokio::test]
+    async fn delete_removes_target_health_by_arn() {
+        let sys = BucketTargetSys::default();
+        let arn = "arn:delete";
+        let endpoint = "https://delete.example:9443".to_string();
+        let client = target_client_for_test(
+            arn,
+            endpoint.clone(),
+            S3Client::from_conf(
+                S3Config::builder()
+                    .endpoint_url(endpoint)
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
+        );
+        sys.targets_map.write().await.insert(
+            "bucket".to_string(),
+            vec![BucketTarget {
+                arn: arn.to_string(),
+                ..Default::default()
+            }],
+        );
+        sys.init_target_health(&client).await;
+
+        sys.delete("bucket").await;
+
+        assert!(!sys.target_health_stats().await.contains_key(arn));
     }
 
     #[test]
@@ -2375,6 +2696,216 @@ mod tests {
         assert_eq!(client.endpoint, "https://192.168.1.10:9000");
     }
 
+    #[tokio::test]
+    async fn target_health_check_rejects_untrusted_self_signed_certificate() {
+        let cert = generate_simple_self_signed(vec!["localhost".to_string()]).expect("certificate should generate");
+        let (port, server) = spawn_https_server(&cert, 1);
+        let target =
+            target_client_for_test("arn:default-tls", format!("https://localhost:{port}"), s3_client_for_test(port, None));
+
+        assert!(!BucketTargetSys::check_endpoint_health(&target).await);
+        server.join().expect("test TLS server should stop");
+    }
+
+    #[tokio::test]
+    async fn target_health_check_honors_skip_tls_verify_client() {
+        let cert = generate_simple_self_signed(vec!["localhost".to_string()]).expect("certificate should generate");
+        let (port, server) = spawn_https_server(&cert, 1);
+        let target = target_client_for_test(
+            "arn:skip-tls",
+            format!("https://localhost:{port}"),
+            s3_client_for_test(port, Some(build_insecure_aws_s3_http_client())),
+        );
+
+        assert!(BucketTargetSys::check_endpoint_health(&target).await);
+        server.join().expect("test TLS server should stop");
+    }
+
+    #[tokio::test]
+    async fn target_health_check_honors_custom_ca_client() {
+        let cert = generate_simple_self_signed(vec!["localhost".to_string()]).expect("certificate should generate");
+        let http_client = build_aws_s3_http_client_from_target_ca_pem(&cert.cert.pem())
+            .await
+            .expect("custom CA client should build");
+        let (port, server) = spawn_https_server(&cert, 1);
+        let target = target_client_for_test(
+            "arn:custom-ca",
+            format!("https://localhost:{port}"),
+            s3_client_for_test(port, Some(http_client)),
+        );
+
+        assert!(BucketTargetSys::check_endpoint_health(&target).await);
+        server.join().expect("test TLS server should stop");
+    }
+
+    #[tokio::test]
+    async fn target_health_check_treats_client_errors_as_online_and_server_errors_as_offline() {
+        for (status, expected_online) in [(403, true), (500, false)] {
+            let (port, server) = spawn_http_status_server(status);
+            let endpoint = format!("http://127.0.0.1:{port}");
+            let target = target_client_for_test(
+                &format!("arn:http-{status}"),
+                endpoint.clone(),
+                s3_client_for_endpoint_test(endpoint, None),
+            );
+
+            assert_eq!(BucketTargetSys::check_endpoint_health(&target).await, expected_online);
+            server.join().expect("test HTTP server should stop");
+        }
+    }
+
+    #[tokio::test]
+    async fn heartbeat_keeps_tls_health_isolated_by_arn_for_shared_endpoint() {
+        let sys = BucketTargetSys::default();
+        let cert = generate_simple_self_signed(vec!["localhost".to_string()]).expect("certificate should generate");
+        let (port, server) = spawn_https_server(&cert, 2);
+        let endpoint = format!("https://localhost:{port}");
+        let strict = target_client_for_test("arn:strict", endpoint.clone(), s3_client_for_test(port, None));
+        let insecure = target_client_for_test(
+            "arn:insecure",
+            endpoint,
+            s3_client_for_test(port, Some(build_insecure_aws_s3_http_client())),
+        );
+        {
+            let mut remotes = sys.arn_remotes_map.write().await;
+            remotes.insert(strict.arn.clone(), ArnTarget::with_client(strict.clone()));
+            remotes.insert(insecure.arn.clone(), ArnTarget::with_client(insecure.clone()));
+        }
+
+        sys.heartbeat_once().await;
+
+        assert!(sys.is_target_offline(&strict).await);
+        assert!(!sys.is_target_offline(&insecure).await);
+        server.join().expect("test TLS server should stop");
+    }
+
+    #[tokio::test]
+    async fn heartbeat_discards_result_from_replaced_client_with_same_arn() {
+        let sys = Arc::new(BucketTargetSys::default());
+        let (port, accepted, release, server) = spawn_delayed_http_server();
+        let endpoint = format!("http://127.0.0.1:{port}");
+        let stale = target_client_for_test("arn:replacement", endpoint.clone(), s3_client_for_endpoint_test(endpoint, None));
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(stale.arn.clone(), ArnTarget::with_client(stale));
+        let heartbeat_sys = sys.clone();
+        let heartbeat = tokio::spawn(async move { heartbeat_sys.heartbeat_once().await });
+        accepted.await.expect("heartbeat request should reach test server");
+
+        let replacement_endpoint = "https://replacement.example:9443".to_string();
+        let replacement = target_client_for_test(
+            "arn:replacement",
+            replacement_endpoint.clone(),
+            S3Client::from_conf(
+                S3Config::builder()
+                    .endpoint_url(replacement_endpoint)
+                    .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
+                    .build(),
+            ),
+        );
+        sys.arn_remotes_map
+            .write()
+            .await
+            .insert(replacement.arn.clone(), ArnTarget::with_client(replacement.clone()));
+        sys.init_target_health(&replacement).await;
+        release.send(()).expect("stale heartbeat response should be released");
+        heartbeat.await.expect("heartbeat should finish");
+
+        assert!(!sys.is_target_offline(&replacement).await);
+        server.join().expect("test HTTP server should stop");
+    }
+
+    #[tokio::test]
+    async fn target_update_mutex_reuses_live_lock_and_reclaims_dead_entries() {
+        let sys = BucketTargetSys::default();
+        let first = sys.target_update_mutex("first").await;
+        let same = sys.target_update_mutex("first").await;
+        assert!(Arc::ptr_eq(&first, &same));
+        drop(first);
+        drop(same);
+
+        let _second = sys.target_update_mutex("second").await;
+        let mutexes = sys.target_update_mutexes.lock().await;
+        assert!(!mutexes.contains_key("first"));
+        assert!(mutexes.contains_key("second"));
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn target_updates_serialize_client_build_through_publication_per_bucket() {
+        let sys = Arc::new(BucketTargetSys::default());
+        let started = Arc::new(tokio::sync::Semaphore::new(0));
+        let release = Arc::new(tokio::sync::Semaphore::new(0));
+        *sys.target_client_build_probe.lock().await = Some(TargetClientBuildProbe {
+            arn: "arn:first".to_string(),
+            started: started.clone(),
+            release: release.clone(),
+        });
+        let target = |arn: &str| BucketTarget {
+            arn: arn.to_string(),
+            endpoint: "192.168.1.10:9000".to_string(),
+            target_bucket: "target-bucket".to_string(),
+            region: "us-east-1".to_string(),
+            credentials: Some(Credentials {
+                access_key: "access".to_string(),
+                secret_key: "secret".to_string(),
+                session_token: None,
+                expiration: None,
+            }),
+            ..Default::default()
+        };
+        let first_targets = BucketTargets {
+            targets: vec![target("arn:first")],
+        };
+        let second_targets = BucketTargets {
+            targets: vec![target("arn:second")],
+        };
+        let first_sys = sys.clone();
+        let first = tokio::spawn(async move {
+            first_sys.update_all_targets("bucket", Some(&first_targets)).await;
+        });
+        tokio::time::timeout(Duration::from_secs(2), started.acquire())
+            .await
+            .expect("first client build should start")
+            .expect("first started semaphore should remain open")
+            .forget();
+        let second_started = Arc::new(tokio::sync::Semaphore::new(0));
+        let second_release = Arc::new(tokio::sync::Semaphore::new(0));
+        *sys.target_client_build_probe.lock().await = Some(TargetClientBuildProbe {
+            arn: "arn:second".to_string(),
+            started: second_started.clone(),
+            release: second_release.clone(),
+        });
+        let second_sys = sys.clone();
+        let second = tokio::spawn(async move {
+            second_sys.update_all_targets("bucket", Some(&second_targets)).await;
+        });
+        assert!(
+            tokio::time::timeout(Duration::from_millis(50), second_started.acquire())
+                .await
+                .is_err()
+        );
+        assert!(!sys.targets_map.read().await.contains_key("bucket"));
+
+        release.add_permits(1);
+        tokio::time::timeout(Duration::from_secs(2), first)
+            .await
+            .expect("first target update should not stall")
+            .expect("first target update should finish");
+        tokio::time::timeout(Duration::from_secs(1), second_started.acquire())
+            .await
+            .expect("second client build should start after first update publishes")
+            .expect("second started semaphore should remain open")
+            .forget();
+        second_release.add_permits(1);
+        tokio::time::timeout(Duration::from_secs(2), second)
+            .await
+            .expect("second target update should not stall")
+            .expect("second target update should finish");
+        let targets = sys.targets_map.read().await;
+        assert_eq!(targets["bucket"][0].arn, "arn:second");
+    }
+
     #[tokio::test]
     async fn replication_trust_store_composes_system_global_and_target_roots_for_real_tls() {
         let tls_dir = tempfile::tempdir().expect("temporary TLS directory should be created");
@@ -2396,8 +2927,8 @@ mod tests {
         );
         let http_client = build_aws_s3_http_client_with_trust_store(trust_store).expect("composed TLS client should build");
 
-        let (global_port, global_server) = spawn_single_request_https_server(&global_ca);
-        s3_client_with_http_client(global_port, http_client.clone())
+        let (global_port, global_server) = spawn_https_server(&global_ca, 1);
+        s3_client_for_test(global_port, Some(http_client.clone()))
             .head_bucket()
             .bucket("test-bucket")
             .send()
@@ -2405,8 +2936,8 @@ mod tests {
             .expect("global RUSTFS_TLS_PATH CA should authenticate its TLS server");
         global_server.join().expect("global CA TLS server should finish");
 
-        let (target_port, target_server) = spawn_single_request_https_server(&target_ca);
-        s3_client_with_http_client(target_port, http_client)
+        let (target_port, target_server) = spawn_https_server(&target_ca, 1);
+        s3_client_for_test(target_port, Some(http_client))
             .head_bucket()
             .bucket("test-bucket")
             .send()

crates/ecstore/src/bucket/replication/mod.rs

@@ -78,7 +78,7 @@ pub use replication_queue_boundary::{
 };
 pub use replication_resync_boundary::{BucketReplicationResyncStatus, ResyncOpts, TargetReplicationResyncStatus};
 pub use replication_scanner_bridge::ReplicationScannerBridge;
-pub use replication_state::{ReplicationStats, RuntimeReplicationTargetBacklog};
+pub use replication_state::ReplicationStats;
 pub use replication_stats_boundary::BucketStats;
 pub use replication_storage_boundary::{ReplicationObjectIO, ReplicationStorage};
 pub(crate) use replication_target_config_bridge::ReplicationTargetConfigBridge;

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -86,26 +86,12 @@ pub struct DurableMrfBucketBacklog {
     pub bytes: u64,
 }
 
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct DurableMrfTargetBacklog {
-    pub bucket: String,
-    pub target_arn: String,
-    pub count: u64,
-    pub bytes: u64,
-}
-
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct DurableMrfBacklogSummary {
     pub available: bool,
     pub buckets: Vec<DurableMrfBucketBacklog>,
 }
 
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-struct DurableMrfBacklogSnapshot {
-    summary: DurableMrfBacklogSummary,
-    targets: Vec<DurableMrfTargetBacklog>,
-}
-
 #[derive(Debug, Clone, Default, PartialEq, Eq)]
 pub struct MrfBucketBacklogObservability {
     pub bucket: String,
@@ -124,8 +110,6 @@ pub struct MrfBacklogObservabilitySummary {
 
 static DURABLE_MRF_BACKLOG_SUMMARY: LazyLock<StdRwLock<DurableMrfBacklogSummary>> =
     LazyLock::new(|| StdRwLock::new(DurableMrfBacklogSummary::default()));
-static DURABLE_MRF_TARGET_BACKLOG: LazyLock<StdRwLock<Vec<DurableMrfTargetBacklog>>> =
-    LazyLock::new(|| StdRwLock::new(Vec::new()));
 static MRF_BACKLOG_OBSERVABILITY: LazyLock<StdRwLock<MrfBacklogObservabilityTracker>> =
     LazyLock::new(|| StdRwLock::new(MrfBacklogObservabilityTracker::default()));
 
@@ -133,15 +117,13 @@ static MRF_BACKLOG_OBSERVABILITY: LazyLock<StdRwLock<MrfBacklogObservabilityTrac
 struct DurableMrfBacklogTracker {
     available: bool,
     buckets: HashMap<String, DurableMrfBucketBacklog>,
-    targets: HashMap<(String, String), DurableMrfTargetBacklog>,
 }
 
 impl DurableMrfBacklogTracker {
-    fn add_entry(&mut self, entry: &MrfReplicateEntry) {
-        let Ok(size) = u64::try_from(entry.size) else {
+    fn add_entry(&mut self, bucket_name: String, entry_size: i64) {
+        let Ok(size) = u64::try_from(entry_size) else {
             self.available = false;
             self.buckets.clear();
-            self.targets.clear();
             return;
         };
 
@@ -149,7 +131,6 @@ impl DurableMrfBacklogTracker {
             return;
         }
 
-        let bucket_name = entry.bucket.clone();
         let bucket = match self.buckets.entry(bucket_name) {
             Entry::Occupied(entry) => entry.into_mut(),
             Entry::Vacant(entry) => {
@@ -162,39 +143,16 @@ impl DurableMrfBacklogTracker {
         };
         bucket.count = bucket.count.saturating_add(1);
         bucket.bytes = bucket.bytes.saturating_add(size);
-
-        for target_arn in &entry.target_arns {
-            if target_arn.is_empty() {
-                continue;
-            }
-            let key = (entry.bucket.clone(), target_arn.clone());
-            let target = match self.targets.entry(key) {
-                Entry::Occupied(entry) => entry.into_mut(),
-                Entry::Vacant(entry) => {
-                    let (bucket, target_arn) = entry.key().clone();
-                    entry.insert(DurableMrfTargetBacklog {
-                        bucket,
-                        target_arn,
-                        ..Default::default()
-                    })
-                }
-            };
-            target.count = target.count.saturating_add(1);
-            target.bytes = target.bytes.saturating_add(size);
-        }
     }
 
-    fn into_snapshot(self) -> DurableMrfBacklogSnapshot {
+    fn into_summary(self) -> DurableMrfBacklogSummary {
         if !self.available {
-            return DurableMrfBacklogSnapshot::default();
+            return DurableMrfBacklogSummary::default();
         }
 
-        DurableMrfBacklogSnapshot {
-            summary: DurableMrfBacklogSummary {
-                available: true,
-                buckets: self.buckets.into_values().collect(),
-            },
-            targets: self.targets.into_values().collect(),
+        DurableMrfBacklogSummary {
+            available: true,
+            buckets: self.buckets.into_values().collect(),
         }
     }
 }
@@ -260,21 +218,7 @@ impl MrfBacklogObservabilityTracker {
     }
 }
 
-fn durable_mrf_backlog_summary_from_entries<'a>(
-    entries: impl IntoIterator<Item = &'a MrfReplicateEntry>,
-) -> DurableMrfBacklogSnapshot {
-    let mut tracker = DurableMrfBacklogTracker {
-        available: true,
-        ..Default::default()
-    };
-    for entry in entries {
-        tracker.add_entry(entry);
-    }
-    tracker.into_snapshot()
-}
-
-#[cfg(test)]
-fn durable_mrf_backlog_summary_from_sizes<I>(entries: I) -> DurableMrfBacklogSnapshot
+fn durable_mrf_backlog_summary_from_sizes<I>(entries: I) -> DurableMrfBacklogSummary
 where
     I: IntoIterator<Item = (String, i64)>,
 {
@@ -283,38 +227,16 @@ where
         ..Default::default()
     };
     for (bucket_name, entry_size) in entries {
-        tracker.add_entry(&MrfReplicateEntry {
-            bucket: bucket_name,
-            object: String::new(),
-            version_id: None,
-            retry_count: 0,
-            size: entry_size,
-            op: MrfOpKind::Object,
-            delete_marker_version_id: None,
-            delete_marker: false,
-            delete_marker_mtime: None,
-            target_arns: Vec::new(),
-        });
-    }
-    tracker.into_snapshot()
-}
-
-fn set_durable_mrf_backlog_snapshot(snapshot: DurableMrfBacklogSnapshot) {
-    match DURABLE_MRF_BACKLOG_SUMMARY.write() {
-        Ok(mut guard) => *guard = snapshot.summary,
-        Err(poisoned) => *poisoned.into_inner() = snapshot.summary,
-    }
-    match DURABLE_MRF_TARGET_BACKLOG.write() {
-        Ok(mut guard) => *guard = snapshot.targets,
-        Err(poisoned) => *poisoned.into_inner() = snapshot.targets,
+        tracker.add_entry(bucket_name, entry_size);
     }
+    tracker.into_summary()
 }
 
 fn set_durable_mrf_backlog_summary(summary: DurableMrfBacklogSummary) {
-    set_durable_mrf_backlog_snapshot(DurableMrfBacklogSnapshot {
-        summary,
-        targets: Vec::new(),
-    });
+    match DURABLE_MRF_BACKLOG_SUMMARY.write() {
+        Ok(mut guard) => *guard = summary,
+        Err(poisoned) => *poisoned.into_inner() = summary,
+    }
 }
 
 pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary {
@@ -324,13 +246,6 @@ pub fn durable_mrf_backlog_summary_snapshot() -> DurableMrfBacklogSummary {
     }
 }
 
-pub fn durable_mrf_target_backlog_snapshot() -> Vec<DurableMrfTargetBacklog> {
-    match DURABLE_MRF_TARGET_BACKLOG.read() {
-        Ok(guard) => guard.clone(),
-        Err(poisoned) => poisoned.into_inner().clone(),
-    }
-}
-
 pub fn mrf_backlog_observability_snapshot() -> MrfBacklogObservabilitySummary {
     match MRF_BACKLOG_OBSERVABILITY.read() {
         Ok(guard) => guard.snapshot(),
@@ -760,7 +675,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
 
     /// Queues a replica task
     pub async fn queue_replica_task(&self, ri: ReplicateObjectInfo) -> ReplicationQueueAdmission {
-        let target_arns = ri.dsc.replicate_target_arns();
         // If object is large, queue it to a static set of large workers
         if should_queue_large_object(ri.size) {
             use std::collections::hash_map::DefaultHasher;
@@ -777,12 +691,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
 
                 if let Some(worker) = lrg_workers.get(index) {
                     self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
-                    self.stats.inc_target_q(&ri.bucket, &target_arns, ri.size);
                     if worker.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() {
                         return ReplicationQueueAdmission::Queued;
                     }
                     self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
-                    self.stats.dec_target_q(&ri.bucket, &target_arns, ri.size);
 
                     // Try to add more workers if possible
                     let max_l_workers = *self.max_l_workers.read().await;
@@ -811,12 +723,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         };
 
         self.stats.inc_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
-        self.stats.inc_target_q(&ri.bucket, &target_arns, ri.size);
         if channel.try_send(ReplicationOperation::Object(Box::new(ri.clone()))).is_ok() {
             return ReplicationQueueAdmission::Queued;
         }
         self.stats.dec_q(&ri.bucket, ri.size, ri.delete_marker, ri.op_type);
-        self.stats.dec_target_q(&ri.bucket, &target_arns, ri.size);
 
         // Queue to MRF if all workers are busy.
         let admission = self.queue_mrf_save_admission(ri.to_mrf_entry(), "object").await;
@@ -830,11 +740,6 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
 
     /// Queues a replica delete task
     pub async fn queue_replica_delete_task(&self, doi: DeletedObjectReplicationInfo) -> ReplicationQueueAdmission {
-        let target_arns = if doi.target_arn.is_empty() {
-            Vec::new()
-        } else {
-            vec![doi.target_arn.clone()]
-        };
         let ch = self
             .worker_queue_channel(&doi.op_type, &doi.bucket, &doi.delete_object.object_name, 0)
             .await;
@@ -844,12 +749,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         };
 
         self.stats.inc_q(&doi.bucket, 0, true, doi.op_type);
-        self.stats.inc_target_q(&doi.bucket, &target_arns, 0);
         if channel.try_send(ReplicationOperation::Delete(Box::new(doi.clone()))).is_ok() {
             return ReplicationQueueAdmission::Queued;
         }
         self.stats.dec_q(&doi.bucket, 0, true, doi.op_type);
-        self.stats.dec_target_q(&doi.bucket, &target_arns, 0);
 
         let admission = self.queue_mrf_save_admission(doi.to_mrf_entry(), "delete").await;
 
@@ -868,11 +771,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         let bucket = entry.bucket.clone();
         let size = entry.size;
         let is_delete = matches!(entry.op, MrfOpKind::Delete);
-        let target_arns = entry.target_arns.clone();
         let admission = queue_mrf_save_entry(&self.mrf_save_tx, entry, queue_type).await;
         if admission == ReplicationQueueAdmission::Queued {
             self.stats.inc_q(&bucket, size, is_delete, ReplicationType::Heal);
-            self.stats.inc_target_q(&bucket, &target_arns, size);
         }
         admission
     }
@@ -926,7 +827,9 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                     return;
                 }
             };
-            set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
+            set_durable_mrf_backlog_summary(durable_mrf_backlog_summary_from_sizes(
+                entries.iter().map(|entry| (entry.bucket.clone(), entry.size)),
+            ));
 
             let total = entries.len();
             let mut queued_count = 0usize;
@@ -1115,7 +1018,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                                 }
                                 continue;
                             }
-                            durable_tracker.add_entry(&e);
+                            durable_tracker.add_entry(e.bucket.clone(), e.size);
                             observe_mrf_pending(&e);
                             pending.push(e);
                             dirty = true;
@@ -1126,7 +1029,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                             if pending.len() - flushed_len >= 1000
                                 && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await
                             {
-                                set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot());
+                                set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
                                 observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                                 dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                                 flushed_len = pending.len();
@@ -1136,7 +1039,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                         None => {
                             // Channel closed (pool shutting down) — final flush.
                             if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await {
-                                set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot());
+                                set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
                                 observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                                 dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                             }
@@ -1145,7 +1048,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                     },
                     _ = interval.tick() => {
                         if dirty && let Some(duration_millis) = flush_mrf_to_disk(&pending, &storage).await {
-                            set_durable_mrf_backlog_snapshot(durable_tracker.clone().into_snapshot());
+                            set_durable_mrf_backlog_summary(durable_tracker.clone().into_summary());
                             observe_mrf_pending_flushed(&pending[flushed_len..], duration_millis);
                             dec_mrf_entries(stats.as_ref(), &pending[flushed_len..]);
                             flushed_len = pending.len();
@@ -1548,7 +1451,6 @@ struct ReplicationBacklogGuard {
     size: i64,
     is_delete_marker: bool,
     op_type: ReplicationType,
-    target_arns: Vec<String>,
 }
 
 impl ReplicationBacklogGuard {
@@ -1559,23 +1461,16 @@ impl ReplicationBacklogGuard {
             size: object.size,
             is_delete_marker: object.delete_marker,
             op_type: object.op_type,
-            target_arns: object.dsc.replicate_target_arns(),
         }
     }
 
     fn for_delete(stats: Arc<ReplicationStats>, delete: &DeletedObjectReplicationInfo) -> Self {
-        let target_arns = if delete.target_arn.is_empty() {
-            Vec::new()
-        } else {
-            vec![delete.target_arn.clone()]
-        };
         Self {
             stats,
             bucket: delete.bucket.clone(),
             size: 0,
             is_delete_marker: true,
             op_type: delete.op_type,
-            target_arns,
         }
     }
 }
@@ -1583,7 +1478,6 @@ impl ReplicationBacklogGuard {
 impl Drop for ReplicationBacklogGuard {
     fn drop(&mut self) {
         self.stats.dec_q(&self.bucket, self.size, self.is_delete_marker, self.op_type);
-        self.stats.dec_target_q(&self.bucket, &self.target_arns, self.size);
     }
 }
 
@@ -1630,7 +1524,6 @@ async fn queue_mrf_save_entry(
 fn dec_mrf_entries(stats: &ReplicationStats, entries: &[MrfReplicateEntry]) {
     for entry in entries {
         stats.dec_q(&entry.bucket, entry.size, matches!(entry.op, MrfOpKind::Delete), ReplicationType::Heal);
-        stats.dec_target_q(&entry.bucket, &entry.target_arns, entry.size);
     }
 }
 
@@ -2022,7 +1915,6 @@ async fn queue_replicate_deletes(batch: ReplicationHealResyncDeletes) -> Replica
 
 #[cfg(test)]
 mod tests {
-    use super::super::replication_filemeta_boundary::ReplicateTargetDecision;
     use super::super::replication_resync_boundary::{decode_mrf_file, encode_mrf_file, encode_resync_file};
     use super::super::replication_storage_boundary::{
         DeletedObject, FileInfo, GetObjectReader, HTTPRangeSpec, ListOperations, ObjectIO, ObjectOperations, PutObjReader,
@@ -2368,22 +2260,6 @@ mod tests {
         (stats.replication_stats.q_stat.curr.count, stats.replication_stats.q_stat.curr.bytes)
     }
 
-    fn current_target_queue(pool: &ReplicationPool<LoadResyncNodeStore>, bucket: &str, target_arn: &str) -> Option<(u64, u64)> {
-        pool.stats
-            .runtime_target_backlog_snapshot()
-            .into_iter()
-            .find(|target| target.bucket == bucket && target.target_arn == target_arn)
-            .map(|target| (target.count, target.bytes))
-    }
-
-    fn test_replicate_decision(target_arns: &[&str]) -> ReplicateDecision {
-        let mut decision = ReplicateDecision::default();
-        for target_arn in target_arns {
-            decision.set(ReplicateTargetDecision::new((*target_arn).to_string(), true, false));
-        }
-        decision
-    }
-
     async fn wait_for_current_queue(pool: &ReplicationPool<LoadResyncNodeStore>, bucket: &str, expected: (i64, i64)) {
         tokio::time::timeout(Duration::from_secs(10), async {
             loop {
@@ -2417,35 +2293,6 @@ mod tests {
         assert_eq!(current_queue(&pool, "admission-bucket").await, (1, 4096));
     }
 
-    #[tokio::test]
-    async fn regular_worker_admission_counts_target_backlog_before_receive() {
-        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
-        let (tx, _rx) = mpsc::channel(1);
-        pool.workers.write().await.push(tx);
-
-        let admission = pool
-            .queue_replica_task(ReplicateObjectInfo {
-                bucket: "target-admission-bucket".to_string(),
-                name: "object".to_string(),
-                size: 4096,
-                op_type: ReplicationType::Object,
-                dsc: test_replicate_decision(&["arn:rustfs:replication:target-b", "arn:rustfs:replication:target-a"]),
-                ..Default::default()
-            })
-            .await;
-
-        assert_eq!(admission, ReplicationQueueAdmission::Queued);
-        assert_eq!(current_queue(&pool, "target-admission-bucket").await, (1, 4096));
-        assert_eq!(
-            current_target_queue(&pool, "target-admission-bucket", "arn:rustfs:replication:target-a"),
-            Some((1, 4096))
-        );
-        assert_eq!(
-            current_target_queue(&pool, "target-admission-bucket", "arn:rustfs:replication:target-b"),
-            Some((1, 4096))
-        );
-    }
-
     #[tokio::test]
     async fn large_worker_admission_counts_channel_backlog_before_receive() {
         let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
@@ -2489,33 +2336,6 @@ mod tests {
         assert_eq!(current_queue(&pool, "delete-admission-bucket").await, (1, 0));
     }
 
-    #[tokio::test]
-    async fn delete_admission_counts_target_backlog_before_receive() {
-        let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
-        let (tx, _rx) = mpsc::channel(1);
-        pool.workers.write().await.push(tx);
-
-        let admission = pool
-            .queue_replica_delete_task(DeletedObjectReplicationInfo {
-                bucket: "delete-target-admission-bucket".to_string(),
-                target_arn: "arn:rustfs:replication:target-a".to_string(),
-                delete_object: ReplicationDeletedObject {
-                    object_name: "deleted-object".to_string(),
-                    ..Default::default()
-                },
-                op_type: ReplicationType::Delete,
-                ..Default::default()
-            })
-            .await;
-
-        assert_eq!(admission, ReplicationQueueAdmission::Queued);
-        assert_eq!(current_queue(&pool, "delete-target-admission-bucket").await, (1, 0));
-        assert_eq!(
-            current_target_queue(&pool, "delete-target-admission-bucket", "arn:rustfs:replication:target-a"),
-            Some((1, 0))
-        );
-    }
-
     #[tokio::test]
     async fn regular_worker_drains_current_backlog_after_processing() {
         let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", empty_resync_shared_state()))).await;
@@ -2882,7 +2702,6 @@ mod tests {
                 name: "fallback-object".to_string(),
                 size: 2048,
                 op_type: ReplicationType::Object,
-                dsc: test_replicate_decision(&["arn:rustfs:replication:target-a"]),
                 ..Default::default()
             })
             .await;
@@ -2891,10 +2710,6 @@ mod tests {
         let queued = pool.stats.get_latest_replication_stats("runtime-backlog").await;
         assert_eq!(queued.replication_stats.q_stat.curr.count, 1);
         assert_eq!(queued.replication_stats.q_stat.curr.bytes, 2048);
-        assert_eq!(
-            current_target_queue(&pool, "runtime-backlog", "arn:rustfs:replication:target-a"),
-            Some((1, 2048))
-        );
     }
 
     #[test]
@@ -2930,7 +2745,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         let second = MrfReplicateEntry {
             object: "second".to_string(),
@@ -2980,7 +2794,6 @@ mod tests {
                 delete_marker_version_id: None,
                 delete_marker: false,
                 delete_marker_mtime: None,
-                target_arns: Vec::new(),
             },
             "test",
         )
@@ -3011,7 +2824,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         observe_mrf_pending(&entry);
 
@@ -3033,14 +2845,12 @@ mod tests {
     async fn replication_backlog_guard_decrements_on_drop() {
         let stats = Arc::new(ReplicationStats::new());
         stats.inc_q("guard-bucket", 256, false, ReplicationType::Object);
-        stats.inc_target_q("guard-bucket", &["arn:rustfs:replication:target-a".to_string()], 256);
 
         {
             let object = ReplicateObjectInfo {
                 bucket: "guard-bucket".to_string(),
                 size: 256,
                 op_type: ReplicationType::Object,
-                dsc: test_replicate_decision(&["arn:rustfs:replication:target-a"]),
                 ..Default::default()
             };
             let _guard = ReplicationBacklogGuard::for_object(stats.clone(), &object);
@@ -3049,30 +2859,6 @@ mod tests {
         let queued = stats.get_latest_replication_stats("guard-bucket").await;
         assert_eq!(queued.replication_stats.q_stat.curr.count, 0);
         assert_eq!(queued.replication_stats.q_stat.curr.bytes, 0);
-        assert!(stats.runtime_target_backlog_snapshot().is_empty());
-    }
-
-    #[test]
-    fn dec_mrf_entries_decrements_target_backlog() {
-        let stats = ReplicationStats::new();
-        let entry = MrfReplicateEntry {
-            bucket: "mrf-target-drain-bucket".to_string(),
-            object: "object".to_string(),
-            version_id: None,
-            retry_count: 1,
-            size: 1024,
-            op: MrfOpKind::Object,
-            delete_marker_version_id: None,
-            delete_marker: false,
-            delete_marker_mtime: None,
-            target_arns: vec!["arn:rustfs:replication:target-a".to_string()],
-        };
-
-        stats.inc_q(&entry.bucket, entry.size, false, ReplicationType::Heal);
-        stats.inc_target_q(&entry.bucket, &entry.target_arns, entry.size);
-        dec_mrf_entries(&stats, std::slice::from_ref(&entry));
-
-        assert!(stats.runtime_target_backlog_snapshot().is_empty());
     }
 
     #[test]
@@ -3087,7 +2873,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         let second = MrfReplicateEntry {
             object: "second".to_string(),
@@ -3199,7 +2984,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
 
         let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode");
@@ -3233,7 +3017,6 @@ mod tests {
             delete_marker_version_id: Some(dm_vid),
             delete_marker: true,
             delete_marker_mtime: Some(mtime_nanos),
-            target_arns: Vec::new(),
         };
 
         let encoded = encode_mrf_file(std::slice::from_ref(&entry)).expect("encode");
@@ -3267,7 +3050,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
 
         let encoded = encode_mrf_file(&[entry]).expect("encode");
@@ -3296,7 +3078,6 @@ mod tests {
                 delete_marker_version_id: None,
                 delete_marker: false,
                 delete_marker_mtime: None,
-                target_arns: Vec::new(),
             },
             MrfReplicateEntry {
                 bucket: "b".to_string(),
@@ -3308,7 +3089,6 @@ mod tests {
                 delete_marker_version_id: Some(del_dm_vid),
                 delete_marker: true,
                 delete_marker_mtime: None,
-                target_arns: Vec::new(),
             },
         ];
 
@@ -3338,7 +3118,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         assert_eq!(obj_entry.op, MrfOpKind::Object);
 
@@ -3353,7 +3132,6 @@ mod tests {
             delete_marker_version_id: Some(Uuid::new_v4()),
             delete_marker: true,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         assert_eq!(del_entry.op, MrfOpKind::Delete);
 
@@ -3369,7 +3147,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         };
         assert_eq!(legacy_entry.op, MrfOpKind::Object, "legacy default must be Object");
     }
@@ -3419,7 +3196,6 @@ mod tests {
         // The "deleteMarkerMtime" key was absent in old files — #[serde(default)] must fill in
         // None so replay falls back to the current time (backlog#867 backward compatibility).
         assert_eq!(entry.delete_marker_mtime, None, "missing deleteMarkerMtime key must default to None");
-        assert!(entry.target_arns.is_empty(), "old MRF entries must not be attributed to a target");
     }
 
     #[test]
@@ -3434,7 +3210,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         }];
         let encoded = encode_mrf_file(&entries).expect("durable MRF backlog should encode");
 
@@ -3451,10 +3226,9 @@ mod tests {
 
     #[test]
     fn durable_mrf_summary_aggregates_entries_by_bucket_for_obs() {
-        let snapshot =
+        let summary =
             durable_mrf_backlog_summary_from_sizes([("b1".to_string(), 1024), ("b1".to_string(), 512), ("b2".to_string(), 0)]);
 
-        let summary = snapshot.summary;
         assert!(summary.available);
         let buckets = summary
             .buckets
@@ -3465,82 +3239,13 @@ mod tests {
         assert_eq!(buckets["b1"].bytes, 1536);
         assert_eq!(buckets["b2"].count, 1);
         assert_eq!(buckets["b2"].bytes, 0);
-        assert!(snapshot.targets.is_empty());
-    }
-
-    #[test]
-    fn durable_mrf_summary_aggregates_target_backlog_without_attributing_legacy_entries() {
-        let entries = vec![
-            MrfReplicateEntry {
-                bucket: "b1".to_string(),
-                object: "object-a".to_string(),
-                version_id: None,
-                retry_count: 0,
-                size: 1024,
-                op: MrfOpKind::Object,
-                delete_marker_version_id: None,
-                delete_marker: false,
-                delete_marker_mtime: None,
-                target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()],
-            },
-            MrfReplicateEntry {
-                bucket: "b1".to_string(),
-                object: "object-b".to_string(),
-                version_id: None,
-                retry_count: 0,
-                size: 512,
-                op: MrfOpKind::Object,
-                delete_marker_version_id: None,
-                delete_marker: false,
-                delete_marker_mtime: None,
-                target_arns: vec!["arn:target-a".to_string()],
-            },
-            MrfReplicateEntry {
-                bucket: "b1".to_string(),
-                object: "legacy-object".to_string(),
-                version_id: None,
-                retry_count: 0,
-                size: 256,
-                op: MrfOpKind::Object,
-                delete_marker_version_id: None,
-                delete_marker: false,
-                delete_marker_mtime: None,
-                target_arns: Vec::new(),
-            },
-        ];
-
-        let snapshot = durable_mrf_backlog_summary_from_entries(&entries);
-
-        let summary = snapshot.summary;
-        assert!(summary.available);
-        let buckets = summary
-            .buckets
-            .into_iter()
-            .map(|bucket| (bucket.bucket.clone(), bucket))
-            .collect::<HashMap<_, _>>();
-        assert_eq!(buckets["b1"].count, 3);
-        assert_eq!(buckets["b1"].bytes, 1792);
-
-        let targets = snapshot
-            .targets
-            .into_iter()
-            .map(|target| ((target.bucket.clone(), target.target_arn.clone()), target))
-            .collect::<HashMap<_, _>>();
-        let target_a = &targets[&("b1".to_string(), "arn:target-a".to_string())];
-        assert_eq!(target_a.count, 2);
-        assert_eq!(target_a.bytes, 1536);
-        let target_b = &targets[&("b1".to_string(), "arn:target-b".to_string())];
-        assert_eq!(target_b.count, 1);
-        assert_eq!(target_b.bytes, 1024);
     }
 
     #[test]
     fn durable_mrf_summary_marks_invalid_sizes_unavailable() {
         let invalid = durable_mrf_backlog_summary_from_sizes([("bucket".to_string(), -1)]);
-        let summary = invalid.summary;
-        assert!(!summary.available);
-        assert!(summary.buckets.is_empty());
-        assert!(invalid.targets.is_empty());
+        assert!(!invalid.available);
+        assert!(invalid.buckets.is_empty());
     }
 
     #[test]
@@ -3559,7 +3264,6 @@ mod tests {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: Vec::new(),
         }])
         .expect("invalid persisted entry should still encode for boundary testing");
         let invalid = durable_mrf_backlog_from_read(Ok(negative));

crates/ecstore/src/bucket/replication/replication_resyncer.rs

@@ -210,7 +210,7 @@ fn is_replication_target_offline_error(err: &(impl Display + ?Sized)) -> bool {
         .any(|marker| message.contains(marker))
 }
 
-async fn mark_replication_target_offline_if_needed(target_client: &TargetClient, err: &(impl Display + ?Sized)) {
+async fn mark_replication_target_offline_if_needed(target_client: &Arc<TargetClient>, err: &(impl Display + ?Sized)) {
     if is_replication_target_offline_error(err) {
         ReplicationTargetStore::mark_target_offline(target_client).await;
     }
@@ -3093,7 +3093,7 @@ mod tests {
     use time::OffsetDateTime;
     use uuid::Uuid;
 
-    fn test_target_client(endpoint: String) -> TargetClient {
+    fn test_target_client(endpoint: String) -> Arc<TargetClient> {
         let config = aws_sdk_s3::Config::builder()
             .endpoint_url(endpoint.clone())
             .region(aws_sdk_s3::config::Region::new("us-east-1"))
@@ -3103,7 +3103,7 @@ mod tests {
             .behavior_version(aws_sdk_s3::config::BehaviorVersion::latest())
             .build();
 
-        TargetClient {
+        Arc::new(TargetClient {
             endpoint,
             credentials: None,
             bucket: "target-bucket".to_string(),
@@ -3115,7 +3115,11 @@ mod tests {
             health_check_duration: std::time::Duration::from_secs(5),
             replicate_sync: false,
             client: Arc::new(aws_sdk_s3::Client::from_conf(config)),
-        }
+        })
+    }
+
+    async fn register_test_target(target: &Arc<TargetClient>) {
+        ReplicationTargetStore::register_test_target(target).await;
     }
 
     #[test]
@@ -3131,6 +3135,7 @@ mod tests {
     async fn replication_target_network_failure_marks_target_offline() {
         let endpoint = format!("http://network-failure-{}.example:9000", Uuid::new_v4());
         let target_client = test_target_client(endpoint);
+        register_test_target(&target_client).await;
 
         assert!(!ReplicationTargetStore::target_is_offline(&target_client).await);
 
@@ -3144,6 +3149,7 @@ mod tests {
     async fn replication_target_service_failure_keeps_target_online() {
         let endpoint = format!("http://service-failure-{}.example:9000", Uuid::new_v4());
         let target_client = test_target_client(endpoint);
+        register_test_target(&target_client).await;
 
         assert!(!ReplicationTargetStore::target_is_offline(&target_client).await);
 

crates/ecstore/src/bucket/replication/replication_state.rs

@@ -22,9 +22,9 @@ use super::replication_stats_boundary::{
     QueueCache, ReplicationMetricScope, SRMetricsSummary, XferStats,
 };
 use super::runtime_boundary as runtime_sources;
-use std::collections::{HashMap, hash_map::Entry};
+use std::collections::HashMap;
 use std::sync::atomic::{AtomicI64, Ordering};
-use std::sync::{Arc, LazyLock, Mutex as StdMutex, Weak};
+use std::sync::{Arc, Mutex as StdMutex};
 use std::time::{Duration, SystemTime};
 use tokio::sync::{Mutex, RwLock};
 use tokio::time::interval;
@@ -153,83 +153,6 @@ pub struct ReplicationStats {
     pub most_recent_stats: Arc<Mutex<HashMap<String, BucketStats>>>,
 }
 
-#[derive(Debug, Clone, Default, PartialEq, Eq)]
-pub struct RuntimeReplicationTargetBacklog {
-    pub bucket: String,
-    pub target_arn: String,
-    pub count: u64,
-    pub bytes: u64,
-}
-
-type TargetQueueKey = (String, String);
-type TargetQueueCache = HashMap<TargetQueueKey, InQueueMetric>;
-
-struct TargetQueueCacheSlot {
-    owner: Weak<StdMutex<QueueCache>>,
-    metrics: TargetQueueCache,
-}
-
-impl TargetQueueCacheSlot {
-    fn new(owner: &Arc<StdMutex<QueueCache>>) -> Self {
-        Self {
-            owner: Arc::downgrade(owner),
-            metrics: TargetQueueCache::default(),
-        }
-    }
-
-    fn belongs_to(&self, owner: &Arc<StdMutex<QueueCache>>) -> bool {
-        self.owner.upgrade().is_some_and(|current| Arc::ptr_eq(&current, owner))
-    }
-}
-
-// Keep runtime target counters outside ReplicationStats to preserve its public struct shape.
-static TARGET_QUEUE_CACHES: LazyLock<StdMutex<Vec<TargetQueueCacheSlot>>> = LazyLock::new(|| StdMutex::new(Vec::new()));
-
-fn i64_to_u64_floor_zero(value: i64) -> u64 {
-    u64::try_from(value.max(0)).unwrap_or(0)
-}
-
-fn normalized_target_arns(target_arns: &[String]) -> Vec<&str> {
-    let mut target_arns = target_arns
-        .iter()
-        .map(String::as_str)
-        .filter(|target_arn| !target_arn.is_empty())
-        .collect::<Vec<_>>();
-    target_arns.sort_unstable();
-    target_arns.dedup();
-    target_arns
-}
-
-fn target_queue_cache_snapshot(cache: &TargetQueueCache) -> Vec<RuntimeReplicationTargetBacklog> {
-    cache
-        .iter()
-        .filter_map(|((bucket, target_arn), metric)| {
-            let count = i64_to_u64_floor_zero(metric.curr.get_current_count());
-            let bytes = i64_to_u64_floor_zero(metric.curr.get_current_bytes());
-            (count > 0 || bytes > 0).then(|| RuntimeReplicationTargetBacklog {
-                bucket: bucket.clone(),
-                target_arn: target_arn.clone(),
-                count,
-                bytes,
-            })
-        })
-        .collect()
-}
-
-fn prune_stale_target_queue_caches(caches: &mut Vec<TargetQueueCacheSlot>) {
-    caches.retain(|slot| slot.owner.strong_count() > 0);
-}
-
-fn with_target_queue_caches<T>(f: impl FnOnce(&mut Vec<TargetQueueCacheSlot>) -> T) -> T {
-    match TARGET_QUEUE_CACHES.lock() {
-        Ok(mut caches) => f(&mut caches),
-        Err(poisoned) => {
-            let mut caches = poisoned.into_inner();
-            f(&mut caches)
-        }
-    }
-}
-
 impl ReplicationStats {
     pub fn new() -> Self {
         Self {
@@ -761,68 +684,6 @@ impl ReplicationStats {
         }
     }
 
-    pub(crate) fn inc_target_q(&self, bucket: &str, target_arns: &[String], size: i64) {
-        let target_arns = normalized_target_arns(target_arns);
-        if target_arns.is_empty() {
-            return;
-        }
-        with_target_queue_caches(|caches| {
-            prune_stale_target_queue_caches(caches);
-            let slot_index = match caches.iter().position(|slot| slot.belongs_to(&self.q_cache)) {
-                Some(index) => index,
-                None => {
-                    caches.push(TargetQueueCacheSlot::new(&self.q_cache));
-                    caches.len() - 1
-                }
-            };
-            let slot = &mut caches[slot_index];
-            let bucket = bucket.to_string();
-            for target_arn in target_arns {
-                let metric = match slot.metrics.entry((bucket.clone(), target_arn.to_string())) {
-                    Entry::Occupied(entry) => entry.into_mut(),
-                    Entry::Vacant(entry) => entry.insert(InQueueMetric::default()),
-                };
-                metric.curr.add_current(size, 1);
-            }
-        });
-    }
-
-    pub(crate) fn dec_target_q(&self, bucket: &str, target_arns: &[String], size: i64) {
-        let target_arns = normalized_target_arns(target_arns);
-        if target_arns.is_empty() {
-            return;
-        }
-        with_target_queue_caches(|caches| {
-            prune_stale_target_queue_caches(caches);
-            if let Some(slot) = caches.iter_mut().find(|slot| slot.belongs_to(&self.q_cache)) {
-                let bucket = bucket.to_string();
-                for target_arn in target_arns {
-                    if let Some(metric) = slot.metrics.get_mut(&(bucket.clone(), target_arn.to_string())) {
-                        metric.curr.subtract_current(size, 1);
-                    }
-                }
-                slot.metrics
-                    .retain(|_, metric| metric.curr.get_current_count() > 0 || metric.curr.get_current_bytes() > 0);
-            }
-        });
-    }
-
-    pub fn runtime_target_backlog_snapshot(&self) -> Vec<RuntimeReplicationTargetBacklog> {
-        let mut snapshot = with_target_queue_caches(|caches| {
-            caches
-                .iter()
-                .find(|slot| slot.belongs_to(&self.q_cache))
-                .map(|slot| target_queue_cache_snapshot(&slot.metrics))
-                .unwrap_or_default()
-        });
-        snapshot.sort_by(|left, right| {
-            left.bucket
-                .cmp(&right.bucket)
-                .then_with(|| left.target_arn.cmp(&right.target_arn))
-        });
-        snapshot
-    }
-
     /// Increase proxy metrics
     pub async fn inc_proxy(&self, bucket: &str, api: &str, is_err: bool) {
         let mut p_cache = self.p_cache.lock().await;
@@ -846,94 +707,6 @@ impl Default for ReplicationStats {
 mod tests {
     use super::*;
 
-    #[test]
-    fn runtime_target_backlog_snapshot_tracks_targets() {
-        let stats = ReplicationStats::new();
-        stats.inc_target_q(
-            "photos",
-            &[
-                "arn:rustfs:replication:target-b".to_string(),
-                "arn:rustfs:replication:target-a".to_string(),
-                "arn:rustfs:replication:target-a".to_string(),
-            ],
-            1024,
-        );
-
-        let snapshot = stats.runtime_target_backlog_snapshot();
-
-        assert_eq!(
-            snapshot,
-            vec![
-                RuntimeReplicationTargetBacklog {
-                    bucket: "photos".to_string(),
-                    target_arn: "arn:rustfs:replication:target-a".to_string(),
-                    count: 1,
-                    bytes: 1024,
-                },
-                RuntimeReplicationTargetBacklog {
-                    bucket: "photos".to_string(),
-                    target_arn: "arn:rustfs:replication:target-b".to_string(),
-                    count: 1,
-                    bytes: 1024,
-                },
-            ]
-        );
-    }
-
-    #[test]
-    fn runtime_target_backlog_ignores_empty_targets() {
-        let stats = ReplicationStats::new();
-        stats.inc_target_q("photos", &["".to_string()], 1024);
-
-        assert!(stats.runtime_target_backlog_snapshot().is_empty());
-    }
-
-    #[test]
-    fn runtime_target_backlog_is_scoped_to_stats_instance() {
-        let first = ReplicationStats::new();
-        let second = ReplicationStats::new();
-        first.inc_target_q("photos", &["arn:rustfs:replication:target-a".to_string()], 1024);
-
-        assert!(second.runtime_target_backlog_snapshot().is_empty());
-        assert_eq!(first.runtime_target_backlog_snapshot()[0].count, 1);
-    }
-
-    #[test]
-    fn runtime_target_backlog_decrements_with_saturation() {
-        let stats = ReplicationStats::new();
-        let target_arns = ["arn:rustfs:replication:target-a".to_string()];
-        stats.inc_target_q("photos", &target_arns, 1024);
-        stats.dec_target_q("photos", &target_arns, 2048);
-
-        assert!(stats.runtime_target_backlog_snapshot().is_empty());
-    }
-
-    #[test]
-    fn runtime_target_backlog_prunes_stale_sidecar_on_next_access() {
-        {
-            let stats = ReplicationStats::new();
-            stats.inc_target_q("photos", &["arn:rustfs:replication:target-a".to_string()], 1024);
-            assert!(
-                TARGET_QUEUE_CACHES
-                    .lock()
-                    .expect("target queue cache mutex")
-                    .iter()
-                    .any(|slot| slot.owner.strong_count() > 0)
-            );
-        }
-
-        let stats = ReplicationStats::new();
-        stats.inc_target_q("photos", &["arn:rustfs:replication:target-b".to_string()], 1024);
-
-        assert!(
-            TARGET_QUEUE_CACHES
-                .lock()
-                .expect("target queue cache mutex")
-                .iter()
-                .all(|slot| slot.owner.strong_count() > 0)
-        );
-    }
-
     #[tokio::test]
     async fn test_replication_stats_new() {
         let stats = ReplicationStats::new();

crates/ecstore/src/bucket/replication/replication_target_boundary.rs

@@ -93,12 +93,20 @@ impl ReplicationTargetStore {
         BucketTargetSys::get().get_remote_target_client(bucket, arn).await
     }
 
-    pub(crate) async fn target_is_offline(target_client: &TargetClient) -> bool {
-        BucketTargetSys::get().is_offline(&target_client.to_url()).await
+    pub(crate) async fn target_is_offline(target_client: &Arc<TargetClient>) -> bool {
+        BucketTargetSys::get().is_target_offline(target_client).await
     }
 
-    pub(crate) async fn mark_target_offline(target_client: &TargetClient) {
-        BucketTargetSys::get().mark_offline(&target_client.to_url()).await
+    pub(crate) async fn mark_target_offline(target_client: &Arc<TargetClient>) {
+        BucketTargetSys::get().mark_target_offline(target_client).await
+    }
+
+    #[cfg(test)]
+    pub(crate) async fn register_test_target(target_client: &Arc<TargetClient>) {
+        BucketTargetSys::get().arn_remotes_map.write().await.insert(
+            target_client.arn.clone(),
+            crate::bucket::bucket_target_sys::ArnTarget::with_client(target_client.clone()),
+        );
     }
 }
 

crates/obs/src/metrics/collectors/bucket_replication.rs

@@ -17,10 +17,8 @@
 use crate::metrics::report::PrometheusMetric;
 use crate::metrics::schema::bucket_replication::{
     BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD,
-    BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD,
-    BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD,
-    BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD,
-    BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD,
+    BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD,
+    BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_LAST_HR_FAILED_BYTES_MD, BUCKET_REPL_LAST_HR_FAILED_COUNT_MD,
     BUCKET_REPL_LAST_MIN_FAILED_BYTES_MD, BUCKET_REPL_LAST_MIN_FAILED_COUNT_MD, BUCKET_REPL_LATENCY_MS_MD,
     BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD, BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD,
     BUCKET_REPL_MRF_MISSED_COUNT_MD, BUCKET_REPL_MRF_PENDING_BYTES_MD, BUCKET_REPL_MRF_PENDING_COUNT_MD,
@@ -39,7 +37,6 @@ use std::borrow::Cow;
 
 const BASE_BUCKET_REPLICATION_METRICS_PER_BUCKET: usize = 25;
 const BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET: usize = 11;
-const BUCKET_REPLICATION_BACKLOG_METRICS_PER_TARGET: usize = 4;
 
 #[derive(Debug, Clone, Default)]
 pub struct BucketReplicationTargetStats {
@@ -102,16 +99,6 @@ pub(crate) struct BucketReplicationBacklogStats {
     pub(crate) mrf_missed_count: u64,
     pub(crate) mrf_flush_failures: u64,
     pub(crate) mrf_last_flush_duration_millis: u64,
-    pub(crate) target_backlogs: Vec<BucketReplicationTargetBacklogStats>,
-}
-
-#[derive(Debug, Clone, Default)]
-pub(crate) struct BucketReplicationTargetBacklogStats {
-    pub(crate) target_arn: String,
-    pub(crate) current_backlog_count: u64,
-    pub(crate) current_backlog_bytes: u64,
-    pub(crate) durable_mrf_backlog_count: u64,
-    pub(crate) durable_mrf_backlog_bytes: u64,
 }
 
 pub fn collect_bucket_replication_bandwidth_metrics(stats: &[BucketReplicationBandwidthStats]) -> Vec<PrometheusMetric> {
@@ -303,14 +290,7 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat
         return Vec::new();
     }
 
-    let metric_count = stats
-        .iter()
-        .map(|stat| {
-            BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET
-                + stat.target_backlogs.len() * BUCKET_REPLICATION_BACKLOG_METRICS_PER_TARGET
-        })
-        .sum();
-    let mut metrics = Vec::with_capacity(metric_count);
+    let mut metrics = Vec::with_capacity(stats.len() * BASE_BUCKET_REPLICATION_BACKLOG_METRICS_PER_BUCKET);
     for stat in stats {
         let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone());
 
@@ -365,42 +345,6 @@ pub(crate) fn collect_bucket_replication_backlog_metrics(stats: &[BucketReplicat
             )
             .with_label(BUCKET_L, bucket_label),
         );
-        for target in &stat.target_backlogs {
-            let bucket_label: Cow<'static, str> = Cow::Owned(stat.bucket.clone());
-            let target_label: Cow<'static, str> = Cow::Owned(target.target_arn.clone());
-            metrics.push(
-                PrometheusMetric::from_descriptor(
-                    &BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD,
-                    target.current_backlog_count as f64,
-                )
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_label.clone()),
-            );
-            metrics.push(
-                PrometheusMetric::from_descriptor(
-                    &BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD,
-                    target.current_backlog_bytes as f64,
-                )
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_label.clone()),
-            );
-            metrics.push(
-                PrometheusMetric::from_descriptor(
-                    &BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD,
-                    target.durable_mrf_backlog_count as f64,
-                )
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_label.clone()),
-            );
-            metrics.push(
-                PrometheusMetric::from_descriptor(
-                    &BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD,
-                    target.durable_mrf_backlog_bytes as f64,
-                )
-                .with_label(BUCKET_L, bucket_label)
-                .with_label(TARGET_ARN_L, target_label),
-            );
-        }
     }
 
     metrics
@@ -525,17 +469,10 @@ mod tests {
             mrf_missed_count: 4,
             mrf_flush_failures: 5,
             mrf_last_flush_duration_millis: 6,
-            target_backlogs: vec![BucketReplicationTargetBacklogStats {
-                target_arn: "arn:rustfs:replication:target-a".to_string(),
-                current_backlog_count: 3,
-                current_backlog_bytes: 4096,
-                durable_mrf_backlog_count: 2,
-                durable_mrf_backlog_bytes: 2048,
-            }],
         }];
 
         let metrics = collect_bucket_replication_backlog_metrics(&stats);
-        assert_eq!(metrics.len(), 15);
+        assert_eq!(metrics.len(), 11);
 
         let backlog_count_name = BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
@@ -572,50 +509,6 @@ mod tests {
                 && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
         }));
 
-        let target_count_name = BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name();
-        assert!(metrics.iter().any(|metric| {
-            metric.name == target_count_name
-                && metric.value == 3.0
-                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
-                && metric
-                    .labels
-                    .iter()
-                    .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a")
-        }));
-
-        let target_bytes_name = BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name();
-        assert!(metrics.iter().any(|metric| {
-            metric.name == target_bytes_name
-                && metric.value == 4096.0
-                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
-                && metric
-                    .labels
-                    .iter()
-                    .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a")
-        }));
-
-        let durable_target_count_name = BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name();
-        assert!(metrics.iter().any(|metric| {
-            metric.name == durable_target_count_name
-                && metric.value == 2.0
-                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
-                && metric
-                    .labels
-                    .iter()
-                    .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a")
-        }));
-
-        let durable_target_bytes_name = BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD.get_full_metric_name();
-        assert!(metrics.iter().any(|metric| {
-            metric.name == durable_target_bytes_name
-                && metric.value == 2048.0
-                && metric.labels.iter().any(|(key, value)| *key == BUCKET_L && value == "b1")
-                && metric
-                    .labels
-                    .iter()
-                    .any(|(key, value)| *key == TARGET_ARN_L && value == "arn:rustfs:replication:target-a")
-        }));
-
         let pending_count_name = BUCKET_REPL_MRF_PENDING_COUNT_MD.get_full_metric_name();
         assert!(metrics.iter().any(|metric| {
             metric.name == pending_count_name
@@ -681,7 +574,6 @@ mod tests {
             mrf_missed_count: 4,
             mrf_flush_failures: 5,
             mrf_last_flush_duration_millis: 6,
-            target_backlogs: Vec::new(),
         }];
 
         let metrics = collect_bucket_replication_backlog_metrics(&stats);

crates/obs/src/metrics/collectors/mod.rs

@@ -42,9 +42,7 @@ pub mod system_process;
 
 pub use audit::{AuditTargetStats, collect_audit_metrics};
 pub use bucket::{BucketStats, collect_bucket_metrics};
-pub(crate) use bucket_replication::{
-    BucketReplicationBacklogStats, BucketReplicationTargetBacklogStats, collect_bucket_replication_backlog_metrics,
-};
+pub(crate) use bucket_replication::{BucketReplicationBacklogStats, collect_bucket_replication_backlog_metrics};
 pub use bucket_replication::{
     BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats,
     collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics,

crates/obs/src/metrics/scheduler.rs

@@ -80,10 +80,8 @@ use crate::metrics::runtime_sources::bucket_monitor_available;
 use crate::metrics::schema::audit::{AUDIT_FAILED_MESSAGES_MD, AUDIT_TARGET_QUEUE_LENGTH_MD, AUDIT_TOTAL_MESSAGES_MD};
 use crate::metrics::schema::bucket_replication::{
     BUCKET_L, BUCKET_REPL_BANDWIDTH_CURRENT_MD, BUCKET_REPL_BANDWIDTH_LIMIT_MD, BUCKET_REPL_CURRENT_BACKLOG_BYTES_MD,
-    BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD,
-    BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD,
-    BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD,
-    BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD,
+    BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD, BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD, BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD,
+    BUCKET_REPL_DURABLE_MRF_BACKLOG_COUNT_MD, BUCKET_REPL_MRF_DROPPED_COUNT_MD, BUCKET_REPL_MRF_FLUSH_FAILURES_MD,
     BUCKET_REPL_MRF_LAST_FLUSH_DURATION_MILLIS_MD, BUCKET_REPL_MRF_MISSED_COUNT_MD, BUCKET_REPL_MRF_PENDING_BYTES_MD,
     BUCKET_REPL_MRF_PENDING_COUNT_MD, TARGET_ARN_L,
 };
@@ -635,17 +633,6 @@ fn repl_backlog_live_keys(stats: &[BucketReplicationBacklogStats]) -> HashSet<Bu
     stats.iter().map(|s| s.bucket.clone()).collect()
 }
 
-fn repl_backlog_target_live_keys(stats: &[BucketReplicationBacklogStats]) -> HashSet<ReplBwKey> {
-    stats
-        .iter()
-        .flat_map(|stat| {
-            stat.target_backlogs
-                .iter()
-                .map(|target| (stat.bucket.clone(), target.target_arn.clone()))
-        })
-        .collect()
-}
-
 fn update_series_zero_tombstones<T: Clone + Eq + std::hash::Hash>(
     has_seen_valid_snapshot: &mut bool,
     prev_live_keys: &mut HashSet<T>,
@@ -1073,40 +1060,6 @@ fn collect_repl_backlog_zero_tombstone_metrics(zero_tombstones: &HashMap<BucketK
     collect_bucket_replication_backlog_metrics(&stats)
 }
 
-fn collect_repl_backlog_target_zero_tombstone_metrics(zero_tombstones: &HashMap<ReplBwKey, u8>) -> Vec<PrometheusMetric> {
-    if zero_tombstones.is_empty() {
-        return Vec::new();
-    }
-
-    let mut zero_metrics = Vec::with_capacity(zero_tombstones.len() * 4);
-    for (bucket, target_arn) in zero_tombstones.keys() {
-        let bucket_label: Cow<'static, str> = Cow::Owned(bucket.clone());
-        let target_arn_label: Cow<'static, str> = Cow::Owned(target_arn.clone());
-        zero_metrics.push(
-            PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD, 0.0)
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_arn_label.clone()),
-        );
-        zero_metrics.push(
-            PrometheusMetric::from_descriptor(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD, 0.0)
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_arn_label.clone()),
-        );
-        zero_metrics.push(
-            PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD, 0.0)
-                .with_label(BUCKET_L, bucket_label.clone())
-                .with_label(TARGET_ARN_L, target_arn_label.clone()),
-        );
-        zero_metrics.push(
-            PrometheusMetric::from_descriptor(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD, 0.0)
-                .with_label(BUCKET_L, bucket_label)
-                .with_label(TARGET_ARN_L, target_arn_label),
-        );
-    }
-
-    zero_metrics
-}
-
 fn retire_repl_bw_metric_series(bucket: &str, target_arn: &str) -> usize {
     let labels = [
         (BUCKET_L, Cow::Owned(bucket.to_string())),
@@ -1136,17 +1089,6 @@ fn retire_repl_backlog_metric_series(bucket: &str) -> usize {
     .sum()
 }
 
-fn retire_repl_backlog_target_metric_series(bucket: &str, target_arn: &str) -> usize {
-    let labels = [
-        (BUCKET_L, Cow::Owned(bucket.to_string())),
-        (TARGET_ARN_L, Cow::Owned(target_arn.to_string())),
-    ];
-    retire_metric_series(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), &labels)
-        + retire_metric_series(&BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(), &labels)
-        + retire_metric_series(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(), &labels)
-        + retire_metric_series(&BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(), &labels)
-}
-
 fn expire_repl_bw_zero_tombstones(monitor_available: bool, zero_tombstones: &mut HashMap<ReplBwKey, u8>) -> Vec<ReplBwKey> {
     if !monitor_available {
         return Vec::new();
@@ -1163,17 +1105,6 @@ fn expire_repl_backlog_zero_tombstones(monitor_available: bool, zero_tombstones:
     expire_series_zero_tombstones(zero_tombstones)
 }
 
-fn expire_repl_backlog_target_zero_tombstones(
-    target_metrics_available: bool,
-    zero_tombstones: &mut HashMap<ReplBwKey, u8>,
-) -> Vec<ReplBwKey> {
-    if !target_metrics_available {
-        return Vec::new();
-    }
-
-    expire_series_zero_tombstones(zero_tombstones)
-}
-
 /// Initialize all metrics collectors.
 ///
 /// This function spawns background tasks that periodically collect metrics
@@ -1468,9 +1399,6 @@ pub fn init_metrics_runtime(token: CancellationToken) {
         let mut prev_backlog_live_keys: HashSet<BucketKey> = HashSet::new();
         let mut backlog_zero_tombstones: HashMap<BucketKey, u8> = HashMap::new();
         let mut has_seen_valid_backlog_snapshot = false;
-        let mut prev_backlog_target_live_keys: HashSet<ReplBwKey> = HashSet::new();
-        let mut backlog_target_zero_tombstones: HashMap<ReplBwKey, u8> = HashMap::new();
-        let mut has_seen_valid_backlog_target_snapshot = false;
         loop {
             tokio::select! {
                 _ = interval.tick() => {
@@ -1501,9 +1429,6 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             metrics.extend(collect_repl_bw_zero_tombstone_metrics(&zero_tombstones));
 
                             let (bucket_replication, bucket_replication_backlog) = collect_bucket_replication_stats_bundle().await;
-                            let durable_mrf_available = bucket_replication_backlog.iter().any(|stat| stat.durable_mrf_available);
-                            let backlog_target_metrics_available =
-                                durable_mrf_available || monitor_available || !bucket_replication_backlog.is_empty();
                             update_repl_backlog_zero_tombstones(
                                 monitor_available,
                                 &mut has_seen_valid_backlog_snapshot,
@@ -1512,19 +1437,9 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                                 repl_backlog_live_keys(&bucket_replication_backlog),
                                 repl_bw_zero_tombstone_cycles,
                             );
-                            if backlog_target_metrics_available {
-                                update_series_zero_tombstones(
-                                    &mut has_seen_valid_backlog_target_snapshot,
-                                    &mut prev_backlog_target_live_keys,
-                                    &mut backlog_target_zero_tombstones,
-                                    repl_backlog_target_live_keys(&bucket_replication_backlog),
-                                    repl_bw_zero_tombstone_cycles,
-                                );
-                            }
                             metrics.extend(collect_bucket_replication_metrics(&bucket_replication));
                             metrics.extend(collect_bucket_replication_backlog_metrics(&bucket_replication_backlog));
                             metrics.extend(collect_repl_backlog_zero_tombstone_metrics(&backlog_zero_tombstones));
-                            metrics.extend(collect_repl_backlog_target_zero_tombstone_metrics(&backlog_target_zero_tombstones));
                             let replication = collect_replication_stats().await;
                             metrics.extend(collect_replication_metrics(&replication));
                             report_metrics(&metrics);
@@ -1536,12 +1451,6 @@ pub fn init_metrics_runtime(token: CancellationToken) {
                             for bucket in expire_repl_backlog_zero_tombstones(monitor_available, &mut backlog_zero_tombstones) {
                                 let _ = retire_repl_backlog_metric_series(&bucket);
                             }
-                            for (bucket, target_arn) in expire_repl_backlog_target_zero_tombstones(
-                                backlog_target_metrics_available,
-                                &mut backlog_target_zero_tombstones,
-                            ) {
-                                let _ = retire_repl_backlog_target_metric_series(&bucket, &target_arn);
-                            }
                         },
                     ).await;
                 }
@@ -2350,69 +2259,6 @@ mod tests {
         assert_eq!(prev_live_keys, bucket_keys(&["photos"]));
     }
 
-    #[test]
-    fn repl_backlog_target_tombstones_zero_removed_targets_then_expire() {
-        let mut has_seen_valid_snapshot = false;
-        let mut prev_live_keys = HashSet::new();
-        let mut zero_tombstones = HashMap::new();
-        let key = repl_bw_key("photos", "arn:rustfs:replication:target-a");
-
-        update_series_zero_tombstones(
-            &mut has_seen_valid_snapshot,
-            &mut prev_live_keys,
-            &mut zero_tombstones,
-            repl_bw_keys(&[("photos", "arn:rustfs:replication:target-a")]),
-            2,
-        );
-        assert!(has_seen_valid_snapshot);
-        assert_eq!(prev_live_keys, repl_bw_keys(&[("photos", "arn:rustfs:replication:target-a")]));
-        assert!(zero_tombstones.is_empty());
-
-        update_series_zero_tombstones(&mut has_seen_valid_snapshot, &mut prev_live_keys, &mut zero_tombstones, HashSet::new(), 2);
-        assert_eq!(zero_tombstones.get(&key), Some(&2));
-
-        let metrics = collect_repl_backlog_target_zero_tombstone_metrics(&zero_tombstones);
-        assert_eq!(metrics.len(), 4);
-        let expected_names = HashSet::from([
-            BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(),
-            BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(),
-            BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD.get_full_metric_name(),
-            BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD.get_full_metric_name(),
-        ]);
-        let mut actual_names = HashSet::new();
-        for metric in metrics {
-            actual_names.insert(metric.name.to_string());
-            assert_eq!(metric.value, 0.0);
-            let labels = metric
-                .labels
-                .into_iter()
-                .map(|(key, value)| (key, value.to_string()))
-                .collect::<HashMap<_, _>>();
-            assert_eq!(labels.get(BUCKET_L).map(String::as_str), Some("photos"));
-            assert_eq!(labels.get(TARGET_ARN_L).map(String::as_str), Some("arn:rustfs:replication:target-a"));
-        }
-        assert_eq!(actual_names, expected_names);
-
-        let expired = expire_repl_backlog_target_zero_tombstones(true, &mut zero_tombstones);
-        assert!(expired.is_empty());
-        assert_eq!(zero_tombstones.get(&key), Some(&1));
-
-        let expired = expire_repl_backlog_target_zero_tombstones(true, &mut zero_tombstones);
-        assert_eq!(expired, vec![key]);
-        assert!(zero_tombstones.is_empty());
-    }
-
-    #[test]
-    fn repl_backlog_target_tombstones_do_not_advance_when_target_metrics_unavailable() {
-        let key = repl_bw_key("videos", "arn:rustfs:replication:target-b");
-        let mut zero_tombstones = HashMap::from([(key.clone(), 1)]);
-
-        let expired = expire_repl_backlog_target_zero_tombstones(false, &mut zero_tombstones);
-
-        assert!(expired.is_empty());
-        assert_eq!(zero_tombstones.get(&key), Some(&1));
-    }
-
     #[test]
     fn bucket_tombstones_zero_removed_buckets_then_expire() {
         let mut has_seen_valid_snapshot = false;

crates/obs/src/metrics/schema/bucket_replication.rs

@@ -35,13 +35,9 @@ const RESYNC_CANCELED_TOTAL: &str = "resync_canceled_total";
 const RESYNC_DURATION_MS_TOTAL: &str = "resync_duration_ms_total";
 const CURRENT_BACKLOG_COUNT: &str = "current_backlog_count";
 const CURRENT_BACKLOG_BYTES: &str = "current_backlog_bytes";
-const CURRENT_TARGET_BACKLOG_COUNT: &str = "current_target_backlog_count";
-const CURRENT_TARGET_BACKLOG_BYTES: &str = "current_target_backlog_bytes";
 const DURABLE_MRF_AVAILABLE: &str = "durable_mrf_available";
 const DURABLE_MRF_BACKLOG_COUNT: &str = "durable_mrf_backlog_count";
 const DURABLE_MRF_BACKLOG_BYTES: &str = "durable_mrf_backlog_bytes";
-const DURABLE_MRF_TARGET_BACKLOG_COUNT: &str = "durable_mrf_target_backlog_count";
-const DURABLE_MRF_TARGET_BACKLOG_BYTES: &str = "durable_mrf_target_backlog_bytes";
 const MRF_PENDING_COUNT: &str = "mrf_pending_count";
 const MRF_PENDING_BYTES: &str = "mrf_pending_bytes";
 const MRF_DROPPED_COUNT: &str = "mrf_dropped_count";
@@ -112,24 +108,6 @@ pub static BUCKET_REPL_CURRENT_BACKLOG_COUNT_MD: LazyLock<MetricDescriptor> = La
     )
 });
 
-pub static BUCKET_REPL_CURRENT_TARGET_BACKLOG_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
-    new_gauge_md(
-        MetricName::from(CURRENT_TARGET_BACKLOG_COUNT),
-        "Current number of target-scoped operations admitted to in-memory replication worker queues for a bucket and target ARN",
-        &[BUCKET_L, TARGET_ARN_L],
-        subsystems::BUCKET_REPLICATION,
-    )
-});
-
-pub static BUCKET_REPL_CURRENT_TARGET_BACKLOG_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
-    new_gauge_md(
-        MetricName::from(CURRENT_TARGET_BACKLOG_BYTES),
-        "Current bytes in target-scoped operations admitted to in-memory replication worker queues for a bucket and target ARN",
-        &[BUCKET_L, TARGET_ARN_L],
-        subsystems::BUCKET_REPLICATION,
-    )
-});
-
 pub static BUCKET_REPL_DURABLE_MRF_AVAILABLE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::from(DURABLE_MRF_AVAILABLE),
@@ -157,24 +135,6 @@ pub static BUCKET_REPL_DURABLE_MRF_BACKLOG_BYTES_MD: LazyLock<MetricDescriptor>
     )
 });
 
-pub static BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
-    new_gauge_md(
-        MetricName::from(DURABLE_MRF_TARGET_BACKLOG_COUNT),
-        "Current number of target-scoped operations in the durable MRF backlog file for a bucket and target ARN on this node",
-        &[BUCKET_L, TARGET_ARN_L],
-        subsystems::BUCKET_REPLICATION,
-    )
-});
-
-pub static BUCKET_REPL_DURABLE_MRF_TARGET_BACKLOG_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
-    new_gauge_md(
-        MetricName::from(DURABLE_MRF_TARGET_BACKLOG_BYTES),
-        "Current bytes in target-scoped operations in the durable MRF backlog file for a bucket and target ARN on this node",
-        &[BUCKET_L, TARGET_ARN_L],
-        subsystems::BUCKET_REPLICATION,
-    )
-});
-
 pub static BUCKET_REPL_MRF_PENDING_COUNT_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::from(MRF_PENDING_COUNT),

crates/obs/src/metrics/stats_collector.rs

@@ -21,11 +21,11 @@
 //! and convert them to the Stats structs used by collectors.
 
 use crate::metrics::collectors::{
-    BucketReplicationBacklogStats, BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetBacklogStats,
-    BucketReplicationTargetStats, BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats,
-    ClusterUsageStats, CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, ErasureSetStats,
-    HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats,
-    ResourceStats, ScannerStats,
+    BucketReplicationBacklogStats, BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats,
+    BucketStats, BucketUsageStats, ClusterConfigStats, ClusterHealthStats, ClusterStats, ClusterUsageStats,
+    CompressionClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, ErasureSetStats, HostNetworkStats,
+    IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, ProcessStatusType, ReplicationStats, ResourceStats,
+    ScannerStats,
 };
 use crate::metrics::runtime_sources::{ObsIlmRuntimeSnapshot, bucket_monitor_handle, iam_metrics_snapshot, ilm_runtime_snapshot};
 use crate::metrics::{
@@ -212,17 +212,6 @@ async fn obs_bucket_replication_stats_bundle() -> (Vec<BucketReplicationStats>,
             mrf_missed_count: stats.mrf_missed_count,
             mrf_flush_failures: stats.mrf_flush_failures,
             mrf_last_flush_duration_millis: stats.mrf_last_flush_duration_millis,
-            target_backlogs: stats
-                .target_backlogs
-                .iter()
-                .map(|target| BucketReplicationTargetBacklogStats {
-                    target_arn: target.target_arn.clone(),
-                    current_backlog_count: target.current_backlog_count,
-                    current_backlog_bytes: target.current_backlog_bytes,
-                    durable_mrf_backlog_count: target.durable_mrf_backlog_count,
-                    durable_mrf_backlog_bytes: target.durable_mrf_backlog_bytes,
-                })
-                .collect(),
         });
         detail_stats.push(bucket_replication_detail_from_snapshot(stats));
     }

crates/obs/src/metrics/storage_api.rs

@@ -18,8 +18,7 @@ use std::time::Duration;
 pub(crate) use rustfs_ecstore::api::bucket::bandwidth::monitor::Monitor as ObsBucketBandwidthMonitor;
 pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::get_quota_config as obs_get_quota_config;
 use rustfs_ecstore::api::bucket::replication::{
-    DurableMrfBucketBacklog, DurableMrfTargetBacklog, MrfBucketBacklogObservability, RuntimeReplicationTargetBacklog,
-    durable_mrf_backlog_summary_snapshot, durable_mrf_target_backlog_snapshot, get_global_replication_stats,
+    DurableMrfBucketBacklog, MrfBucketBacklogObservability, durable_mrf_backlog_summary_snapshot, get_global_replication_stats,
     mrf_backlog_observability_snapshot,
 };
 pub(crate) use rustfs_ecstore::api::capacity::{
@@ -45,15 +44,6 @@ pub(crate) struct ObsBucketReplicationTargetStatsSnapshot {
     pub(crate) latency_ms: f64,
 }
 
-#[derive(Debug, Clone, PartialEq)]
-pub(crate) struct ObsBucketReplicationTargetBacklogSnapshot {
-    pub(crate) target_arn: String,
-    pub(crate) current_backlog_count: u64,
-    pub(crate) current_backlog_bytes: u64,
-    pub(crate) durable_mrf_backlog_count: u64,
-    pub(crate) durable_mrf_backlog_bytes: u64,
-}
-
 #[derive(Debug, Clone, PartialEq)]
 pub(crate) struct ObsBucketReplicationStatsSnapshot {
     pub(crate) bucket: String,
@@ -94,7 +84,6 @@ pub(crate) struct ObsBucketReplicationStatsSnapshot {
     pub(crate) mrf_flush_failures: u64,
     pub(crate) mrf_last_flush_duration_millis: u64,
     pub(crate) targets: Vec<ObsBucketReplicationTargetStatsSnapshot>,
-    pub(crate) target_backlogs: Vec<ObsBucketReplicationTargetBacklogSnapshot>,
 }
 
 #[derive(Debug, Clone, Default, PartialEq)]
@@ -133,15 +122,6 @@ struct ObsBucketReplicationProxySnapshot {
     proxied_delete_tagging_requests_failures: u64,
 }
 
-#[derive(Debug, Clone, Default, PartialEq)]
-struct ObsBucketReplicationBacklogSnapshot {
-    durable_mrf_available: bool,
-    durable_bucket: DurableMrfBucketBacklog,
-    runtime_targets: Vec<RuntimeReplicationTargetBacklog>,
-    durable_targets: Vec<DurableMrfTargetBacklog>,
-    mrf_observability: MrfBucketBacklogObservability,
-}
-
 #[derive(Debug, Clone, Default, PartialEq)]
 pub(crate) struct ObsReplicationSiteStatsSnapshot {
     pub(crate) average_active_workers: f64,
@@ -177,40 +157,10 @@ fn bucket_replication_stats_snapshot_from_parts(
     bucket: String,
     runtime: ObsBucketReplicationRuntimeSnapshot,
     proxy: ObsBucketReplicationProxySnapshot,
-    backlog: ObsBucketReplicationBacklogSnapshot,
+    durable_mrf_available: bool,
+    durable_bucket: DurableMrfBucketBacklog,
+    mrf_observability: MrfBucketBacklogObservability,
 ) -> ObsBucketReplicationStatsSnapshot {
-    let mut target_backlogs = HashMap::with_capacity(backlog.runtime_targets.len().saturating_add(backlog.durable_targets.len()));
-    for target in backlog.runtime_targets {
-        let entry =
-            target_backlogs
-                .entry(target.target_arn.clone())
-                .or_insert_with(|| ObsBucketReplicationTargetBacklogSnapshot {
-                    target_arn: target.target_arn,
-                    current_backlog_count: 0,
-                    current_backlog_bytes: 0,
-                    durable_mrf_backlog_count: 0,
-                    durable_mrf_backlog_bytes: 0,
-                });
-        entry.current_backlog_count = target.count;
-        entry.current_backlog_bytes = target.bytes;
-    }
-    for target in backlog.durable_targets {
-        let entry =
-            target_backlogs
-                .entry(target.target_arn.clone())
-                .or_insert_with(|| ObsBucketReplicationTargetBacklogSnapshot {
-                    target_arn: target.target_arn,
-                    current_backlog_count: 0,
-                    current_backlog_bytes: 0,
-                    durable_mrf_backlog_count: 0,
-                    durable_mrf_backlog_bytes: 0,
-                });
-        entry.durable_mrf_backlog_count = target.count;
-        entry.durable_mrf_backlog_bytes = target.bytes;
-    }
-    let mut target_backlogs = target_backlogs.into_values().collect::<Vec<_>>();
-    target_backlogs.sort_by(|left, right| left.target_arn.cmp(&right.target_arn));
-
     ObsBucketReplicationStatsSnapshot {
         bucket,
         total_failed_bytes: runtime.total_failed_bytes,
@@ -240,17 +190,16 @@ fn bucket_replication_stats_snapshot_from_parts(
         resync_duration_ms: runtime.resync_duration_ms,
         current_backlog_count: runtime.current_backlog_count,
         current_backlog_bytes: runtime.current_backlog_bytes,
-        durable_mrf_available: backlog.durable_mrf_available,
-        durable_mrf_backlog_count: backlog.durable_bucket.count,
-        durable_mrf_backlog_bytes: backlog.durable_bucket.bytes,
-        mrf_pending_count: backlog.mrf_observability.pending_count,
-        mrf_pending_bytes: backlog.mrf_observability.pending_bytes,
-        mrf_dropped_count: backlog.mrf_observability.dropped_count,
-        mrf_missed_count: backlog.mrf_observability.missed_count,
-        mrf_flush_failures: backlog.mrf_observability.flush_failure_count,
-        mrf_last_flush_duration_millis: backlog.mrf_observability.last_flush_duration_millis,
+        durable_mrf_available,
+        durable_mrf_backlog_count: durable_bucket.count,
+        durable_mrf_backlog_bytes: durable_bucket.bytes,
+        mrf_pending_count: mrf_observability.pending_count,
+        mrf_pending_bytes: mrf_observability.pending_bytes,
+        mrf_dropped_count: mrf_observability.dropped_count,
+        mrf_missed_count: mrf_observability.missed_count,
+        mrf_flush_failures: mrf_observability.flush_failure_count,
+        mrf_last_flush_duration_millis: mrf_observability.last_flush_duration_millis,
         targets: runtime.targets,
-        target_backlogs,
     }
 }
 
@@ -273,27 +222,6 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
         .into_iter()
         .map(|bucket| (bucket.bucket.clone(), bucket))
         .collect::<HashMap<String, DurableMrfBucketBacklog>>();
-    let mut durable_targets_by_bucket: HashMap<String, Vec<DurableMrfTargetBacklog>> = HashMap::new();
-    let durable_mrf_targets = if replication_storage_available && durable_mrf_available {
-        durable_mrf_target_backlog_snapshot()
-    } else {
-        Vec::new()
-    };
-    for target in durable_mrf_targets {
-        durable_targets_by_bucket
-            .entry(target.bucket.clone())
-            .or_default()
-            .push(target);
-    }
-    let mut runtime_targets_by_bucket: HashMap<String, Vec<RuntimeReplicationTargetBacklog>> = HashMap::new();
-    if let Some(stats) = &stats {
-        for target in stats.runtime_target_backlog_snapshot() {
-            runtime_targets_by_bucket
-                .entry(target.bucket.clone())
-                .or_default()
-                .push(target);
-        }
-    }
     let mrf_observability = if replication_storage_available {
         mrf_backlog_observability_snapshot()
     } else {
@@ -308,8 +236,7 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
         all_bucket_stats
             .len()
             .saturating_add(durable_buckets.len())
-            .saturating_add(mrf_observability_buckets.len())
-            .saturating_add(runtime_targets_by_bucket.len()),
+            .saturating_add(mrf_observability_buckets.len()),
     );
     bucket_names.extend(all_bucket_stats.keys().cloned());
     bucket_names.extend(
@@ -324,16 +251,6 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
             .filter(|bucket| !all_bucket_stats.contains_key(*bucket) && !durable_buckets.contains_key(*bucket))
             .cloned(),
     );
-    bucket_names.extend(
-        runtime_targets_by_bucket
-            .keys()
-            .filter(|bucket| {
-                !all_bucket_stats.contains_key(*bucket)
-                    && !durable_buckets.contains_key(*bucket)
-                    && !mrf_observability_buckets.contains_key(*bucket)
-            })
-            .cloned(),
-    );
     let mut buckets = Vec::with_capacity(bucket_names.len());
 
     for bucket in bucket_names {
@@ -410,20 +327,14 @@ pub(crate) async fn obs_bucket_replication_stats_snapshot() -> Vec<ObsBucketRepl
             runtime.current_backlog_bytes = i64_to_u64_floor_zero(bucket_stats.q_stat.curr.bytes);
         }
         let durable_bucket = durable_buckets.get(&bucket).cloned().unwrap_or_default();
-        let runtime_targets = runtime_targets_by_bucket.remove(&bucket).unwrap_or_default();
-        let durable_targets = durable_targets_by_bucket.remove(&bucket).unwrap_or_default();
         let mrf_observability = mrf_observability_buckets.get(&bucket).cloned().unwrap_or_default();
         buckets.push(bucket_replication_stats_snapshot_from_parts(
             bucket,
             runtime,
             proxy,
-            ObsBucketReplicationBacklogSnapshot {
-                durable_mrf_available,
-                durable_bucket,
-                runtime_targets,
-                durable_targets,
-                mrf_observability,
-            },
+            durable_mrf_available,
+            durable_bucket,
+            mrf_observability,
         ));
     }
 
@@ -515,34 +426,20 @@ mod tests {
                 proxied_get_requests_failures: 1,
                 ..Default::default()
             },
-            ObsBucketReplicationBacklogSnapshot {
-                durable_mrf_available: true,
-                durable_bucket: DurableMrfBucketBacklog {
-                    bucket: "runtime-bucket".to_string(),
-                    count: 5,
-                    bytes: 8192,
-                },
-                runtime_targets: vec![RuntimeReplicationTargetBacklog {
-                    bucket: "runtime-bucket".to_string(),
-                    target_arn: "arn:rustfs:replication:target-a".to_string(),
-                    count: 3,
-                    bytes: 4096,
-                }],
-                durable_targets: vec![DurableMrfTargetBacklog {
-                    bucket: "runtime-bucket".to_string(),
-                    target_arn: "arn:rustfs:replication:target-a".to_string(),
-                    count: 2,
-                    bytes: 4096,
-                }],
-                mrf_observability: MrfBucketBacklogObservability {
-                    bucket: "runtime-bucket".to_string(),
-                    pending_count: 1,
-                    pending_bytes: 512,
-                    dropped_count: 2,
-                    missed_count: 3,
-                    flush_failure_count: 4,
-                    last_flush_duration_millis: 5,
-                },
+            true,
+            DurableMrfBucketBacklog {
+                bucket: "runtime-bucket".to_string(),
+                count: 5,
+                bytes: 8192,
+            },
+            MrfBucketBacklogObservability {
+                bucket: "runtime-bucket".to_string(),
+                pending_count: 1,
+                pending_bytes: 512,
+                dropped_count: 2,
+                missed_count: 3,
+                flush_failure_count: 4,
+                last_flush_duration_millis: 5,
             },
         );
 
@@ -552,12 +449,6 @@ mod tests {
         assert!(snapshot.durable_mrf_available);
         assert_eq!(snapshot.durable_mrf_backlog_count, 5);
         assert_eq!(snapshot.durable_mrf_backlog_bytes, 8192);
-        assert_eq!(snapshot.target_backlogs.len(), 1);
-        assert_eq!(snapshot.target_backlogs[0].target_arn, "arn:rustfs:replication:target-a");
-        assert_eq!(snapshot.target_backlogs[0].current_backlog_count, 3);
-        assert_eq!(snapshot.target_backlogs[0].current_backlog_bytes, 4096);
-        assert_eq!(snapshot.target_backlogs[0].durable_mrf_backlog_count, 2);
-        assert_eq!(snapshot.target_backlogs[0].durable_mrf_backlog_bytes, 4096);
         assert_eq!(snapshot.mrf_pending_count, 1);
         assert_eq!(snapshot.mrf_pending_bytes, 512);
         assert_eq!(snapshot.mrf_dropped_count, 2);
@@ -575,15 +466,13 @@ mod tests {
             "durable-only".to_string(),
             ObsBucketReplicationRuntimeSnapshot::default(),
             ObsBucketReplicationProxySnapshot::default(),
-            ObsBucketReplicationBacklogSnapshot {
-                durable_mrf_available: true,
-                durable_bucket: DurableMrfBucketBacklog {
-                    bucket: "durable-only".to_string(),
-                    count: 11,
-                    bytes: 2048,
-                },
-                ..Default::default()
+            true,
+            DurableMrfBucketBacklog {
+                bucket: "durable-only".to_string(),
+                count: 11,
+                bytes: 2048,
             },
+            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.bucket, "durable-only");
@@ -599,18 +488,16 @@ mod tests {
             "mrf-observability-only".to_string(),
             ObsBucketReplicationRuntimeSnapshot::default(),
             ObsBucketReplicationProxySnapshot::default(),
-            ObsBucketReplicationBacklogSnapshot {
-                durable_mrf_available: true,
-                mrf_observability: MrfBucketBacklogObservability {
-                    bucket: "mrf-observability-only".to_string(),
-                    pending_count: 13,
-                    pending_bytes: 4096,
-                    dropped_count: 1,
-                    missed_count: 2,
-                    flush_failure_count: 3,
-                    last_flush_duration_millis: 4,
-                },
-                ..Default::default()
+            true,
+            DurableMrfBucketBacklog::default(),
+            MrfBucketBacklogObservability {
+                bucket: "mrf-observability-only".to_string(),
+                pending_count: 13,
+                pending_bytes: 4096,
+                dropped_count: 1,
+                missed_count: 2,
+                flush_failure_count: 3,
+                last_flush_duration_millis: 4,
             },
         );
 
@@ -638,7 +525,9 @@ mod tests {
                 ..Default::default()
             },
             ObsBucketReplicationProxySnapshot::default(),
-            ObsBucketReplicationBacklogSnapshot::default(),
+            false,
+            DurableMrfBucketBacklog::default(),
+            MrfBucketBacklogObservability::default(),
         );
 
         assert_eq!(snapshot.current_backlog_count, 1);

crates/replication/src/delete.rs

@@ -49,11 +49,6 @@ impl ReplicationWorkerOperation for DeletedObjectReplicationInfo {
                 .delete_object
                 .delete_marker_mtime
                 .and_then(|t| i64::try_from(t.unix_timestamp_nanos()).ok()),
-            target_arns: if self.target_arn.is_empty() {
-                Vec::new()
-            } else {
-                vec![self.target_arn.clone()]
-            },
         }
     }
 
@@ -116,7 +111,6 @@ mod tests {
                 delete_marker_mtime: Some(mtime),
                 ..Default::default()
             },
-            target_arn: "arn:target-a".to_string(),
             ..Default::default()
         };
 
@@ -135,7 +129,6 @@ mod tests {
             Some(mtime.unix_timestamp_nanos() as i64),
             "delete-marker mtime must be persisted in the MRF entry"
         );
-        assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]);
         assert_eq!(info.get_object(), "object");
     }
 
@@ -155,7 +148,6 @@ mod tests {
         };
 
         assert_eq!(info.to_mrf_entry().delete_marker_mtime, None);
-        assert!(info.to_mrf_entry().target_arns.is_empty());
     }
 
     #[test]

crates/replication/src/filemeta.rs

@@ -593,9 +593,6 @@ pub struct MrfReplicateEntry {
     // to preserve pre-existing behaviour (backlog#867).
     #[serde(rename = "deleteMarkerMtime", skip_serializing_if = "Option::is_none", default)]
     pub delete_marker_mtime: Option<i64>,
-
-    #[serde(rename = "targetARNs", skip_serializing_if = "Vec::is_empty", default)]
-    pub target_arns: Vec<String>,
 }
 
 fn retry_count_to_mrf(retry_count: u32) -> i32 {
@@ -675,18 +672,6 @@ impl ReplicateDecision {
         }
         if result.is_empty() { None } else { Some(result) }
     }
-
-    pub fn replicate_target_arns(&self) -> Vec<String> {
-        let mut arns = self
-            .targets_map
-            .values()
-            .filter(|target| target.replicate && !target.arn.is_empty())
-            .map(|target| target.arn.clone())
-            .collect::<Vec<_>>();
-        arns.sort();
-        arns.dedup();
-        arns
-    }
 }
 
 impl fmt::Display for ReplicateDecision {
@@ -814,7 +799,6 @@ impl ReplicationWorkerOperation for ReplicateObjectInfo {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: self.dsc.replicate_target_arns(),
         }
     }
 
@@ -869,7 +853,6 @@ impl ReplicateObjectInfo {
             delete_marker_version_id: None,
             delete_marker: false,
             delete_marker_mtime: None,
-            target_arns: self.dsc.replicate_target_arns(),
         }
     }
 }
@@ -1011,62 +994,6 @@ impl Default for ResyncDecision {
 mod tests {
     use super::*;
 
-    #[test]
-    fn replicate_decision_returns_sorted_unique_replicating_target_arns() {
-        let mut decision = ReplicateDecision::new();
-        decision.set(ReplicateTargetDecision {
-            arn: "arn:target-b".to_string(),
-            replicate: true,
-            ..Default::default()
-        });
-        decision.set(ReplicateTargetDecision {
-            arn: "arn:target-a".to_string(),
-            replicate: true,
-            ..Default::default()
-        });
-        decision.set(ReplicateTargetDecision {
-            arn: "arn:target-c".to_string(),
-            replicate: false,
-            ..Default::default()
-        });
-        decision.set(ReplicateTargetDecision {
-            arn: String::new(),
-            replicate: true,
-            ..Default::default()
-        });
-
-        assert_eq!(
-            decision.replicate_target_arns(),
-            vec!["arn:target-a".to_string(), "arn:target-b".to_string()]
-        );
-    }
-
-    #[test]
-    fn replicate_object_info_mrf_entry_carries_replicating_targets() {
-        let mut decision = ReplicateDecision::new();
-        decision.set(ReplicateTargetDecision {
-            arn: "arn:target-a".to_string(),
-            replicate: true,
-            ..Default::default()
-        });
-        decision.set(ReplicateTargetDecision {
-            arn: "arn:target-b".to_string(),
-            replicate: false,
-            ..Default::default()
-        });
-        let info = ReplicateObjectInfo {
-            bucket: "bucket".to_string(),
-            name: "object".to_string(),
-            size: 42,
-            dsc: decision,
-            ..Default::default()
-        };
-
-        let entry = info.to_mrf_entry();
-
-        assert_eq!(entry.target_arns, vec!["arn:target-a".to_string()]);
-    }
-
     #[test]
     fn target_state_reads_resync_timestamp_from_target_reset_header_key() {
         let arn = "arn:rustfs:replication:us-east-1:target:bucket";

crates/replication/src/mrf.rs

@@ -71,7 +71,6 @@ mod tests {
                 delete_marker_version_id: None,
                 delete_marker: false,
                 delete_marker_mtime: None,
-                target_arns: vec!["arn:target-a".to_string(), "arn:target-b".to_string()],
             },
             MrfReplicateEntry {
                 bucket: "bucket-a".to_string(),
@@ -83,7 +82,6 @@ mod tests {
                 delete_marker_version_id: Some(del_vid),
                 delete_marker: true,
                 delete_marker_mtime: Some(1_705_312_200_123_456_789),
-                target_arns: vec!["arn:target-a".to_string()],
             },
         ];
 
@@ -93,11 +91,9 @@ mod tests {
         assert_eq!(decoded.len(), 2);
         assert_eq!(decoded[0].version_id, Some(obj_vid));
         assert_eq!(decoded[0].op, MrfOpKind::Object);
-        assert_eq!(decoded[0].target_arns, vec!["arn:target-a".to_string(), "arn:target-b".to_string()]);
         assert_eq!(decoded[0].delete_marker_mtime, None);
         assert_eq!(decoded[1].delete_marker_version_id, Some(del_vid));
         assert_eq!(decoded[1].op, MrfOpKind::Delete);
-        assert_eq!(decoded[1].target_arns, vec!["arn:target-a".to_string()]);
         assert!(decoded[1].delete_marker);
         assert_eq!(
             decoded[1].delete_marker_mtime,
@@ -133,7 +129,6 @@ mod tests {
         assert_eq!(decoded[0].retry_count, 2);
         assert_eq!(decoded[0].size, 100);
         assert_eq!(decoded[0].op, MrfOpKind::Object);
-        assert!(decoded[0].target_arns.is_empty());
         // Old files lack the deleteMarkerMtime key; it must default to None so replay keeps the
         // pre-#867 fallback to the current time.
         assert_eq!(decoded[0].delete_marker_mtime, None);

rustfs/src/admin/handlers/replication.rs

@@ -835,10 +835,6 @@ struct MrfTargetBacklog {
     failed_count: i64,
     #[serde(rename = "FailedSize")]
     failed_size: i64,
-    #[serde(rename = "DurableCount")]
-    durable_count: i64,
-    #[serde(rename = "DurableSize")]
-    durable_size: i64,
     #[serde(rename = "ObservationScope")]
     observation_scope: &'static str,
 }
@@ -846,7 +842,7 @@ struct MrfTargetBacklog {
 /// Response body for `GET /v3/replication/mrf`.
 ///
 /// Runtime failed/queued totals and the durable MRF recovery backlog are kept
-/// separate because older persisted MRF entries do not contain a target ARN.
+/// separate because persisted MRF entries do not contain a target ARN.
 #[derive(Debug, Serialize)]
 struct MrfResponse {
     #[serde(rename = "Bucket")]
@@ -892,60 +888,32 @@ fn build_mrf_response(
         "partial_cluster"
     };
     let mut targets: Vec<MrfTargetBacklog> = Vec::with_capacity(bucket_stats.replication_stats.stats.len());
-    let mut targets_by_arn: HashMap<String, usize> = HashMap::with_capacity(bucket_stats.replication_stats.stats.len());
     let mut total_failed_count: i64 = 0;
     let mut total_failed_size: i64 = 0;
     for (arn, stat) in &bucket_stats.replication_stats.stats {
         total_failed_count = total_failed_count.saturating_add(stat.failed.count);
         total_failed_size = total_failed_size.saturating_add(stat.failed.size);
-        targets_by_arn.insert(arn.clone(), targets.len());
         targets.push(MrfTargetBacklog {
             arn: arn.clone(),
             failed_count: stat.failed.count,
             failed_size: stat.failed.size,
-            durable_count: 0,
-            durable_size: 0,
             observation_scope,
         });
     }
+    targets.sort_by(|a, b| a.arn.cmp(&b.arn));
 
     let queued = &bucket_stats.replication_stats.q_stat.curr;
-    let (durable_count, durable_size, missing_target_arns) = if durable.available {
-        durable.entries.iter().filter(|entry| entry.bucket == bucket).fold(
-            (0i64, 0i64, false),
-            |(count, size, missing_target_arns), entry| {
-                let mut missing_target_arns = missing_target_arns;
-                for target_arn in &entry.target_arns {
-                    if target_arn.is_empty() {
-                        continue;
-                    }
-                    let index = if let Some(index) = targets_by_arn.get(target_arn).copied() {
-                        index
-                    } else {
-                        targets_by_arn.insert(target_arn.clone(), targets.len());
-                        targets.push(MrfTargetBacklog {
-                            arn: target_arn.clone(),
-                            failed_count: 0,
-                            failed_size: 0,
-                            durable_count: 0,
-                            durable_size: 0,
-                            observation_scope,
-                        });
-                        targets.len() - 1
-                    };
-                    targets[index].durable_count = targets[index].durable_count.saturating_add(1);
-                    targets[index].durable_size = targets[index].durable_size.saturating_add(entry.size);
-                }
-                if entry.target_arns.is_empty() {
-                    missing_target_arns = true;
-                }
-                (count.saturating_add(1), size.saturating_add(entry.size), missing_target_arns)
-            },
-        )
+    let (durable_count, durable_size) = if durable.available {
+        durable
+            .entries
+            .iter()
+            .filter(|entry| entry.bucket == bucket)
+            .fold((0i64, 0i64), |(count, size), entry| {
+                (count.saturating_add(1), size.saturating_add(entry.size))
+            })
     } else {
-        (0, 0, false)
+        (0, 0)
     };
-    targets.sort_by(|a, b| a.arn.cmp(&b.arn));
 
     MrfResponse {
         bucket,
@@ -962,7 +930,7 @@ fn build_mrf_response(
         durable_backlog_available: durable.available,
         durable_count,
         durable_size,
-        per_target_durable_entries_available: durable.available && !missing_target_arns,
+        per_target_durable_entries_available: false,
     }
 }
 
@@ -972,9 +940,8 @@ fn build_mrf_response(
 ///
 /// Compatibility note: MinIO returns a stream of individual MRF entries. RustFS
 /// deliberately returns aggregate runtime and durable counters instead.
-/// `PerObjectEntriesAvailable` remains false until an enumerable API exists.
-/// `PerTargetDurableEntriesAvailable` is false when the durable backlog includes
-/// older entries that cannot be attributed to a target.
+/// `PerObjectEntriesAvailable` and `PerTargetDurableEntriesAvailable` remain
+/// false until an enumerable API and target-bearing durable format exist.
 pub struct ReplicationMrfHandler {}
 
 #[async_trait::async_trait]
@@ -1092,7 +1059,6 @@ mod tests {
                     delete_marker_version_id: None,
                     delete_marker: false,
                     delete_marker_mtime: None,
-                    target_arns: vec!["arn-a".to_string(), "arn-durable-only".to_string()],
                 },
                 MrfReplicateEntry {
                     bucket: "other-bucket".to_string(),
@@ -1104,7 +1070,6 @@ mod tests {
                     delete_marker_version_id: None,
                     delete_marker: false,
                     delete_marker_mtime: None,
-                    target_arns: Vec::new(),
                 },
             ],
         };
@@ -1122,64 +1087,7 @@ mod tests {
         assert_eq!(json["ClusterComplete"], false);
         assert_eq!(json["Targets"][0]["ObservationScope"], "partial_cluster");
         assert_eq!(json["PerObjectEntriesAvailable"], false);
-        assert_eq!(json["PerTargetDurableEntriesAvailable"], true);
-
-        let targets = json["Targets"].as_array().expect("targets should serialize as an array");
-        let target_a = targets
-            .iter()
-            .find(|target| target["ARN"] == "arn-a")
-            .expect("runtime target should remain present");
-        assert_eq!(target_a["FailedCount"], 3);
-        assert_eq!(target_a["FailedSize"], 900);
-        assert_eq!(target_a["DurableCount"], 1);
-        assert_eq!(target_a["DurableSize"], 250);
-        let target_durable_only = targets
-            .iter()
-            .find(|target| target["ARN"] == "arn-durable-only")
-            .expect("durable-only target should be surfaced");
-        assert_eq!(target_durable_only["FailedCount"], 0);
-        assert_eq!(target_durable_only["FailedSize"], 0);
-        assert_eq!(target_durable_only["DurableCount"], 1);
-        assert_eq!(target_durable_only["DurableSize"], 250);
-    }
-
-    #[test]
-    fn mrf_response_keeps_legacy_durable_entries_bucket_only() {
-        let mut stats = BucketStats::default();
-        stats.replication_stats.provider_available = true;
-        stats.replication_stats.cluster_complete = true;
-        stats.replication_stats.observed_node_count = 1;
-        stats.replication_stats.expected_node_count = 1;
-        let durable = DurableMrfBacklog {
-            available: true,
-            entries: vec![MrfReplicateEntry {
-                bucket: "bucket-a".to_string(),
-                object: "legacy-object".to_string(),
-                version_id: None,
-                retry_count: 0,
-                size: 250,
-                op: MrfOpKind::Object,
-                delete_marker_version_id: None,
-                delete_marker: false,
-                delete_marker_mtime: None,
-                target_arns: Vec::new(),
-            }],
-        };
-
-        let response = build_mrf_response("bucket-a".to_string(), &stats, durable);
-        let json = serde_json::to_value(response).expect("MRF response should serialize");
-
-        assert_eq!(json["DurableBacklogAvailable"], true);
-        assert_eq!(json["DurableCount"], 1);
-        assert_eq!(json["DurableSize"], 250);
         assert_eq!(json["PerTargetDurableEntriesAvailable"], false);
-        assert_eq!(
-            json["Targets"]
-                .as_array()
-                .expect("targets should serialize as an array")
-                .len(),
-            0
-        );
     }
 
     #[test]
@@ -1207,7 +1115,6 @@ mod tests {
         assert_eq!(valid_empty_json["DurableBacklogAvailable"], true);
         assert_eq!(valid_empty_json["TotalFailedCount"], 0);
         assert_eq!(valid_empty_json["DurableCount"], 0);
-        assert_eq!(valid_empty_json["PerTargetDurableEntriesAvailable"], true);
     }
 
     #[test]